Improve concurrency in TaskQueue

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -98,6 +98,7 @@
 import org.apache.druid.segment.incremental.ParseExceptionReport;
 import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.utils.CollectionUtils;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 
@@ -4309,24 +4310,22 @@ public void setPartitionIdsForTests(
   }
 
   /**
-   * Get all active tasks from metadata storage
+   * Get all active tasks either from memory or metadata store, if not cached.
    *
    * @return map from taskId to Task
    */
   private Map<String, Task> getActiveTaskMap()
   {
-    final ImmutableMap.Builder<String, Task> activeTaskMap = ImmutableMap.builder();
     final Optional<TaskQueue> taskQueue = taskMaster.getTaskQueue();
-    final List<Task> tasks;
     if (taskQueue.isPresent()) {
-      tasks = taskQueue.get().getActiveTasksForDatasource(dataSource);
+      return taskQueue.get().getActiveTasksForDatasource(dataSource);
     } else {
-      tasks = taskStorage.getActiveTasksByDatasource(dataSource);
-    }
-    for (Task task : tasks) {
-      activeTaskMap.put(task.getId(), task);
+      return CollectionUtils.toMap(
+          taskStorage.getActiveTasksByDatasource(dataSource),
+          Task::getId,
+          task -> task
+      );
     }
-    return activeTaskMap.build();
   }
 
   /**