Include empty servers in BrokerServerView.

benchmarks/src/test/java/org/apache/druid/benchmark/query/CachingClusteredClientBenchmark.java

@@ -556,7 +556,7 @@ public void registerTimelineCallback(Executor exec, TimelineCallback callback)
     }
 
     @Override
-    public void registerServerRemovedCallback(Executor exec, ServerRemovedCallback callback)
+    public void registerServerCallback(Executor exec, ServerCallback callback)
     {
       // do nothing
     }

extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java

@@ -352,7 +352,7 @@ public void registerSegmentCallback(Executor exec, SegmentCallback callback)
           }
 
           @Override
-          public void registerServerRemovedCallback(Executor exec, ServerRemovedCallback callback)
+          public void registerServerCallback(Executor exec, ServerCallback callback)
           {
 
           }

server/src/main/java/org/apache/druid/client/BatchServerInventoryView.java

@@ -65,7 +65,7 @@ public class BatchServerInventoryView implements ServerInventoryView, FilteredSe
   private final CuratorInventoryManager<DruidServer, Set<DataSegment>> inventoryManager;
   private final AtomicBoolean started = new AtomicBoolean(false);
 
-  private final ConcurrentMap<ServerRemovedCallback, Executor> serverRemovedCallbacks = new ConcurrentHashMap<>();
+  private final ConcurrentMap<ServerCallback, Executor> serverCallbacks = new ConcurrentHashMap<>();
   private final ConcurrentMap<SegmentCallback, Executor> segmentCallbacks = new ConcurrentHashMap<>();
 
   private final ConcurrentMap<String, Set<DataSegment>> zNodes = new ConcurrentHashMap<>();
@@ -127,13 +127,14 @@ public Set<DataSegment> deserializeInventory(byte[] bytes)
           public void newContainer(DruidServer container)
           {
             log.info("New Server[%s]", container);
+            runServerCallbacks(callback -> callback.serverAdded(container));
           }
 
           @Override
           public void deadContainer(DruidServer deadContainer)
           {
             log.info("Server Disappeared[%s]", deadContainer);
-            runServerRemovedCallbacks(deadContainer);
+            runServerCallbacks(callback -> callback.serverRemoved(deadContainer));
           }
 
           @Override
@@ -216,9 +217,9 @@ public Collection<DruidServer> getInventory()
   }
 
   @Override
-  public void registerServerRemovedCallback(Executor exec, ServerRemovedCallback callback)
+  public void registerServerCallback(Executor exec, ServerCallback callback)
   {
-    serverRemovedCallbacks.put(callback, exec);
+    serverCallbacks.put(callback, exec);
   }
 
   @Override
@@ -243,13 +244,13 @@ protected void runSegmentCallbacks(
     }
   }
 
