Fix huge number of watches in zk

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

@@ -25,11 +25,10 @@
 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.NodeAnnouncer;
 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 +53,7 @@ public class WorkerCuratorCoordinator
   private final ObjectMapper jsonMapper;
   private final RemoteTaskRunnerConfig config;
   private final CuratorFramework curatorFramework;
-  private final Announcer announcer;
+  private final NodeAnnouncer announcer;
 
   private final String baseAnnouncementsPath;
   private final String baseTaskPath;
@@ -77,7 +76,7 @@ public WorkerCuratorCoordinator(
     this.curatorFramework = curatorFramework;
     this.worker = worker;
 
-    this.announcer = new Announcer(curatorFramework, Execs.directExecutor());
+    this.announcer = new NodeAnnouncer(curatorFramework);
 
     this.baseAnnouncementsPath = getPath(Arrays.asList(indexerZkConfig.getAnnouncementsPath(), worker.getHost()));
     this.baseTaskPath = getPath(Arrays.asList(indexerZkConfig.getTasksPath(), worker.getHost()));

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

@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+class Announceable
+{
+  final String path;
+  final byte[] bytes;
+  final boolean removeParentsIfCreated;
+
+  public Announceable(String path, byte[] bytes, boolean removeParentsIfCreated)
+  {
+    this.path = path;
+    this.bytes = bytes;
+    this.removeParentsIfCreated = removeParentsIfCreated;
+  }
+}

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

@@ -36,10 +36,12 @@
 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;
+import org.apache.druid.server.ZKPathsUtils;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.data.Stat;
 
+import javax.annotation.concurrent.GuardedBy;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -53,7 +55,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
- * Announces things on Zookeeper.
+ * {@link NodeAnnouncer} announces a single node on Zookeeper and only watches this node,
+ * while {@link Announcer} watches all child paths, not only this node.
  */
 public class Announcer
 {
@@ -63,11 +66,13 @@ public class Announcer
   private final PathChildrenCacheFactory factory;
   private final ExecutorService pathChildrenCacheExecutor;
 
+  @GuardedBy("toAnnounce")
   private final List<Announceable> toAnnounce = new ArrayList<>();
+  @GuardedBy("toAnnounce")
   private final List<Announceable> toUpdate = new ArrayList<>();
   private final ConcurrentMap<String, PathChildrenCache> listeners = new ConcurrentHashMap<>();
   private final ConcurrentMap<String, ConcurrentMap<String, byte[]>> announcements = new ConcurrentHashMap<>();
-  private final List<String> parentsIBuilt = new CopyOnWriteArrayList<String>();
+  private final List<String> parentsIBuilt = new CopyOnWriteArrayList<>();
 
   // Used for testing
   private Set<String> addedChildren;
@@ -229,7 +234,7 @@ public void announce(String path, byte[] bytes, boolean removeParentIfCreated)
           cache.getListenable().addListener(
               new PathChildrenCacheListener()
               {
-                private final AtomicReference<Set<String>> pathsLost = new AtomicReference<Set<String>>(null);
+                private final AtomicReference<Set<String>> pathsLost = new AtomicReference<>(null);
 
                 @Override
                 public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
@@ -240,8 +245,7 @@ public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) th
                   switch (event.getType()) {
                     case CHILD_REMOVED:
                       final ChildData child = event.getData();
-                      final ZKPaths.PathAndNode childPath = ZKPaths.getPathAndNode(child.getPath());
-                      final byte[] value = finalSubPaths.get(childPath.getNode());
+                      final byte[] value = finalSubPaths.get(ZKPathsUtils.getParentNode(child.getPath()));
                       if (value != null) {
                         log.info("Node[%s] dropped, reinstating.", child.getPath());
                         createAnnouncement(child.getPath(), value);
@@ -432,18 +436,4 @@ private void createPath(String parentPath, boolean removeParentsIfCreated)
       log.info(e, "Problem creating parentPath[%s], someone else created it first?", parentPath);
     }
   }
-
-  private static class Announceable
-  {
-    final String path;
-    final byte[] bytes;
-    final boolean removeParentsIfCreated;
-
-    public Announceable(String path, byte[] bytes, boolean removeParentsIfCreated)
-    {
-      this.path = path;
-      this.bytes = bytes;
-      this.removeParentsIfCreated = removeParentsIfCreated;
-    }
-  }
 }