Improve logging to include taskId in handoff notifier thread

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java

@@ -719,7 +719,7 @@ protected boolean waitForSegmentAvailability(
 
     try (
         SegmentHandoffNotifier notifier = toolbox.getSegmentHandoffNotifierFactory()
-                                                 .createSegmentHandoffNotifier(segmentsToWaitFor.get(0).getDataSource())
+                                                 .createSegmentHandoffNotifier(segmentsToWaitFor.get(0).getDataSource(), getId())
     ) {
 
       final ExecutorService exec = Execs.directExecutor();

indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java

@@ -1181,7 +1181,7 @@ public void testWaitForSegmentAvailabilityMultipleSegmentsTimeout()
     EasyMock.expect(mockToolbox.getSegmentHandoffNotifierFactory()).andReturn(mockFactory).once();
     EasyMock.expect(mockToolbox.getEmitter()).andReturn(new NoopServiceEmitter()).anyTimes();
     EasyMock.expect(mockDataSegment1.getDataSource()).andReturn("MockDataSource").once();
-    EasyMock.expect(mockFactory.createSegmentHandoffNotifier("MockDataSource")).andReturn(mockNotifier).once();
+    EasyMock.expect(mockFactory.createSegmentHandoffNotifier("MockDataSource", indexTask.getId())).andReturn(mockNotifier).once();
     mockNotifier.start();
     EasyMock.expectLastCall().once();
     mockNotifier.registerSegmentHandoffCallback(EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject());

indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java

@@ -446,7 +446,7 @@ private SegmentHandoffNotifierFactory setUpSegmentHandOffNotifierFactory()
     return new SegmentHandoffNotifierFactory()
     {
       @Override
-      public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource)
+      public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource, String taskId)
       {
         return new SegmentHandoffNotifier()
         {

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java

@@ -629,7 +629,7 @@ public boolean checkPointDataSourceMetadata(
     final TaskActionClientFactory taskActionClientFactory = new LocalTaskActionClientFactory(
         taskActionToolbox
     );
-    final SegmentHandoffNotifierFactory handoffNotifierFactory = dataSource -> new SegmentHandoffNotifier()
+    final SegmentHandoffNotifierFactory handoffNotifierFactory = (dataSource, taskId) -> new SegmentHandoffNotifier()
     {
       @Override
       public boolean registerSegmentHandoffCallback(

server/src/main/java/org/apache/druid/segment/handoff/CoordinatorBasedSegmentHandoffNotifier.java

@@ -44,22 +44,25 @@ public class CoordinatorBasedSegmentHandoffNotifier implements SegmentHandoffNot
   private volatile ScheduledExecutorService scheduledExecutor;
   private final Duration pollDuration;
   private final String dataSource;
+  private final String taskId;
 
   public CoordinatorBasedSegmentHandoffNotifier(
       String dataSource,
       CoordinatorClient coordinatorClient,
-      CoordinatorBasedSegmentHandoffNotifierConfig config
+      CoordinatorBasedSegmentHandoffNotifierConfig config,
+      String taskId
   )
   {
     this.dataSource = dataSource;
     this.coordinatorClient = coordinatorClient;
     this.pollDuration = config.getPollDuration();
+    this.taskId = taskId;
   }
 
   @Override
   public boolean registerSegmentHandoffCallback(SegmentDescriptor descriptor, Executor exec, Runnable handOffRunnable)
   {
-    log.debug("Adding SegmentHandoffCallback for dataSource[%s] Segment[%s]", dataSource, descriptor);
+    log.debug("Adding SegmentHandoffCallback for dataSource[%s] Segment[%s] for task[%s]", dataSource, descriptor, taskId);
     Pair<Executor, Runnable> prev = handOffCallbacks.putIfAbsent(
         descriptor,
         new Pair<>(exec, handOffRunnable)
@@ -91,30 +94,32 @@ void checkForSegmentHandoffs()
           Boolean handOffComplete =
               FutureUtils.getUnchecked(coordinatorClient.isHandoffComplete(dataSource, descriptor), true);
           if (Boolean.TRUE.equals(handOffComplete)) {
-            log.debug("Segment handoff complete for dataSource[%s] segment[%s]", dataSource, descriptor);
+            log.debug("Segment handoff complete for dataSource[%s] segment[%s] for task[%s]", dataSource, descriptor, taskId);
             entry.getValue().lhs.execute(entry.getValue().rhs);
             itr.remove();
           }
         }
         catch (Exception e) {
           log.error(
               e,
-              "Exception while checking handoff for dataSource[%s] Segment[%s]; will try again after [%s]",
+              "Exception while checking handoff for dataSource[%s] Segment[%s], taskId[%s]; will try again after [%s]",
               dataSource,
               descriptor,
+              taskId,
               pollDuration
           );
         }
       }
       if (!handOffCallbacks.isEmpty()) {
-        log.info("Still waiting for handoff for [%d] segments", handOffCallbacks.size());
+        log.info("Still waiting for handoff for [%d] segments for task[%s]", handOffCallbacks.size(), taskId);
       }
     }
     catch (Throwable t) {
       log.error(
           t,
-          "Exception while checking handoff for dataSource[%s]; will try again after [%s]",
+          "Exception while checking handoff for dataSource[%s], taskId[%s]; will try again after [%s]",
           dataSource,
+          taskId,
           pollDuration
       );
     }

server/src/main/java/org/apache/druid/segment/handoff/CoordinatorBasedSegmentHandoffNotifierFactory.java

@@ -39,12 +39,13 @@ public CoordinatorBasedSegmentHandoffNotifierFactory(
   }
 
   @Override
-  public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource)
+  public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource, String taskId)
   {
     return new CoordinatorBasedSegmentHandoffNotifier(
         dataSource,
         client,
-        config
+        config,
+        taskId
     );
   }
 }

server/src/main/java/org/apache/druid/segment/handoff/NoopSegmentHandoffNotifierFactory.java

@@ -50,7 +50,7 @@ public void close()
   };
 
   @Override
-  public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource)
+  public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource, String taskId)
   {
     return NOTIFIER;
   }

