refactor: remove deprecated curator ServiceAnnouncer stuff (#19523)

Author: clintropolis

docs/configuration/index.md

@@ -156,12 +156,6 @@ Druid interacts with ZooKeeper through a set of standard path configurations. We
 |`druid.zk.paths.base`|Base ZooKeeper path.|`/druid`|
 |`druid.zk.paths.coordinatorPath`|Used by the Coordinator for leader election.|`${druid.zk.paths.base}/coordinator`|
 
-The following path is used for service discovery. It is **not** affected by `druid.zk.paths.base` and **must** be specified separately.
-
-|Property|Description|Default|
-|--------|-----------|-------|
-|`druid.discovery.curator.path`|Services announce themselves under this ZooKeeper path.|`/druid/discovery`|
-
 ### TLS
 
 #### General configuration

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskSerdeTest.java

@@ -35,7 +35,6 @@
 import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
 import org.apache.druid.segment.indexing.DataSchema;
 import org.apache.druid.segment.realtime.ChatHandlerProvider;
-import org.apache.druid.segment.realtime.NoopChatHandlerProvider;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.server.security.Action;
 import org.apache.druid.server.security.Resource;
@@ -160,7 +159,7 @@ private static ObjectMapper createObjectMapper()
               binder.bindConstant().annotatedWith(Names.named("serviceName")).to("test");
               binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8000);
               binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(9000);
-              binder.bind(ChatHandlerProvider.class).toInstance(new NoopChatHandlerProvider());
+              binder.bind(ChatHandlerProvider.class).toInstance(new ChatHandlerProvider());
               binder.bind(RowIngestionMetersFactory.class).toInstance(new DropwizardRowIngestionMetersFactory());
               binder.bind(AppenderatorsManager.class).toInstance(new TestAppenderatorsManager());
             }

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

@@ -435,7 +435,7 @@ public TaskStatus runTask(final TaskToolbox toolbox)
         // ParallelIndexSupervisorTask because it doesn't support APIs for live ingestion reports.
         log.warn("Chat handler is already registered. Skipping chat handler registration.");
       } else {
-        toolbox.getChatHandlerProvider().register(getId(), this, false);
+        toolbox.getChatHandlerProvider().register(getId(), this);
       }
 
       this.authorizerMapper = toolbox.getAuthorizerMapper();

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java

@@ -522,7 +522,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
         Preconditions.checkNotNull(toolbox.getChatHandlerProvider(), "chatHandlerProvider").getClass().getName()
     );
     authorizerMapper = toolbox.getAuthorizerMapper();
-    toolbox.getChatHandlerProvider().register(getId(), this, false);
+    toolbox.getChatHandlerProvider().register(getId(), this);
 
     // the lineage-based segment allocation protocol must be used as the legacy protocol has a critical bug
     // (see SinglePhaseParallelIndexTaskRunner.allocateNewSegment()). However, we tell subtasks to use

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java

@@ -242,7 +242,7 @@ public TaskStatus runTask(final TaskToolbox toolbox) throws Exception
       }
       this.authorizerMapper = toolbox.getAuthorizerMapper();
 
