Fix Huge Number of Watches in ZooKeeper (#17482)

---------

Co-authored-by: asdf2014 <[email protected]>

Author: GWphua

docs/api-reference/tasks-api.md

@@ -1059,9 +1059,9 @@ Host: http://ROUTER_IP:ROUTER_PORT
     2023-07-03T22:11:17,933 INFO [task-runner-0-priority-0] org.apache.kafka.common.metrics.Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter
     2023-07-03T22:11:17,933 INFO [task-runner-0-priority-0] org.apache.kafka.common.metrics.Metrics - Metrics reporters closed
     2023-07-03T22:11:17,935 INFO [task-runner-0-priority-0] org.apache.kafka.common.utils.AppInfoParser - App info kafka.consumer for consumer-kafka-supervisor-dcanhmig-1 unregistered
-    2023-07-03T22:11:17,936 INFO [task-runner-0-priority-0] org.apache.druid.curator.announcement.Announcer - Unannouncing [/druid/internal-discovery/PEON/localhost:8100]
+    2023-07-03T22:11:17,936 INFO [task-runner-0-priority-0] org.apache.druid.curator.announcement.PathChildrenAnnouncer - Unannouncing [/druid/internal-discovery/PEON/localhost:8100]
     2023-07-03T22:11:17,972 INFO [task-runner-0-priority-0] org.apache.druid.curator.discovery.CuratorDruidNodeAnnouncer - Unannounced self [{"druidNode":{"service":"druid/middleManager","host":"localhost","bindOnHost":false,"plaintextPort":8100,"port":-1,"tlsPort":-1,"enablePlaintextPort":true,"enableTlsPort":false},"nodeType":"peon","services":{"dataNodeService":{"type":"dataNodeService","tier":"_default_tier","maxSize":0,"type":"indexer-executor","serverType":"indexer-executor","priority":0},"lookupNodeService":{"type":"lookupNodeService","lookupTier":"__default"}}}].
-    2023-07-03T22:11:17,972 INFO [task-runner-0-priority-0] org.apache.druid.curator.announcement.Announcer - Unannouncing [/druid/announcements/localhost:8100]
+    2023-07-03T22:11:17,972 INFO [task-runner-0-priority-0] org.apache.druid.curator.announcement.PathChildrenAnnouncer - Unannouncing [/druid/announcements/localhost:8100]
     2023-07-03T22:11:17,996 INFO [task-runner-0-priority-0] org.apache.druid.indexing.worker.executor.ExecutorLifecycle - Task completed with status: {
     "id" : "index_kafka_social_media_0e905aa31037879_nommnaeg",
     "status" : "SUCCESS",

docs/configuration/index.md

@@ -148,6 +148,7 @@ We recommend just setting the base ZK path and the ZK service host, but all ZK p
 |`druid.zk.service.connectionTimeoutMs`|ZooKeeper connection timeout, in milliseconds.|`15000`|
 |`druid.zk.service.compress`|Boolean flag for whether or not created Znodes should be compressed.|`true`|
 |`druid.zk.service.acl`|Boolean flag for whether or not to enable ACL security for ZooKeeper. If ACL is enabled, zNode creators will have all permissions.|`false`|
+|`druid.zk.service.pathChildrenCacheStrategy`|Dictates the underlying caching strategy for service announcements. Set true to let announcers to use Apache Curator's PathChildrenCache strategy, otherwise NodeCache strategy. Consider using NodeCache strategy when you are dealing with huge number of ZooKeeper watches in your cluster.|`true`|
 
 #### Path configuration
 

indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerCuratorCoordinator.java

@@ -25,11 +25,11 @@
 import com.google.inject.Inject;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.druid.curator.CuratorUtils;
-import org.apache.druid.curator.announcement.Announcer;
+import org.apache.druid.curator.announcement.ServiceAnnouncer;
+import org.apache.druid.guice.annotations.DirectExecutorAnnouncer;
 import org.apache.druid.indexing.overlord.config.RemoteTaskRunnerConfig;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
 import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
 import org.apache.druid.java.util.common.logger.Logger;
@@ -54,7 +54,7 @@ public class WorkerCuratorCoordinator
   private final ObjectMapper jsonMapper;
   private final RemoteTaskRunnerConfig config;
   private final CuratorFramework curatorFramework;
-  private final Announcer announcer;
+  private final ServiceAnnouncer announcer;
 
   private final String baseAnnouncementsPath;
   private final String baseTaskPath;
@@ -69,15 +69,15 @@ public WorkerCuratorCoordinator(
       IndexerZkConfig indexerZkConfig,
       RemoteTaskRunnerConfig config,
       CuratorFramework curatorFramework,
+      @DirectExecutorAnnouncer ServiceAnnouncer announcer,
       Worker worker
   )
   {
     this.jsonMapper = jsonMapper;
     this.config = config;
     this.curatorFramework = curatorFramework;
     this.worker = worker;
-
-    this.announcer = new Announcer(curatorFramework, Execs.directExecutor());
+    this.announcer = announcer;
 
     this.baseAnnouncementsPath = getPath(Arrays.asList(indexerZkConfig.getAnnouncementsPath(), worker.getHost()));
     this.baseTaskPath = getPath(Arrays.asList(indexerZkConfig.getTasksPath(), worker.getHost()));
@@ -87,7 +87,7 @@ public WorkerCuratorCoordinator(
   @LifecycleStart
   public void start() throws Exception
   {
-    log.info("WorkerCuratorCoordinator good to go sir. Server[%s]", worker.getHost());
+    log.info("WorkerCuratorCoordinator good to go. Server[%s]", worker.getHost());
     synchronized (lock) {
       if (started) {
         return;

indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskMonitorTest.java

@@ -28,6 +28,7 @@
 import org.apache.curator.test.TestingCluster;
 import org.apache.druid.client.coordinator.NoopCoordinatorClient;
 import org.apache.druid.curator.PotentiallyGzippedCompressionProvider;
+import org.apache.druid.curator.announcement.NodeAnnouncer;
 import org.apache.druid.indexer.TaskState;
 import org.apache.druid.indexing.common.IndexingServiceCondition;
 import org.apache.druid.indexing.common.SegmentCacheManagerFactory;
@@ -47,6 +48,7 @@
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.query.policy.NoopPolicyEnforcer;
 import org.apache.druid.rpc.indexing.NoopOverlordClient;
 import org.apache.druid.rpc.indexing.OverlordClient;
@@ -142,6 +144,7 @@ public String getBase()
         ),
         new TestRemoteTaskRunnerConfig(new Period("PT1S")),
         cf,
+        new NodeAnnouncer(cf, Execs.directExecutor()),
         worker
     );
     workerCuratorCoordinator.start();

indexing-service/src/test/java/org/apache/druid/indexing/worker/http/WorkerResourceTest.java

@@ -26,13 +26,15 @@
 import org.apache.curator.test.TestingCluster;
 import org.apache.druid.curator.PotentiallyGzippedCompressionProvider;
 import org.apache.druid.curator.ZkEnablementConfig;
+import org.apache.druid.curator.announcement.NodeAnnouncer;
 import org.apache.druid.indexing.overlord.config.RemoteTaskRunnerConfig;
 import org.apache.druid.indexing.worker.Worker;
 import org.apache.druid.indexing.worker.WorkerCuratorCoordinator;
 import org.apache.druid.indexing.worker.WorkerTaskMonitor;
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.server.initialization.IndexerZkConfig;
 import org.apache.druid.server.initialization.ZkPathsConfig;
 import org.easymock.EasyMock;
@@ -95,6 +97,7 @@ public String getBase()
         }, null, null, null, null),
         new RemoteTaskRunnerConfig(),
         cf,
+        new NodeAnnouncer(cf, Execs.directExecutor()),
         worker
     );
     curatorCoordinator.start();

server/src/main/java/org/apache/druid/curator/CuratorConfig.java

@@ -63,6 +63,9 @@ public class CuratorConfig
   @JsonProperty("maxZkRetries")
   private int maxZkRetries = 29;
 
+  @JsonProperty("pathChildrenCacheStrategy")
+  private boolean pathChildrenCacheStrategy = true;
+
   public static CuratorConfig create(String hosts)
   {
     CuratorConfig config = new CuratorConfig();
@@ -141,4 +144,9 @@ public int getMaxZkRetries()
   {
     return maxZkRetries;
   }
+
+  public boolean getPathChildrenCacheStrategy()
+  {
+    return pathChildrenCacheStrategy;
+  }
 }

server/src/main/java/org/apache/druid/curator/announcement/Announceable.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.curator.announcement;
+
+/**
+ * The {@link Announceable} is a representation of an announcement to be made in ZooKeeper.
+ */
+class Announceable
+{
+  /**
+   * Represents the path in ZooKeeper where the announcement will be made.
+   */
+  final String path;
+
+  /**
+   * Holds the actual data to be announced.
+   */
+  final byte[] bytes;
+
+  /**
+   * Indicates whether parent nodes should be removed if the announcement is created successfully.
+   * This can be useful for cleaning up unused paths in ZooKeeper.
+   */
+  final boolean removeParentsIfCreated;
+
+  public Announceable(String path, byte[] bytes, boolean removeParentsIfCreated)
+  {
+    this.path = path;
+    this.bytes = bytes;
+    this.removeParentsIfCreated = removeParentsIfCreated;
+  }
+
+  // This should be used for updates only, where removeParentsIfCreated is not relevant.
+  public Announceable(String path, byte[] bytes)
+  {
+    // removeParentsIfCreated is irrelevant, so we can use dummy value "false".
+    this(path, bytes, false);
+  }
+}