NPE in HelixTaskExecutor.reset() when a task completes during shutdown

[Jackie-Jiang]

Describe the bug

HelixTaskExecutor.reset() ends with a loop that logs tasks failing to terminate:

for (String taskId : _taskMap.keySet()) {
  MessageTaskInfo info = _taskMap.get(taskId);
  sb.append("Task: " + taskId + " fails to terminate. Message: " + info._task.getMessage() + "\n");
}

_taskMap is a ConcurrentHashMap, and message tasks remove themselves from it when they finish. A task that completes while reset() runs disappears between the keySet() iteration and the get(taskId), so get returns null and reading info._task throws:

java.lang.NullPointerException: Cannot read field "_task" because "info" is null

Note the loop only exists to log tasks that failed to terminate — the crash is caused by a task that terminated. The NPE propagates out of reset() and can fail the whole participant disconnect. We hit it intermittently in Apache Pinot integration tests during server shutdown (ClusterTest.stopServer -> participant disconnect -> HelixTaskExecutor.reset()).

To Reproduce

Race window: disconnect a participant while a message task is completing. Deterministic reproduction is hard; any shutdown with in-flight tasks can hit it.

Expected behavior

reset() completes and logs whatever tasks are genuinely still present; a concurrently-completing task must not fail the shutdown.

Additional context

Observed on 2.0.1; the same unguarded code is present on current master. Fix: iterate entrySet() so each entry is read atomically (a concurrent removal then either shows the entry or skips it, but never yields a null value).