server/src/main/java/org/apache/druid/segment/handoff/SegmentHandoffNotifierFactory.java

@@ -22,5 +22,5 @@
 
 public interface SegmentHandoffNotifierFactory
 {
-  SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource);
+  SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource, String taskId);
 }

server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorDriver.java

@@ -108,7 +108,7 @@ public StreamAppenderatorDriver(
     super(appenderator, segmentAllocator, segmentRetriever, dataSegmentKiller);
 
     this.handoffNotifier = Preconditions.checkNotNull(handoffNotifierFactory, "handoffNotifierFactory")
-                                        .createSegmentHandoffNotifier(appenderator.getDataSource());
+                                        .createSegmentHandoffNotifier(appenderator.getDataSource(), appenderator.getId());
     this.metrics = Preconditions.checkNotNull(metrics, "metrics");
     this.objectMapper = Preconditions.checkNotNull(objectMapper, "objectMapper");
   }
@@ -360,7 +360,7 @@ public ListenableFuture<SegmentsAndCommitMetadata> registerHandoff(SegmentsAndCo
             ),
             Execs.directExecutor(),
             () -> {
-              log.debug("Segment[%s] successfully handed off, dropping.", segmentIdentifier);
+              log.debug("Segment[%s] successfully handed off for task[%s], dropping.", segmentIdentifier, appenderator.getId());
               metrics.incrementHandOffCount();
 
               final ListenableFuture<?> dropFuture = appenderator.drop(segmentIdentifier);

server/src/test/java/org/apache/druid/segment/handoff/CoordinatorBasedSegmentHandoffNotifierTest.java

@@ -58,7 +58,8 @@ public void testHandoffCallbackNotCalled()
     CoordinatorBasedSegmentHandoffNotifier notifier = new CoordinatorBasedSegmentHandoffNotifier(
         "test_ds",
         coordinatorClient,
-        notifierConfig
+        notifierConfig,
+        "test_task"
     );
     final AtomicBoolean callbackCalled = new AtomicBoolean(false);
     notifier.registerSegmentHandoffCallback(
@@ -89,7 +90,8 @@ public void testHandoffCallbackCalled()
     CoordinatorBasedSegmentHandoffNotifier notifier = new CoordinatorBasedSegmentHandoffNotifier(
         "test_ds",
         coordinatorClient,
-        notifierConfig
+        notifierConfig,
+        "test_task"
     );
 
     notifier.registerSegmentHandoffCallback(

server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorDriverTest.java

@@ -543,7 +543,7 @@ public Set<SegmentDescriptor> getHandedOffSegmentDescriptors()
     }
 
     @Override
-    public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource)
+    public SegmentHandoffNotifier createSegmentHandoffNotifier(String dataSource, String taskId)
     {
       return new SegmentHandoffNotifier()
       {