From e44eb743152d2f7839578319b90d1b1457d77c39 Mon Sep 17 00:00:00 2001 From: sarkaramrit1993 Date: Tue, 13 Oct 2026 20:56:16 -0700 Subject: [PATCH 1/4] SOLR-17971: Check the distributed async tracker in REBALANCELEADERS With the Overseer disabled, sub-command results go to DistributedApiAsyncTracker, not the Overseer maps, so the wait loop always ran to maxWaitSeconds and never cleared the async ids. --- ...lr-17971-rebalance-leaders-distributed.yml | 7 ++++ .../solr/handler/admin/RebalanceLeaders.java | 33 +++++++++------ .../solr/cloud/TestRebalanceLeaders.java | 40 +++++++++++++++++++ 3 files changed, 68 insertions(+), 12 deletions(-) create mode 100644 changelog/unreleased/solr-17971-rebalance-leaders-distributed.yml diff --git a/changelog/unreleased/solr-17971-rebalance-leaders-distributed.yml b/changelog/unreleased/solr-17971-rebalance-leaders-distributed.yml new file mode 100644 index 000000000000..a4c4fb79316c --- /dev/null +++ b/changelog/unreleased/solr-17971-rebalance-leaders-distributed.yml @@ -0,0 +1,7 @@ +title: "REBALANCELEADERS no longer waits until timeout or leaves async ids behind when the Overseer is disabled." +type: fixed +authors: + - name: Amrit Sarkar +links: + - name: SOLR-17971 + url: https://issues.apache.org/jira/browse/SOLR-17971 diff --git a/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java b/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java index 1a435a1ad978..ffa73c037c9c 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java @@ -33,11 +33,14 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.solr.cloud.LeaderElector; import org.apache.solr.cloud.OverseerTaskProcessor; +import org.apache.solr.cloud.ZkController; import org.apache.solr.cloud.api.collections.AdminCmdContext; +import org.apache.solr.cloud.api.collections.DistributedCollectionConfigSetCommandRunner; import org.apache.solr.cloud.overseer.SliceMutator; import org.apache.solr.common.SolrException; import org.apache.solr.common.cloud.ClusterState; @@ -122,7 +125,7 @@ class RebalanceLeaders { coreContainer = collectionsHandler.getCoreContainer(); } - void execute() throws KeeperException, InterruptedException { + void execute() throws Exception { DocCollection dc = checkParams(); int max = req.getParams().getInt(MAX_AT_ONCE_PROP, Integer.MAX_VALUE); @@ -471,8 +474,7 @@ private void rejoinElectionQueue( // mean failure! // - private boolean waitAsyncRequests(final int maxWaitSecs, Boolean waitForAll) - throws KeeperException, InterruptedException { + private boolean waitAsyncRequests(final int maxWaitSecs, Boolean waitForAll) throws Exception { if (asyncRequests.size() == 0) { return true; @@ -482,15 +484,7 @@ private boolean waitAsyncRequests(final int maxWaitSecs, Boolean waitForAll) Iterator iter = asyncRequests.iterator(); boolean foundChange = false; while (iter.hasNext()) { - String asyncId = iter.next(); - if (coreContainer.getZkController().getOverseerFailureMap().contains(asyncId)) { - coreContainer.getZkController().getOverseerFailureMap().remove(asyncId); - coreContainer.getZkController().clearAsyncId(asyncId); - iter.remove(); - foundChange = true; - } else if (coreContainer.getZkController().getOverseerCompletedMap().contains(asyncId)) { - coreContainer.getZkController().getOverseerCompletedMap().remove(asyncId); - coreContainer.getZkController().clearAsyncId(asyncId); + if (removeIfDone(iter.next())) { iter.remove(); foundChange = true; } @@ -509,6 +503,21 @@ private boolean waitAsyncRequests(final int maxWaitSecs, Boolean waitForAll) return false; } + private boolean removeIfDone(String asyncId) throws Exception { + ZkController zkController = coreContainer.getZkController(); + Optional runner = + zkController.getDistributedCommandRunner(); + if (runner.isPresent()) { + return runner.get().deleteSingleAsyncId(asyncId); + } + if (zkController.getOverseerFailureMap().remove(asyncId) + || zkController.getOverseerCompletedMap().remove(asyncId)) { + zkController.clearAsyncId(asyncId); + return true; + } + return false; + } + // If we actually changed the leader, we should send that fact back in the response. private void addToSuccesses(Slice slice, Replica replica) { SimpleOrderedMap> successes = results.get("successes"); diff --git a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java index de4f130fecbc..71571501f262 100644 --- a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java +++ b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java @@ -157,6 +157,46 @@ public void testRebalanceLeaders() throws Exception { checkPreferredsAreLeaders(); } + // Sub-command async ids must be found and cleared in both Overseer and distributed modes. + @Test + public void testRebalanceLeadersClearsAsyncIds() throws Exception { + DocCollection docCollection = + cluster.getSolrClient().getClusterState().getCollection(COLLECTION_NAME); + for (Slice slice : docCollection.getSlices()) { + Replica nonLeader = + slice.getReplicas().stream() + .filter(rep -> rep.getName().equals(slice.getLeader().getName()) == false) + .findFirst() + .orElseThrow(); + setProp(slice, nonLeader, "preferredLeader"); + } + + int maxWaitSeconds = 30; + long start = System.nanoTime(); + CollectionAdminResponse resp = + CollectionAdminRequest.rebalanceLeaders(COLLECTION_NAME) + .setMaxWaitSeconds(maxWaitSeconds) + .process(cluster.getSolrClient()); + long elapsedSeconds = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - start); + assertTrue(resp.getResponse().get("Summary").toString().contains("Success")); + + for (String path : + List.of( + "/distributedapi/async/persistent", + "/overseer/collection-map-completed", + "/overseer/collection-map-failure")) { + if (zkClient().exists(path)) { + List leftover = + zkClient().getChildren(path, null).stream() + .filter(id -> id.startsWith("mn-rebalanceleaders_")) + .toList(); + assertTrue("Async ids not cleared from " + path + ": " + leftover, leftover.isEmpty()); + } + } + assertTrue( + "Waited " + elapsedSeconds + "s, hit maxWaitSeconds", elapsedSeconds < maxWaitSeconds); + } + // Ensure that the property is set on only one replica per slice when changing a unique property // on an individual // replica. From 0b820daec95ca644f99d1a82c5a95b9cf8a58906 Mon Sep 17 00:00:00 2001 From: Amrit Sarkar Date: Wed, 14 Oct 2026 20:18:19 -0700 Subject: [PATCH 2/4] SOLR-17971: Share stored async response removal between REBALANCELEADERS and DELETESTATUS --- .../org/apache/solr/cloud/ZkController.java | 18 +++++++++++++++++ .../solr/handler/admin/RebalanceLeaders.java | 20 +------------------ .../handler/admin/api/ClusterCommands.java | 15 +------------- 3 files changed, 20 insertions(+), 33 deletions(-) diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java index 657b2728d36c..5db9f8d7256a 100644 --- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java +++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java @@ -2510,6 +2510,24 @@ public boolean clearAsyncId(String asyncId) throws KeeperException { } } + /** + * Removes the stored response of a completed or failed async operation and releases its asyncId, + * in both Overseer and distributed modes. An asyncId with no stored response is left claimed, + * since its task may still be running or pending. + * + * @return True if a stored response was found and removed. + */ + public boolean removeStoredAsyncResponse(String asyncId) throws Exception { + if (distributedCommandRunner.isPresent()) { + return distributedCommandRunner.get().deleteSingleAsyncId(asyncId); + } + if (overseerCompletedMap.remove(asyncId) || overseerFailureMap.remove(asyncId)) { + clearAsyncId(asyncId); + return true; + } + return false; + } + public Overseer getOverseer() { return overseer; } diff --git a/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java b/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java index ffa73c037c9c..eb27f99f9315 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/RebalanceLeaders.java @@ -33,14 +33,11 @@ import java.util.List; import java.util.Locale; import java.util.Map; -import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; import org.apache.solr.cloud.LeaderElector; import org.apache.solr.cloud.OverseerTaskProcessor; -import org.apache.solr.cloud.ZkController; import org.apache.solr.cloud.api.collections.AdminCmdContext; -import org.apache.solr.cloud.api.collections.DistributedCollectionConfigSetCommandRunner; import org.apache.solr.cloud.overseer.SliceMutator; import org.apache.solr.common.SolrException; import org.apache.solr.common.cloud.ClusterState; @@ -484,7 +481,7 @@ private boolean waitAsyncRequests(final int maxWaitSecs, Boolean waitForAll) thr Iterator iter = asyncRequests.iterator(); boolean foundChange = false; while (iter.hasNext()) { - if (removeIfDone(iter.next())) { + if (coreContainer.getZkController().removeStoredAsyncResponse(iter.next())) { iter.remove(); foundChange = true; } @@ -503,21 +500,6 @@ private boolean waitAsyncRequests(final int maxWaitSecs, Boolean waitForAll) thr return false; } - private boolean removeIfDone(String asyncId) throws Exception { - ZkController zkController = coreContainer.getZkController(); - Optional runner = - zkController.getDistributedCommandRunner(); - if (runner.isPresent()) { - return runner.get().deleteSingleAsyncId(asyncId); - } - if (zkController.getOverseerFailureMap().remove(asyncId) - || zkController.getOverseerCompletedMap().remove(asyncId)) { - zkController.clearAsyncId(asyncId); - return true; - } - return false; - } - // If we actually changed the leader, we should send that fact back in the response. private void addToSuccesses(Slice slice, Replica replica) { SimpleOrderedMap> successes = results.get("successes"); diff --git a/solr/core/src/java/org/apache/solr/handler/admin/api/ClusterCommands.java b/solr/core/src/java/org/apache/solr/handler/admin/api/ClusterCommands.java index 60cc18e483e5..4094f3732054 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/api/ClusterCommands.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/api/ClusterCommands.java @@ -129,20 +129,7 @@ public DeleteClusterCommandStatusResponse deleteClusterCommandStatus(String requ final DeleteClusterCommandStatusResponse response = instantiateJerseyResponse(DeleteClusterCommandStatusResponse.class); - final ZkController zkController = coreContainer.getZkController(); - - if (zkController.getDistributedCommandRunner().isEmpty()) { - if (zkController.getOverseerCompletedMap().remove(requestId)) { - zkController.clearAsyncId(requestId); - response.status = "successfully removed stored response for [" + requestId + "]"; - } else if (zkController.getOverseerFailureMap().remove(requestId)) { - zkController.clearAsyncId(requestId); - response.status = "successfully removed stored response for [" + requestId + "]"; - } else { - // Don't call zkController.clearAsyncId for this, since it could be a running/pending task - response.status = "[" + requestId + "] not found in stored responses"; - } - } else if (zkController.getDistributedCommandRunner().get().deleteSingleAsyncId(requestId)) { + if (coreContainer.getZkController().removeStoredAsyncResponse(requestId)) { response.status = "successfully removed stored response for [" + requestId + "]"; } else { response.status = "[" + requestId + "] not found in stored responses"; From f1242ddb50da6a41b1ad06838a62ce07c965489e Mon Sep 17 00:00:00 2001 From: Amrit Sarkar Date: Thu, 15 Oct 2026 02:03:02 -0700 Subject: [PATCH 3/4] SOLR-17971: Assert per mode through ZkController instead of hardcoded ZK paths --- .../solr/cloud/TestRebalanceLeaders.java | 42 +++++++++++-------- 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java index 71571501f262..6a163a96eded 100644 --- a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java +++ b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java @@ -157,7 +157,6 @@ public void testRebalanceLeaders() throws Exception { checkPreferredsAreLeaders(); } - // Sub-command async ids must be found and cleared in both Overseer and distributed modes. @Test public void testRebalanceLeadersClearsAsyncIds() throws Exception { DocCollection docCollection = @@ -171,30 +170,39 @@ public void testRebalanceLeadersClearsAsyncIds() throws Exception { setProp(slice, nonLeader, "preferredLeader"); } - int maxWaitSeconds = 30; + ZkController zkController = + cluster.getRandomJetty(random()).getCoreContainer().getZkController(); + boolean overseerMode = zkController.getDistributedCommandRunner().isEmpty(); + Set idsBefore = overseerMode ? overseerAsyncIds(zkController) : Set.of(); + + int maxWaitSeconds = 60; long start = System.nanoTime(); CollectionAdminResponse resp = CollectionAdminRequest.rebalanceLeaders(COLLECTION_NAME) .setMaxWaitSeconds(maxWaitSeconds) .process(cluster.getSolrClient()); long elapsedSeconds = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - start); - assertTrue(resp.getResponse().get("Summary").toString().contains("Success")); - - for (String path : - List.of( - "/distributedapi/async/persistent", - "/overseer/collection-map-completed", - "/overseer/collection-map-failure")) { - if (zkClient().exists(path)) { - List leftover = - zkClient().getChildren(path, null).stream() - .filter(id -> id.startsWith("mn-rebalanceleaders_")) - .toList(); - assertTrue("Async ids not cleared from " + path + ": " + leftover, leftover.isEmpty()); - } - } + assertTrue( + resp.getResponse().toString(), + resp.getResponse().get("Summary").toString().contains("Success")); + assertNotNull("No leader change was requested", resp.getResponse().get("successes")); + // With the Overseer disabled this is the only signal: the wait ends early only once every + // sub-command's stored response was removed. assertTrue( "Waited " + elapsedSeconds + "s, hit maxWaitSeconds", elapsedSeconds < maxWaitSeconds); + + if (overseerMode) { + Set leftover = overseerAsyncIds(zkController); + leftover.removeAll(idsBefore); + assertTrue("Async ids not cleared: " + leftover, leftover.isEmpty()); + } + } + + private Set overseerAsyncIds(ZkController zkController) throws Exception { + Set ids = new HashSet<>(zkController.getOverseerCompletedMap().keys()); + ids.addAll(zkController.getOverseerFailureMap().keys()); + ids.addAll(Overseer.getAsyncIdsMap(zkController.getZkClient()).keys()); + return ids; } // Ensure that the property is set on only one replica per slice when changing a unique property From 52b28342fde50509a403178e0245c92488f5c504 Mon Sep 17 00:00:00 2001 From: sarkaramrit1993 Date: Thu, 15 Oct 2026 14:25:28 -0700 Subject: [PATCH 4/4] SOLR-17971: Check for leaked async ids with the Overseer disabled too --- .../cloud/DistributedApiAsyncTracker.java | 5 +++++ ...butedCollectionConfigSetCommandRunner.java | 8 +++++++ .../solr/cloud/TestRebalanceLeaders.java | 22 ++++++++++--------- 3 files changed, 25 insertions(+), 10 deletions(-) diff --git a/solr/core/src/java/org/apache/solr/cloud/DistributedApiAsyncTracker.java b/solr/core/src/java/org/apache/solr/cloud/DistributedApiAsyncTracker.java index 556ea22211d9..c10a6870728e 100644 --- a/solr/core/src/java/org/apache/solr/cloud/DistributedApiAsyncTracker.java +++ b/solr/core/src/java/org/apache/solr/cloud/DistributedApiAsyncTracker.java @@ -286,6 +286,11 @@ public boolean deleteSingleAsyncId(String asyncId) throws Exception { && trackedAsyncTasks.remove(asyncId); } + @VisibleForTesting + public Collection getTrackedAsyncIds() throws Exception { + return trackedAsyncTasks.keys(); + } + /** * Deletes all async id's for completed or failed async jobs. Does not touch id's for submitted or * running jobs. diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/DistributedCollectionConfigSetCommandRunner.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/DistributedCollectionConfigSetCommandRunner.java index 80797e0e91fb..77f9481b196a 100644 --- a/solr/core/src/java/org/apache/solr/cloud/api/collections/DistributedCollectionConfigSetCommandRunner.java +++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/DistributedCollectionConfigSetCommandRunner.java @@ -24,8 +24,10 @@ import static org.apache.solr.common.cloud.ZkStateReader.SHARD_ID_PROP; import static org.apache.solr.common.params.CommonParams.NAME; +import com.google.common.annotations.VisibleForTesting; import java.io.IOException; import java.lang.invoke.MethodHandles; +import java.util.Collection; import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; @@ -152,6 +154,12 @@ public boolean deleteSingleAsyncId(String asyncId) throws Exception { return asyncTaskTracker.deleteSingleAsyncId(asyncId); } + /** See {@link DistributedApiAsyncTracker#getTrackedAsyncIds()} */ + @VisibleForTesting + public Collection getTrackedAsyncIds() throws Exception { + return asyncTaskTracker.getTrackedAsyncIds(); + } + /** See {@link DistributedApiAsyncTracker#deleteAllAsyncIds()} */ public void deleteAllAsyncIds() throws Exception { asyncTaskTracker.deleteAllAsyncIds(); diff --git a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java index 6a163a96eded..7015e77c916a 100644 --- a/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java +++ b/solr/core/src/test/org/apache/solr/cloud/TestRebalanceLeaders.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.TreeMap; import java.util.TreeSet; @@ -36,6 +37,7 @@ import org.apache.solr.client.solrj.request.GenericSolrRequest; import org.apache.solr.client.solrj.response.CollectionAdminResponse; import org.apache.solr.client.solrj.response.SimpleSolrResponse; +import org.apache.solr.cloud.api.collections.DistributedCollectionConfigSetCommandRunner; import org.apache.solr.common.cloud.DocCollection; import org.apache.solr.common.cloud.Replica; import org.apache.solr.common.cloud.Slice; @@ -172,8 +174,7 @@ public void testRebalanceLeadersClearsAsyncIds() throws Exception { ZkController zkController = cluster.getRandomJetty(random()).getCoreContainer().getZkController(); - boolean overseerMode = zkController.getDistributedCommandRunner().isEmpty(); - Set idsBefore = overseerMode ? overseerAsyncIds(zkController) : Set.of(); + Set idsBefore = asyncIds(zkController); int maxWaitSeconds = 60; long start = System.nanoTime(); @@ -186,19 +187,20 @@ public void testRebalanceLeadersClearsAsyncIds() throws Exception { resp.getResponse().toString(), resp.getResponse().get("Summary").toString().contains("Success")); assertNotNull("No leader change was requested", resp.getResponse().get("successes")); - // With the Overseer disabled this is the only signal: the wait ends early only once every - // sub-command's stored response was removed. assertTrue( "Waited " + elapsedSeconds + "s, hit maxWaitSeconds", elapsedSeconds < maxWaitSeconds); - if (overseerMode) { - Set leftover = overseerAsyncIds(zkController); - leftover.removeAll(idsBefore); - assertTrue("Async ids not cleared: " + leftover, leftover.isEmpty()); - } + Set leftover = asyncIds(zkController); + leftover.removeAll(idsBefore); + assertTrue("Async ids not cleared: " + leftover, leftover.isEmpty()); } - private Set overseerAsyncIds(ZkController zkController) throws Exception { + private Set asyncIds(ZkController zkController) throws Exception { + Optional runner = + zkController.getDistributedCommandRunner(); + if (runner.isPresent()) { + return new HashSet<>(runner.get().getTrackedAsyncIds()); + } Set ids = new HashSet<>(zkController.getOverseerCompletedMap().keys()); ids.addAll(zkController.getOverseerFailureMap().keys()); ids.addAll(Overseer.getAsyncIdsMap(zkController.getZkClient()).keys());