-      toolbox.getChatHandlerProvider().register(getId(), this, false);
+      toolbox.getChatHandlerProvider().register(getId(), this);
 
       rowIngestionMeters = toolbox.getRowIngestionMetersFactory().createRowIngestionMeters();
       parseExceptionHandler = new ParseExceptionHandler(

indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java

@@ -23,9 +23,7 @@
 import com.google.common.base.Optional;
 import com.google.inject.Inject;
 import org.apache.druid.client.indexing.IndexingService;
-import org.apache.druid.curator.discovery.ServiceAnnouncer;
 import org.apache.druid.discovery.DruidLeaderSelector;
-import org.apache.druid.guice.annotations.Self;
 import org.apache.druid.indexing.common.actions.SegmentAllocationQueue;
 import org.apache.druid.indexing.common.actions.TaskActionClientFactory;
 import org.apache.druid.indexing.common.task.TaskContextEnricher;
@@ -42,7 +40,6 @@
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.metadata.segment.cache.SegmentMetadataCache;
-import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.coordinator.CoordinatorOverlordServiceConfig;
 
 import java.util.concurrent.atomic.AtomicReference;
@@ -67,10 +64,9 @@ public class DruidOverlord
   private final AtomicReference<Lifecycle> leaderLifecycleRef = new AtomicReference<>(null);
 
   /**
-   * Indicates that all services have been started and the node can now announce
-   * itself with {@link ServiceAnnouncer#announce}. This must be set to false
-   * as soon as {@link DruidLeaderSelector.Listener#stopBeingLeader()} is
-   * called.
+   * Indicates that all services have been started and the node is ready to serve
+   * leader-only HTTP routes. This must be set to false as soon as
+   * {@link DruidLeaderSelector.Listener#stopBeingLeader()} is called.
    */
   private volatile boolean initialized;
 
@@ -83,9 +79,7 @@ public DruidOverlord(
       final GlobalTaskLockbox taskLockbox,
       final TaskStorage taskStorage,
       final TaskActionClientFactory taskActionClientFactory,
-      @Self final DruidNode selfNode,
       final TaskRunnerFactory runnerFactory,
-      final ServiceAnnouncer serviceAnnouncer,
       final CoordinatorOverlordServiceConfig coordinatorOverlordServiceConfig,
       final ServiceEmitter emitter,
       final SupervisorManager supervisorManager,
@@ -103,9 +97,6 @@ public DruidOverlord(
     this.segmentMetadataCache = segmentMetadataCache;
     this.coordinatorOverlordServiceConfig = coordinatorOverlordServiceConfig;
 
-    final DruidNode node = coordinatorOverlordServiceConfig.getOverlordService() == null ? selfNode :
-                           selfNode.withService(coordinatorOverlordServiceConfig.getOverlordService());
-
     this.leadershipListener = new DruidLeaderSelector.Listener()
     {
       @Override
@@ -173,15 +164,13 @@ public void start()
                   compactionScheduler.becomeLeader();
                   scheduledBatchTaskManager.start();
 
-                  // Announce the node only after all the services have been initialized
+                  // Mark ready only after all the services have been initialized
                   initialized = true;
-                  serviceAnnouncer.announce(node);
                 }
 
                 @Override
                 public void stop()
                 {
-                  serviceAnnouncer.unannounce(node);
                   scheduledBatchTaskManager.stop();
                   compactionScheduler.stopBeingLeader();
                   taskMaster.downgradeToHalfLeader();

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

@@ -423,7 +423,7 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
     initializeSequences();
 
     log.debug("Found chat handler of class[%s]", toolbox.getChatHandlerProvider().getClass().getName());
-    toolbox.getChatHandlerProvider().register(task.getId(), this, false);
+    toolbox.getChatHandlerProvider().register(task.getId(), this);
 
     runThread = Thread.currentThread();
 

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

@@ -53,7 +53,7 @@
 import org.apache.druid.segment.loading.SegmentLoaderConfig;
 import org.apache.druid.segment.loading.SegmentLocalCacheManager;
 import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
-import org.apache.druid.segment.realtime.NoopChatHandlerProvider;
+import org.apache.druid.segment.realtime.ChatHandlerProvider;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.appenderator.UnifiedIndexerAppenderatorsManager;
 import org.apache.druid.server.DruidNode;
@@ -150,7 +150,7 @@ public void setUp() throws IOException
         new NoopTestTaskReportFileWriter(),
         null,
         AuthTestUtils.TEST_AUTHORIZER_MAPPER,
-        new NoopChatHandlerProvider(),
+        new ChatHandlerProvider(),
         new DropwizardRowIngestionMetersFactory(),
         new TestAppenderatorsManager(),
         new NoopOverlordClient(),

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

@@ -47,7 +47,6 @@
 import org.apache.druid.segment.loading.LocalDataSegmentPuller;
 import org.apache.druid.segment.loading.LocalLoadSpec;
 import org.apache.druid.segment.realtime.ChatHandlerProvider;
-import org.apache.druid.segment.realtime.NoopChatHandlerProvider;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
 import org.apache.druid.server.security.AuthConfig;
@@ -94,7 +93,7 @@ public TestUtils()
             .addValue(ExprMacroTable.class, LookupEnabledTestExprMacroTable.INSTANCE)
             .addValue(IndexIO.class, indexIO)
             .addValue(ObjectMapper.class, jsonMapper)
-            .addValue(ChatHandlerProvider.class, new NoopChatHandlerProvider())
+            .addValue(ChatHandlerProvider.class, new ChatHandlerProvider())
             .addValue(AuthConfig.class, new AuthConfig())
             .addValue(AuthorizerMapper.class, null)
             .addValue(RowIngestionMetersFactory.class, rowIngestionMetersFactory)

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

@@ -60,7 +60,6 @@
 import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
 import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
 import org.apache.druid.segment.realtime.ChatHandlerProvider;
-import org.apache.druid.segment.realtime.NoopChatHandlerProvider;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.transform.CompactionTransformSpec;
 import org.apache.druid.segment.writeout.TmpFileSegmentWriteOutMediumFactory;
@@ -174,7 +173,7 @@ private static ObjectMapper setupInjectablesInObjectMapper(ObjectMapper objectMa
             ImmutableList.of(
                 binder -> {
                   binder.bind(AuthorizerMapper.class).toInstance(AuthTestUtils.TEST_AUTHORIZER_MAPPER);
-                  binder.bind(ChatHandlerProvider.class).toInstance(new NoopChatHandlerProvider());
+                  binder.bind(ChatHandlerProvider.class).toInstance(new ChatHandlerProvider());
                   binder.bind(RowIngestionMetersFactory.class).toInstance(ROW_INGESTION_METERS_FACTORY);
                   binder.bind(CoordinatorClient.class).toInstance(COORDINATOR_CLIENT);
                   binder.bind(SegmentCacheManagerFactory.class).toInstance(new SegmentCacheManagerFactory(TestIndex.INDEX_IO, objectMapper));