Reduce log noise for k8s NodeRoleWatcher (#19077)

Author: capistrant

server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java

@@ -141,8 +141,6 @@ public void childAdded(DiscoveryDruidNode druidNode)
         return;
       }
 
-      LOGGER.info("Node [%s] of role [%s] detected.", druidNode.getDruidNode().getUriToUse(), nodeRole.getJsonName());
-
       addNode(druidNode);
     }
   }
@@ -152,6 +150,7 @@ private void addNode(DiscoveryDruidNode druidNode)
   {
     DiscoveryDruidNode prev = nodes.putIfAbsent(druidNode.getDruidNode().getHostAndPortToUse(), druidNode);
     if (prev == null) {
+      LOGGER.info("Node [%s] of role [%s] detected.", druidNode.getDruidNode().getUriToUse(), nodeRole.getJsonName());
       // No need to wait on CountDownLatch, because we are holding the lock under which it could only be counted down.
       if (cacheInitialized.getCount() == 0) {
         List<DiscoveryDruidNode> newNode = ImmutableList.of(druidNode);
@@ -165,7 +164,7 @@ private void addNode(DiscoveryDruidNode druidNode)
         }
       }
     } else {
-      LOGGER.error(
+      LOGGER.debug(
           "Node [%s] of role [%s] discovered but existed already [%s].",
           druidNode.getDruidNode().getUriToUse(),
           nodeRole.getJsonName(),

server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java

@@ -203,6 +203,47 @@ public void testGetAllNodesBeforeTimeout() throws InterruptedException
     assertListener(listener1, true, ImmutableList.of(broker1, broker3), ImmutableList.of());
   }
 
+  @Test(timeout = 60_000L)
+  public void testDuplicateChildAddedAfterResetNodesDoesNotNotifyListeners()
+  {
+    BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER);
+
+    DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1");
+    DiscoveryDruidNode broker2 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker2");
+
+    // Initial discovery and cache initialization
+    nodeRoleWatcher.childAdded(broker1);
+    nodeRoleWatcher.childAdded(broker2);
+    nodeRoleWatcher.cacheInitialized();
+
+    TestListener listener = new TestListener();
+    nodeRoleWatcher.registerListener(listener);
+
+    // Verify listener received the initial nodes
+    Assert.assertEquals(ImmutableList.of(broker1, broker2), listener.nodesAddedList);
+
+    // Simulate watch reconnect: resetNodes with the same set of nodes
+    LinkedHashMap<String, DiscoveryDruidNode> resetMap = new LinkedHashMap<>();
+    resetMap.put(broker1.getDruidNode().getHostAndPortToUse(), broker1);
+    resetMap.put(broker2.getDruidNode().getHostAndPortToUse(), broker2);
+    nodeRoleWatcher.resetNodes(resetMap);
+
+    // No new additions or removals since the node set is unchanged
+    Assert.assertEquals(ImmutableList.of(broker1, broker2), listener.nodesAddedList);
+    Assert.assertTrue(listener.nodesRemovedList.isEmpty());
+
+    // Simulate K8s watch replaying ADDED events for already-present nodes
+    nodeRoleWatcher.childAdded(broker1);
+    nodeRoleWatcher.childAdded(broker2);
+
+    // Listeners should NOT be notified again — the duplicate adds are no-ops
+    Assert.assertEquals(ImmutableList.of(broker1, broker2), listener.nodesAddedList);
+    Assert.assertTrue(listener.nodesRemovedList.isEmpty());
+
+    // The nodes map should still contain exactly the same two nodes
+    Assert.assertEquals(ImmutableSet.of(broker1, broker2), new HashSet<>(nodeRoleWatcher.getAllNodes()));
+  }
+
   private DiscoveryDruidNode buildDiscoveryDruidNode(NodeRole role, String host)
   {
     return new DiscoveryDruidNode(