Merge core CoordinatorClient with MSQ CoordinatorServiceClient.

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java

@@ -1106,7 +1106,7 @@ private DataSegmentTimelineView makeDataSegmentTimelineView()
   {
     return (dataSource, intervals) -> {
       final Collection<DataSegment> dataSegments =
-          context.coordinatorClient().fetchUsedSegmentsInDataSourceForIntervals(dataSource, intervals);
+          FutureUtils.getUnchecked(context.coordinatorClient().fetchUsedSegments(dataSource, intervals), true);
 
       if (dataSegments.isEmpty()) {
         return Optional.empty();

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/TaskDataSegmentProvider.java

@@ -20,6 +20,7 @@
 package org.apache.druid.msq.exec;
 
 import com.google.errorprone.annotations.concurrent.GuardedBy;
+import org.apache.druid.client.coordinator.CoordinatorClient;
 import org.apache.druid.collections.ReferenceCountingResourceHolder;
 import org.apache.druid.collections.ResourceHolder;
 import org.apache.druid.common.guava.FutureUtils;
@@ -28,7 +29,6 @@
 import org.apache.druid.java.util.common.io.Closer;
 import org.apache.druid.msq.counters.ChannelCounters;
 import org.apache.druid.msq.querykit.DataSegmentProvider;
-import org.apache.druid.msq.rpc.CoordinatorServiceClient;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.QueryableIndex;
 import org.apache.druid.segment.QueryableIndexSegment;
@@ -52,13 +52,13 @@
  */
 public class TaskDataSegmentProvider implements DataSegmentProvider
 {
-  private final CoordinatorServiceClient coordinatorClient;
+  private final CoordinatorClient coordinatorClient;
   private final SegmentCacheManager segmentCacheManager;
   private final IndexIO indexIO;
   private final ConcurrentHashMap<SegmentId, SegmentHolder> holders;
 
   public TaskDataSegmentProvider(
-      CoordinatorServiceClient coordinatorClient,
+      CoordinatorClient coordinatorClient,
       SegmentCacheManager segmentCacheManager,
       IndexIO indexIO
   )

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java

@@ -43,7 +43,6 @@
 import org.apache.druid.msq.indexing.client.WorkerChatHandler;
 import org.apache.druid.msq.kernel.FrameContext;
 import org.apache.druid.msq.kernel.QueryDefinition;
-import org.apache.druid.msq.rpc.CoordinatorServiceClient;
 import org.apache.druid.rpc.ServiceClientFactory;
 import org.apache.druid.rpc.ServiceLocations;
 import org.apache.druid.rpc.ServiceLocator;
@@ -95,8 +94,6 @@ public IndexerWorkerContext(
   public static IndexerWorkerContext createProductionInstance(final TaskToolbox toolbox, final Injector injector)
   {
     final IndexIO indexIO = injector.getInstance(IndexIO.class);
-    final CoordinatorServiceClient coordinatorServiceClient =
-        injector.getInstance(CoordinatorServiceClient.class).withRetryPolicy(StandardRetryPolicy.unlimited());
     final SegmentCacheManager segmentCacheManager =
         injector.getInstance(SegmentCacheManagerFactory.class)
                 .manufacturate(new File(toolbox.getIndexingTmpDir(), "segment-fetch"));
@@ -107,7 +104,7 @@ public static IndexerWorkerContext createProductionInstance(final TaskToolbox to
         toolbox,
         injector,
         indexIO,
-        new TaskDataSegmentProvider(coordinatorServiceClient, segmentCacheManager, indexIO),
+        new TaskDataSegmentProvider(toolbox.getCoordinatorClient(), segmentCacheManager, indexIO),
         serviceClientFactory
     );
   }

extensions-core/multi-stage-query/src/main/resources/META-INF/services/org.apache.druid.initialization.DruidModule

@@ -16,6 +16,5 @@
 org.apache.druid.msq.guice.MSQExternalDataSourceModule
 org.apache.druid.msq.guice.MSQIndexingModule
 org.apache.druid.msq.guice.MSQDurableStorageModule
-org.apache.druid.msq.guice.MSQServiceClientModule
 org.apache.druid.msq.guice.MSQSqlModule
 org.apache.druid.msq.guice.SqlTaskModule

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/TaskDataSegmentProviderTest.java

@@ -29,6 +29,7 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.client.coordinator.NoopCoordinatorClient;
 import org.apache.druid.collections.ResourceHolder;
 import org.apache.druid.collections.bitmap.BitmapFactory;
 import org.apache.druid.collections.bitmap.RoaringBitmapFactory;
@@ -41,8 +42,6 @@
 import org.apache.druid.java.util.common.jackson.JacksonUtils;
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.msq.counters.ChannelCounters;
-import org.apache.druid.msq.rpc.CoordinatorServiceClient;
-import org.apache.druid.rpc.ServiceRetryPolicy;
 import org.apache.druid.segment.DimensionHandler;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.Metadata;
@@ -149,7 +148,7 @@ public void setUp() throws Exception
     );
 
     provider = new TaskDataSegmentProvider(
-        new TestCoordinatorServiceClientImpl(),
+        new TestCoordinatorClientImpl(),
         cacheManager,
         indexIO
     );
@@ -229,7 +228,7 @@ public void testConcurrency()
     Assert.assertArrayEquals(new String[]{}, cacheDir.list());
   }
 
-  private class TestCoordinatorServiceClientImpl implements CoordinatorServiceClient
+  private class TestCoordinatorClientImpl extends NoopCoordinatorClient
   {
     @Override
     public ListenableFuture<DataSegment> fetchUsedSegment(String dataSource, String segmentId)
@@ -242,12 +241,6 @@ public ListenableFuture<DataSegment> fetchUsedSegment(String dataSource, String
 
       return Futures.immediateFailedFuture(new ISE("No such segment[%s] for dataSource[%s]", segmentId, dataSource));
     }
-
-    @Override
-    public CoordinatorServiceClient withRetryPolicy(ServiceRetryPolicy retryPolicy)
-    {
-      return this;
-    }
   }
 
   @JsonTypeName("test")

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestControllerContext.java

@@ -98,16 +98,18 @@ public MSQTestControllerContext(
     this.injector = injector;
     this.taskActionClient = taskActionClient;
     coordinatorClient = Mockito.mock(CoordinatorClient.class);
-    Mockito.when(coordinatorClient.fetchUsedSegmentsInDataSourceForIntervals(
+    Mockito.when(coordinatorClient.fetchUsedSegments(
                      ArgumentMatchers.anyString(),
                      ArgumentMatchers.anyList()
                  )
     ).thenAnswer(invocation ->
-                     (injector.getInstance(SpecificSegmentsQuerySegmentWalker.class)
-                              .getSegments()
-                              .stream()
-                              .filter(dataSegment -> dataSegment.getDataSource().equals(invocation.getArguments()[0]))
-                              .collect(Collectors.toList())
+                     Futures.immediateFuture(
+                         injector.getInstance(SpecificSegmentsQuerySegmentWalker.class)
+                                 .getSegments()
+                                 .stream()
+                                 .filter(dataSegment -> dataSegment.getDataSource()
+                                                                   .equals(invocation.getArguments()[0]))
+                                 .collect(Collectors.toList())
                      )
     );
     this.workerMemoryParameters = workerMemoryParameters;

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

@@ -45,6 +45,7 @@
 import org.apache.druid.java.util.metrics.MonitorScheduler;
 import org.apache.druid.query.QueryProcessingPool;
 import org.apache.druid.query.QueryRunnerFactoryConglomerate;
+import org.apache.druid.rpc.StandardRetryPolicy;
 import org.apache.druid.rpc.indexing.OverlordClient;
 import org.apache.druid.segment.IndexIO;
 import org.apache.druid.segment.IndexMergerV9Factory;
@@ -245,8 +246,11 @@ public TaskToolbox build(TaskConfig config, Task task)
         .chatHandlerProvider(chatHandlerProvider)
         .rowIngestionMetersFactory(rowIngestionMetersFactory)
         .appenderatorsManager(appenderatorsManager)
-        .overlordClient(overlordClient)
-        .coordinatorClient(coordinatorClient)
+        // Most tasks are written in such a way that if an Overlord or Coordinator RPC fails, the task fails.
+        // Set the retry policy to "about an hour", so tasks are resilient to brief Coordinator/Overlord problems.
+        // Calls will still eventually fail if problems persist.
+        .overlordClient(overlordClient.withRetryPolicy(StandardRetryPolicy.aboutAnHour()))
+        .coordinatorClient(coordinatorClient.withRetryPolicy(StandardRetryPolicy.aboutAnHour()))
         .supervisorTaskClientProvider(supervisorTaskClientProvider)
         .shuffleClient(shuffleClient)
         .taskLogPusher(taskLogPusher)