From 6c711153078230b1b64c4e1e87f3a7832338120f Mon Sep 17 00:00:00 2001 From: capistrant Date: Tue, 3 Mar 2026 15:14:20 -0600 Subject: [PATCH] Reduce log noise for k8s NodeRoleWatcher --- .../druid/discovery/BaseNodeRoleWatcher.java | 5 +-- .../discovery/BaseNodeRoleWatcherTest.java | 41 +++++++++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java b/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java index 997e863296a6..f17b43cf2c70 100644 --- a/server/src/main/java/org/apache/druid/discovery/BaseNodeRoleWatcher.java +++ b/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 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(), diff --git a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java index fa3cd0f44835..0decfcdef386 100644 --- a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java +++ b/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 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(