This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new b02a7651269 Fix broker routing ignoring a shutting down server while
another server is excluded (#19997)
b02a7651269 is described below
commit b02a76512697ff971368d1c1e23be91d8da8bd23
Author: Yash Mayya <[email protected]>
AuthorDate: Thu Oct 22 11:38:22 2026 -0700
Fix broker routing ignoring a shutting down server while another server is
excluded (#19997)
---
.../routing/manager/BaseBrokerRoutingManager.java | 21 +++++----------
.../routing/manager/BrokerRoutingManagerTest.java | 30 ++++++++++++++++++++++
2 files changed, 37 insertions(+), 14 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
index bb28278db52..6503606f600 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
@@ -486,21 +486,14 @@ public abstract class BaseBrokerRoutingManager
}
}
- // Calculate the routable servers and the changed routable servers
+ // Calculate the routable servers and the changed servers
+ // NOTE: All new enabled servers are routable because they are removed
from the excluded servers above. All new
+ // disabled servers are counted as changed, even excluded ones whose
routing does not change, so that the
+ // early termination below cannot skip removing them from
_enabledServerInstanceMap.
List<String> changedServers = new ArrayList<>(newEnabledServers.size() +
newDisabledServers.size());
- if (_excludedServers.isEmpty()) {
- changedServers.addAll(newEnabledServers);
- changedServers.addAll(newDisabledServers);
- } else {
- enabledServers.removeAll(_excludedServers);
- // NOTE: All new enabled servers are routable
- changedServers.addAll(newEnabledServers);
- for (String newDisabledServer : newDisabledServers) {
- if (_excludedServers.contains(newDisabledServer)) {
- changedServers.add(newDisabledServer);
- }
- }
- }
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ enabledServers.removeAll(_excludedServers);
_routableServerInstanceMap =
buildRoutableServerInstanceMap(enabledServers);
long calculateChangedServersEndTimeMs = System.currentTimeMillis();
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
index e82fac469fc..4bbd9d88535 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/manager/BrokerRoutingManagerTest.java
@@ -860,6 +860,36 @@ public class BrokerRoutingManagerTest {
assertTrue(_routingManager.getRoutableServerInstanceMap().containsKey(SERVER_INSTANCE_ID));
}
+ @Test
+ public void
testShuttingDownServerRemovedFromRoutingWhileAnotherServerIsExcluded()
+ throws Exception {
+ // Enable three servers, then exclude one of them (simulating
FailureDetector marking it unhealthy)
+ String excludedServer = "Server_excludedhost_8000";
+ String healthyServer = "Server_healthyhost_8000";
+ ZNRecord serverConfig = createEnabledServerZNRecord(SERVER_INSTANCE_ID);
+ ZNRecord excludedServerConfig =
createEnabledServerZNRecord(excludedServer);
+ ZNRecord healthyServerConfig = createEnabledServerZNRecord(healthyServer);
+ when(_zkDataAccessor.getChildren(eq(INSTANCE_CONFIGS_PATH), any(),
eq(AccessOption.PERSISTENT), anyInt(), anyInt()))
+ .thenReturn(List.of(serverConfig, excludedServerConfig,
healthyServerConfig));
+ _routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+ _routingManager.excludeServerFromRouting(excludedServer);
+ InstanceSelector instanceSelector = mock(InstanceSelector.class);
+ putRoutingEntry(TEST_TABLE, createRoutingEntry(TEST_TABLE, null, null,
Map.of(), instanceSelector, false));
+
+ // The server starts shutting down while the other server is still excluded
+ ZNRecord shuttingDownServerConfig =
createEnabledServerZNRecord(SERVER_INSTANCE_ID);
+ shuttingDownServerConfig.setSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS,
"true");
+ when(_zkDataAccessor.getChildren(eq(INSTANCE_CONFIGS_PATH), any(),
eq(AccessOption.PERSISTENT), anyInt(), anyInt()))
+ .thenReturn(List.of(shuttingDownServerConfig, excludedServerConfig,
healthyServerConfig));
+ _routingManager.processClusterChange(ChangeType.INSTANCE_CONFIG);
+
+ // The routing entry must stop selecting the shutting down server, and
only the healthy server stays routable
+ verify(instanceSelector).onInstancesChange(Set.of(healthyServer),
List.of(SERVER_INSTANCE_ID));
+ assertEquals(_routingManager.getRoutableServerInstanceMap().keySet(),
Set.of(healthyServer));
+ // The excluded server stays enabled
+ assertEquals(_routingManager.getEnabledServerInstanceMap().keySet(),
Set.of(excludedServer, healthyServer));
+ }
+
@Test
public void testPartitionPruningThresholdUpdates()
throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