-  private void runServerRemovedCallbacks(final DruidServer server)
+  private void runServerCallbacks(final Function<ServerCallback, CallbackAction> fn)
   {
-    for (final Map.Entry<ServerRemovedCallback, Executor> entry : serverRemovedCallbacks.entrySet()) {
+    for (final Map.Entry<ServerCallback, Executor> entry : serverCallbacks.entrySet()) {
       entry.getValue().execute(
           () -> {
-            if (CallbackAction.UNREGISTER == entry.getKey().serverRemoved(server)) {
-              serverRemovedCallbacks.remove(entry.getKey());
+            if (CallbackAction.UNREGISTER == fn.apply(entry.getKey())) {
+              serverCallbacks.remove(entry.getKey());
             }
           }
       );

server/src/main/java/org/apache/druid/client/BrokerServerView.java

@@ -157,11 +157,25 @@ public CallbackAction segmentSchemasAnnounced(SegmentSchemas segmentSchemas)
         segmentFilter
     );
 
-    baseView.registerServerRemovedCallback(
+    baseView.registerServerCallback(
         exec,
-        server -> {
-          removeServer(server);
-          return CallbackAction.CONTINUE;
+        new ServerCallback() {
+          @Override
+          public CallbackAction serverAdded(DruidServer server)
+          {
+            // We don't track brokers in this view.
+            if (!server.getType().equals(ServerType.BROKER)) {
+              addServer(server);
+            }
+            return CallbackAction.CONTINUE;
+          }
+
+          @Override
+          public CallbackAction serverRemoved(DruidServer server)
+          {
+            removeServer(server);
+            return CallbackAction.CONTINUE;
+          }
         }
     );
   }
@@ -378,9 +392,9 @@ public <T> QueryRunner<T> getQueryRunner(DruidServer server)
   }
 
   @Override
-  public void registerServerRemovedCallback(Executor exec, ServerRemovedCallback callback)
+  public void registerServerCallback(Executor exec, ServerCallback callback)
   {
-    baseView.registerServerRemovedCallback(exec, callback);
+    baseView.registerServerCallback(exec, callback);
   }
 
   @Override

server/src/main/java/org/apache/druid/client/CoordinatorServerView.java

@@ -125,10 +125,16 @@ public ServerView.CallbackAction segmentSchemasAnnounced(SegmentSchemas segmentS
         }
     );
 
-    baseView.registerServerRemovedCallback(
+    baseView.registerServerCallback(
         exec,
-        new ServerView.ServerRemovedCallback()
+        new ServerView.ServerCallback()
         {
+          @Override
+          public ServerView.CallbackAction serverAdded(DruidServer server)
+          {
+            return ServerView.CallbackAction.CONTINUE;
+          }
+
           @Override
           public ServerView.CallbackAction serverRemoved(DruidServer server)
           {

server/src/main/java/org/apache/druid/client/FilteredServerInventoryView.java

@@ -34,5 +34,5 @@ void registerSegmentCallback(
       Predicate<Pair<DruidServerMetadata, DataSegment>> filter
   );
 
-  void registerServerRemovedCallback(Executor exec, ServerView.ServerRemovedCallback callback);
+  void registerServerCallback(Executor exec, ServerView.ServerCallback callback);
 }

server/src/main/java/org/apache/druid/client/HttpServerInventoryView.java

@@ -93,7 +93,7 @@ public class HttpServerInventoryView implements ServerInventoryView, FilteredSer
 
   private final LifecycleLock lifecycleLock = new LifecycleLock();
 
-  private final ConcurrentMap<ServerRemovedCallback, Executor> serverCallbacks = new ConcurrentHashMap<>();
+  private final ConcurrentMap<ServerCallback, Executor> serverCallbacks = new ConcurrentHashMap<>();
   private final ConcurrentMap<SegmentCallback, Executor> segmentCallbacks = new ConcurrentHashMap<>();
 
   private final ConcurrentMap<SegmentCallback, Predicate<Pair<DruidServerMetadata, DataSegment>>> segmentPredicates =
@@ -286,7 +286,7 @@ public void registerSegmentCallback(
   }
 
   @Override
-  public void registerServerRemovedCallback(Executor exec, ServerRemovedCallback callback)
+  public void registerServerCallback(Executor exec, ServerCallback callback)
   {
     if (lifecycleLock.isStarted()) {
       throw new ISE("Lifecycle has already started.");
@@ -347,18 +347,13 @@ public void run()
     }
   }
 
-  private void runServerRemovedCallbacks(final DruidServer server)
+  private void runServerCallbacks(final Function<ServerCallback, CallbackAction> fn)
   {
-    for (final Map.Entry<ServerRemovedCallback, Executor> entry : serverCallbacks.entrySet()) {
+    for (final Map.Entry<ServerCallback, Executor> entry : serverCallbacks.entrySet()) {
       entry.getValue().execute(
-          new Runnable()
-          {
-            @Override
-            public void run()
-            {
-              if (CallbackAction.UNREGISTER == entry.getKey().serverRemoved(server)) {
-                serverCallbacks.remove(entry.getKey());
-              }
+          () -> {
+            if (CallbackAction.UNREGISTER == fn.apply(entry.getKey())) {
+              serverCallbacks.remove(entry.getKey());
             }
           }
       );
@@ -421,12 +416,13 @@ private void updateFinalPredicate()
   void serverAdded(DruidServer server)
   {
     synchronized (servers) {
-      DruidServerHolder holder = servers.get(server.getName());
-      if (holder == null) {
+      DruidServerHolder existing = servers.get(server.getName());
+      if (existing == null) {
         log.info("Server[%s] appeared.", server.getName());
-        holder = new DruidServerHolder(server);
-        servers.put(server.getName(), holder);
-        holder.start();
+        final DruidServerHolder newHolder = new DruidServerHolder(server);
+        servers.put(server.getName(), newHolder);
+        runServerCallbacks(callback -> callback.serverAdded(newHolder.druidServer));
+        newHolder.start();
       } else {
         log.info("Server[%s] already exists.", server.getName());
       }
@@ -440,7 +436,7 @@ private void serverRemoved(DruidServer server)
       if (holder != null) {
         log.info("Server[%s] disappeared.", server.getName());
         holder.stop();
-        runServerRemovedCallbacks(holder.druidServer);
+        runServerCallbacks(callback -> callback.serverRemoved(holder.druidServer));
       } else {
         log.info("Ignoring remove notification for unknown server[%s].", server.getName());
       }

server/src/main/java/org/apache/druid/client/ServerView.java

@@ -29,7 +29,7 @@
  */
 public interface ServerView
 {
-  void registerServerRemovedCallback(Executor exec, ServerRemovedCallback callback);
+  void registerServerCallback(Executor exec, ServerCallback callback);
   void registerSegmentCallback(Executor exec, SegmentCallback callback);
 
   enum CallbackAction
@@ -38,8 +38,24 @@ enum CallbackAction
     UNREGISTER,
   }
 
-  interface ServerRemovedCallback
+  interface ServerCallback
   {
+    /**
+     * Called when a server is added.
+     *
+     * The return value indicates if this callback has completed its work.  Note that even if this callback
+     * indicates that it should be unregistered, it is not possible to guarantee that this callback will not
+     * get called again.  There is a race condition between when this callback runs and other events that can cause
+     * the callback to be queued for running.  Thus, callbacks shouldn't assume that they will not get called
+     * again after they are done.  The contract is that the callback will eventually be unregistered, enforcing
+     * a happens-before relationship is not part of the contract.
+     *
+     * @param server The server that was added.
+     * @return UNREGISTER if the callback has completed its work and should be unregistered.  CONTINUE if the callback
+     * should remain registered.
+     */
+    CallbackAction serverAdded(DruidServer server);
+
     /**
      * Called when a server is removed.
      *

server/src/main/java/org/apache/druid/curator/inventory/CuratorInventoryManager.java

@@ -259,8 +259,8 @@ public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) th
               containers.put(containerKey, new ContainerHolder(container, inventoryCache));
 
               log.debug("Starting inventory cache for %s, inventoryPath %s", containerKey, inventoryPath);
-              inventoryCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
               strategy.newContainer(container);
+              inventoryCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
             }
           }
           break;

server/src/main/java/org/apache/druid/server/coordinator/balancer/CachingCostBalancerStrategyFactory.java

@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.annotation.JacksonInject;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.google.common.util.concurrent.ListeningExecutorService;
+import org.apache.druid.client.DruidServer;
 import org.apache.druid.client.ServerInventoryView;
 import org.apache.druid.client.ServerView;
 import org.apache.druid.java.util.common.concurrent.Execs;
@@ -106,13 +107,23 @@ public ServerView.CallbackAction segmentSchemasAnnounced(SegmentSchemas segmentS
         }
     );
 
-    serverInventoryView.registerServerRemovedCallback(
+    serverInventoryView.registerServerCallback(
         executor,
-        server -> {
-          if (server.isSegmentReplicationTarget()) {
-            clusterCostCacheBuilder.removeServer(server.getName());
+        new ServerView.ServerCallback() {
+          @Override
+          public ServerView.CallbackAction serverAdded(DruidServer server)
+          {
+            return ServerView.CallbackAction.CONTINUE;
+          }
+
+          @Override
+          public ServerView.CallbackAction serverRemoved(DruidServer server)
+          {
+            if (server.isSegmentReplicationTarget()) {
+              clusterCostCacheBuilder.removeServer(server.getName());
+            }
+            return ServerView.CallbackAction.CONTINUE;
           }
-          return ServerView.CallbackAction.CONTINUE;
         }
     );
   }

server/src/test/java/org/apache/druid/client/BrokerServerViewTest.java

@@ -76,6 +76,7 @@ public class BrokerServerViewTest extends CuratorTestBase
   private final ZkPathsConfig zkPathsConfig;
 
   private CountDownLatch segmentViewInitLatch;
+  private CountDownLatch serverAddedLatch;
   private CountDownLatch segmentAddedLatch;
   private CountDownLatch segmentRemovedLatch;
 
@@ -158,7 +159,7 @@ public void testMultipleServerAddedRemovedSegment() throws Exception
     setupViews();
 
     final List<DruidServer> druidServers = Lists.transform(
-        ImmutableList.of("locahost:0", "localhost:1", "localhost:2", "localhost:3", "localhost:4"),
+        ImmutableList.of("localhost:0", "localhost:1", "localhost:2", "localhost:3", "localhost:4"),
         hostname -> setupHistoricalServer("default_tier", hostname, 0)
     );
 
@@ -237,6 +238,7 @@ public void testMultipleServerAddedRemovedSegment() throws Exception
   public void testMultipleServerAndBroker() throws Exception
   {
     segmentViewInitLatch = new CountDownLatch(1);
+    serverAddedLatch = new CountDownLatch(6);
     segmentAddedLatch = new CountDownLatch(6);
 
     // temporarily set latch count to 1
@@ -254,13 +256,26 @@ public void testMultipleServerAndBroker() throws Exception
         0
     );
 
-    final List<DruidServer> druidServers = Lists.transform(
-        ImmutableList.of("locahost:0", "localhost:1", "localhost:2", "localhost:3", "localhost:4"),
-        hostname -> setupHistoricalServer("default_tier", hostname, 0)
-    );
+    // Materialize this list so all servers are set up
+    final List<DruidServer> druidServers =
+        ImmutableList.copyOf(
+            Lists.transform(
+                ImmutableList.of("localhost:0", "localhost:1", "localhost:2", "localhost:3", "localhost:4"),
+                hostname -> setupHistoricalServer("default_tier", hostname, 0)
+            )
+        );
 
     setupZNodeForServer(druidBroker, zkPathsConfig, jsonMapper);
 
+    Assert.assertTrue(timing.forWaiting().awaitLatch(segmentViewInitLatch));
+    Assert.assertTrue(timing.forWaiting().awaitLatch(serverAddedLatch));
+
+    // check server metadatas
+    Assert.assertEquals(
+        druidServers.stream().map(DruidServer::getMetadata).collect(Collectors.toSet()),
+        ImmutableSet.copyOf(brokerServerView.getDruidServerMetadatas())
+    );
+
     final List<DataSegment> segments = Lists.transform(
         ImmutableList.of(
             Pair.of("2011-04-01/2011-04-03", "v1"),
@@ -277,7 +292,6 @@ public void testMultipleServerAndBroker() throws Exception
     for (int i = 0; i < 5; ++i) {
       announceSegmentForServer(druidServers.get(i), segments.get(i), zkPathsConfig, jsonMapper);
     }
-    Assert.assertTrue(timing.forWaiting().awaitLatch(segmentViewInitLatch));
     Assert.assertTrue(timing.forWaiting().awaitLatch(segmentAddedLatch));
 
     TimelineLookup timeline = brokerServerView.getTimeline(
@@ -297,12 +311,6 @@ public void testMultipleServerAndBroker() throws Exception
         )
     );
 
-    // check server metadatas
-    Assert.assertEquals(
-        druidServers.stream().map(DruidServer::getMetadata).collect(Collectors.toSet()),
-        ImmutableSet.copyOf(brokerServerView.getDruidServerMetadatas())
-    );
-
     // unannounce the broker segment should do nothing to announcements
     unannounceSegmentForServer(druidBroker, brokerSegment, zkPathsConfig);
     Assert.assertTrue(timing.forWaiting().awaitLatch(segmentRemovedLatch));
@@ -617,6 +625,29 @@ private void setupViews(Set<String> watchedTiers, Set<String> ignoredTiers, bool
         "test"
     )
     {
+      @Override
+      public void registerServerCallback(Executor exec, ServerCallback callback)
+      {
+        super.registerServerCallback(
+            exec,
+            new ServerCallback() {
+              @Override
+              public CallbackAction serverAdded(DruidServer server)
+              {
+                final CallbackAction res = callback.serverAdded(server);
+                serverAddedLatch.countDown();
+                return res;
+              }
+
+              @Override
+              public CallbackAction serverRemoved(DruidServer server)
+              {
+                return callback.serverRemoved(server);
+              }
+            }
+        );
+      }
+
       @Override
       public void registerSegmentCallback(Executor exec, final SegmentCallback callback)
       {