Use applied state after DiskThresholdMonitor reroute (#94916)

Today when the `DiskThresholdMonitor` triggers a reroute, it supplies a
listener which accepts the exact cluster state that the reroute
produced. However it may be that there have been some other cluster
state updates in between the reroute computation and the completion of
this listener, so the cluster state it receives is potentially stale.
Moreover by accepting the exact cluster state that the reroute produced,
we require the `BatchedRerouteService` to retain every resulting state
for this and all other batched reroute listeners. There's no need for
this, we can instead retrieve the last-applied cluster state afresh, and
avoid this unnecessary retention of cluster states.

Relates #94914

Author: DaveCTurner

docs/changelog/94916.yaml

@@ -0,0 +1,5 @@
+pr: 94916
+summary: Use applied state after `DiskThresholdMonitor` reroute
+area: Allocation
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportUpdateDesiredNodesAction.java

@@ -161,7 +161,7 @@ public void onFailure(Exception e) {
     }
 
     private static class UpdateDesiredNodesExecutor implements ClusterStateTaskExecutor<UpdateDesiredNodesTask> {
-        private static final ActionListener<ClusterState> REROUTE_LISTENER = ActionListener.wrap(
+        private static final ActionListener<Void> REROUTE_LISTENER = ActionListener.wrap(
             r -> logger.trace("reroute after desired nodes update completed"),
             e -> logger.debug("reroute after desired nodes update failed", e)
         );

server/src/main/java/org/elasticsearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java

@@ -202,7 +202,7 @@ private void reroute(final boolean updateSettingsAcked) {
                 // completed.
                 clusterService.getRerouteService().reroute(REROUTE_TASK_SOURCE, Priority.URGENT, new ActionListener<>() {
                     @Override
-                    public void onResponse(ClusterState clusterState) {
+                    public void onResponse(Void ignored) {
                         listener.onResponse(
                             new ClusterUpdateSettingsResponse(
                                 updateSettingsAcked,

server/src/main/java/org/elasticsearch/cluster/routing/BatchedRerouteService.java

@@ -43,7 +43,7 @@ public class BatchedRerouteService implements RerouteService {
 
     private final Object mutex = new Object();
     @Nullable // null if no reroute is currently pending
-    private List<ActionListener<ClusterState>> pendingRerouteListeners;
+    private List<ActionListener<Void>> pendingRerouteListeners;
     private Priority pendingTaskPriority = Priority.LANGUID;
 
     public interface RerouteAction {
@@ -62,12 +62,12 @@ public BatchedRerouteService(ClusterService clusterService, RerouteAction rerout
      * Initiates a reroute.
      */
     @Override
-    public final void reroute(String reason, Priority priority, ActionListener<ClusterState> listener) {
-        final ActionListener<ClusterState> wrappedListener = ContextPreservingActionListener.wrapPreservingContext(
+    public final void reroute(String reason, Priority priority, ActionListener<Void> listener) {
+        final ActionListener<Void> wrappedListener = ContextPreservingActionListener.wrapPreservingContext(
             listener,
             clusterService.getClusterApplierService().threadPool().getThreadContext()
         );
-        final List<ActionListener<ClusterState>> currentListeners;
+        final List<ActionListener<Void>> currentListeners;
         synchronized (mutex) {
             if (pendingRerouteListeners != null) {
                 if (priority.sameOrAfter(pendingTaskPriority)) {
@@ -152,7 +152,7 @@ public void onFailure(Exception e) {
 
                 @Override
                 public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
-                    future.addListener(ActionListener.running(() -> ActionListener.onResponse(currentListeners, newState)));
+                    future.addListener(ActionListener.running(() -> ActionListener.onResponse(currentListeners, null)));
                 }
             });
         } catch (Exception e) {

server/src/main/java/org/elasticsearch/cluster/routing/RerouteService.java

@@ -8,7 +8,6 @@
 package org.elasticsearch.cluster.routing;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Priority;
 
 /**
@@ -23,5 +22,5 @@ public interface RerouteService {
      *                 this reroute is batched with the pending one; if there is already a pending reroute at a lower priority then
      *                 the priority of the pending batch is raised to the given priority.
      */
-    void reroute(String reason, Priority priority, ActionListener<ClusterState> listener);
+    void reroute(String reason, Priority priority, ActionListener<Void> listener);
 }

server/src/main/java/org/elasticsearch/cluster/routing/allocation/DiskThresholdMonitor.java

@@ -317,8 +317,8 @@ public void onNewInfo(ClusterInfo info) {
                 rerouteService.reroute(
                     "disk threshold monitor",
                     Priority.HIGH,
-                    ActionListener.releaseAfter(ActionListener.runAfter(ActionListener.wrap(reroutedClusterState -> {
-
+                    ActionListener.releaseAfter(ActionListener.runAfter(ActionListener.wrap(ignored -> {
+                        final var reroutedClusterState = clusterStateSupplier.get();
                         for (DiskUsage diskUsage : usagesOverHighThreshold) {
                             final RoutingNode routingNode = reroutedClusterState.getRoutingNodes().node(diskUsage.getNodeId());
                             final DiskUsage usageIncludingRelocations;

server/src/main/java/org/elasticsearch/snapshots/InternalSnapshotsInfoService.java

@@ -53,7 +53,7 @@ public class InternalSnapshotsInfoService implements ClusterStateListener, Snaps
 
     private static final Logger logger = LogManager.getLogger(InternalSnapshotsInfoService.class);
 
-    private static final ActionListener<ClusterState> REROUTE_LISTENER = ActionListener.wrap(
+    private static final ActionListener<Void> REROUTE_LISTENER = ActionListener.wrap(
         r -> logger.trace("reroute after snapshot shard size update completed"),
         e -> logger.debug("reroute after snapshot shard size update failed", e)
     );

server/src/test/java/org/elasticsearch/cluster/action/shard/ShardStartedClusterStateTaskExecutorTests.java

@@ -44,7 +44,7 @@ public class ShardStartedClusterStateTaskExecutorTests extends ESAllocationTestC
     private ShardStateAction.ShardStartedClusterStateTaskExecutor executor;
 
     @SuppressWarnings("unused")
-    private static void neverReroutes(String reason, Priority priority, ActionListener<ClusterState> listener) {
+    private static void neverReroutes(String reason, Priority priority, ActionListener<Void> listener) {
         fail("unexpectedly ran a deferred reroute");
     }
 

server/src/test/java/org/elasticsearch/cluster/routing/allocation/DiskThresholdMonitorTests.java

@@ -266,7 +266,7 @@ private void doTestDoesNotSubmitRerouteTaskTooFrequently(boolean testMaxHeadroom
             .nodes(DiscoveryNodes.builder().add(newNormalNode("node1")).add(newNormalNode("node2")))
             .build();
         AtomicLong currentTime = new AtomicLong();
-        AtomicReference<ActionListener<ClusterState>> listenerReference = new AtomicReference<>();
+        AtomicReference<ActionListener<Void>> listenerReference = new AtomicReference<>();
         DiskThresholdMonitor monitor = new DiskThresholdMonitor(
             Settings.EMPTY,
             () -> clusterState,
@@ -316,7 +316,7 @@ protected void updateIndicesReadOnly(Set<String> indicesToMarkReadOnly, Releasab
         currentTime.addAndGet(randomLongBetween(0, 120000));
         monitor.onNewInfo(clusterInfo(allDisksOk));
         assertNotNull(listenerReference.get());
-        listenerReference.getAndSet(null).onResponse(clusterState);
+        listenerReference.getAndSet(null).onResponse(null);
 
         // should not reroute when all disks are ok and no new info received
         currentTime.addAndGet(randomLongBetween(0, 120000));
@@ -327,7 +327,7 @@ protected void updateIndicesReadOnly(Set<String> indicesToMarkReadOnly, Releasab
         if (randomBoolean()) {
             currentTime.addAndGet(randomLongBetween(0, 120000));
             monitor.onNewInfo(clusterInfo(oneDiskAboveWatermark));
-            Optional.ofNullable(listenerReference.getAndSet(null)).ifPresent(l -> l.onResponse(clusterState));
+            Optional.ofNullable(listenerReference.getAndSet(null)).ifPresent(l -> l.onResponse(null));
         }
 
         // however once the reroute interval has elapsed then we must reroute again
@@ -339,7 +339,7 @@ protected void updateIndicesReadOnly(Set<String> indicesToMarkReadOnly, Releasab
         );
         monitor.onNewInfo(clusterInfo(oneDiskAboveWatermark));
         assertNotNull(listenerReference.get());
-        listenerReference.getAndSet(null).onResponse(clusterState);
+        listenerReference.getAndSet(null).onResponse(null);
 
         if (randomBoolean()) {
             // should not re-route again within the reroute interval
@@ -362,15 +362,15 @@ protected void updateIndicesReadOnly(Set<String> indicesToMarkReadOnly, Releasab
         );
         monitor.onNewInfo(clusterInfo(oneDiskAboveWatermark));
         assertNotNull(listenerReference.get());
-        final ActionListener<ClusterState> rerouteListener1 = listenerReference.getAndSet(null);
+        final ActionListener<Void> rerouteListener1 = listenerReference.getAndSet(null);
 
         // should not re-route again before reroute has completed
         currentTime.addAndGet(randomLongBetween(0, 120000));
         monitor.onNewInfo(clusterInfo(allDisksOk));
         assertNull(listenerReference.get());
 
         // complete reroute
-        rerouteListener1.onResponse(clusterState);
+        rerouteListener1.onResponse(null);
 
         if (randomBoolean()) {
             // should not re-route again within the reroute interval
@@ -475,7 +475,7 @@ private void doTestAutoReleaseIndices(boolean testMaxHeadroom) {
             (reason, priority, listener) -> {
                 assertNotNull(listener);
                 assertThat(priority, equalTo(Priority.HIGH));
-                listener.onResponse(clusterState);
+                listener.onResponse(null);
             }
         ) {
             @Override
@@ -567,7 +567,7 @@ protected void updateIndicesReadOnly(Set<String> indicesToUpdate, Releasable onC
             (reason, priority, listener) -> {
                 assertNotNull(listener);
                 assertThat(priority, equalTo(Priority.HIGH));
-                listener.onResponse(clusterStateWithBlocks);
+                listener.onResponse(null);
             }
         ) {
             @Override
@@ -813,7 +813,7 @@ private void doTestNoAutoReleaseOfIndicesOnReplacementNodes(boolean testMaxHeadr
             (reason, priority, listener) -> {
                 assertNotNull(listener);
                 assertThat(priority, equalTo(Priority.HIGH));
-                listener.onResponse(currentClusterState.get());
+                listener.onResponse(null);
             }
         ) {
             @Override
@@ -1058,7 +1058,7 @@ public long getAsLong() {
             new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
             null,
             timeSupplier,
-            (reason, priority, listener) -> listener.onResponse(clusterStateRef.get())
+            (reason, priority, listener) -> listener.onResponse(null)
         ) {
             @Override
             protected void updateIndicesReadOnly(Set<String> indicesToMarkReadOnly, Releasable onCompletion, boolean readOnly) {

server/src/test/java/org/elasticsearch/snapshots/InternalSnapshotsInfoServiceTests.java

@@ -91,7 +91,7 @@ public void setUp() throws Exception {
         threadPool = new TestThreadPool(getTestName());
         clusterService = ClusterServiceUtils.createClusterService(threadPool);
         repositoriesService = mock(RepositoriesService.class);
-        rerouteService = (reason, priority, listener) -> listener.onResponse(clusterService.state());
+        rerouteService = (reason, priority, listener) -> listener.onResponse(null);
     }
 
     @After
@@ -109,7 +109,7 @@ public void testSnapshotShardSizes() throws Exception {
         final int numberOfShards = randomIntBetween(1, 50);
         final CountDownLatch rerouteLatch = new CountDownLatch(numberOfShards);
         final RerouteService rerouteService = (reason, priority, listener) -> {
-            listener.onResponse(clusterService.state());
+            listener.onResponse(null);
             assertThat(rerouteLatch.getCount(), greaterThanOrEqualTo(0L));
             rerouteLatch.countDown();
         };
@@ -181,7 +181,7 @@ public void testErroneousSnapshotShardSizes() throws Exception {
         final CountDown reroutes = new CountDown(maxShardsToCreate);
         final RerouteService rerouteService = (reason, priority, listener) -> {
             try {
-                listener.onResponse(clusterService.state());
+                listener.onResponse(null);
             } finally {
                 if (reroutes.countDown()) {
                     waitForAllReroutesProcessed.onResponse(null);