Preemptively initialize routing nodes and indices lookup on all node types (#89032)

Follow up to #89005 running the initialization as soon as possible on non-master
nodes as well.

docs/changelog/89032.yaml

@@ -0,0 +1,5 @@
+pr: 89032
+summary: Preemptively initialize routing nodes and indices lookup on all node types
+area: Cluster Coordination
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -54,6 +54,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.Executor;
 import java.util.function.Consumer;
 
 import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
@@ -351,6 +352,19 @@ public class ClusterState implements ToXContentFragment, Diffable<ClusterState>
         return RoutingNodes.mutable(routingTable, this.nodes);
     }
 
+    /**
+     * Initialize data structures that lazy computed for this instance in the background by using the giving executor.
+     * @param executor executor to run initialization tasks on
+     */
+    public void initializeAsync(Executor executor) {
+        if (routingNodes == null) {
+            executor.execute(this::getRoutingNodes);
+        }
+        if (metadata.indicesLookupInitialized() == false) {
+            executor.execute(metadata::getIndicesLookup);
+        }
+    }
+
     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -414,8 +414,14 @@ public class Coordinator extends AbstractLifecycleComponent implements ClusterSt
         assert publishRequest.getAcceptedState().nodes().getLocalNode().equals(getLocalNode())
             : publishRequest.getAcceptedState().nodes().getLocalNode() + " != " + getLocalNode();
 
+        final ClusterState newClusterState = publishRequest.getAcceptedState();
+        if (newClusterState.nodes().isLocalNodeElectedMaster() == false) {
+            // background initialization on the current master has been started by the master service already
+            newClusterState.initializeAsync(transportService.getThreadPool().generic());
+        }
+
         synchronized (mutex) {
-            final DiscoveryNode sourceNode = publishRequest.getAcceptedState().nodes().getMasterNode();
+            final DiscoveryNode sourceNode = newClusterState.nodes().getMasterNode();
             logger.trace("handlePublishRequest: handling [{}] from [{}]", publishRequest, sourceNode);
 
             if (sourceNode.equals(getLocalNode()) && mode != Mode.LEADER) {
@@ -427,30 +433,30 @@ public class Coordinator extends AbstractLifecycleComponent implements ClusterSt
             final ClusterState localState = coordinationState.get().getLastAcceptedState();
 
             if (localState.metadata().clusterUUIDCommitted()
-                && localState.metadata().clusterUUID().equals(publishRequest.getAcceptedState().metadata().clusterUUID()) == false) {
+                && localState.metadata().clusterUUID().equals(newClusterState.metadata().clusterUUID()) == false) {
                 logger.warn(
                     "received cluster state from {} with a different cluster uuid {} than local cluster uuid {}, rejecting",
                     sourceNode,
-                    publishRequest.getAcceptedState().metadata().clusterUUID(),
+                    newClusterState.metadata().clusterUUID(),
                     localState.metadata().clusterUUID()
                 );
                 throw new CoordinationStateRejectedException(
                     "received cluster state from "
                         + sourceNode
                         + " with a different cluster uuid "
-                        + publishRequest.getAcceptedState().metadata().clusterUUID()
+                        + newClusterState.metadata().clusterUUID()
                         + " than local cluster uuid "
                         + localState.metadata().clusterUUID()
                         + ", rejecting"
                 );
             }
 
-            if (publishRequest.getAcceptedState().term() > localState.term()) {
+            if (newClusterState.term() > localState.term()) {
                 // only do join validation if we have not accepted state from this master yet
-                onJoinValidators.forEach(a -> a.accept(getLocalNode(), publishRequest.getAcceptedState()));
+                onJoinValidators.forEach(a -> a.accept(getLocalNode(), newClusterState));
             }
 
-            ensureTermAtLeast(sourceNode, publishRequest.getAcceptedState().term());
+            ensureTermAtLeast(sourceNode, newClusterState.term());
             final PublishResponse publishResponse = coordinationState.get().handlePublishRequest(publishRequest);
 
             if (sourceNode.equals(getLocalNode())) {
@@ -459,10 +465,7 @@ public class Coordinator extends AbstractLifecycleComponent implements ClusterSt
                 becomeFollower("handlePublishRequest", sourceNode); // also updates preVoteCollector
             }
 
-            return new PublishWithJoinResponse(
-                publishResponse,
-                joinWithDestination(lastJoin, sourceNode, publishRequest.getAcceptedState().term())
-            );
+            return new PublishWithJoinResponse(publishResponse, joinWithDestination(lastJoin, sourceNode, newClusterState.term()));
         }
     }
 

server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java

@@ -509,6 +509,10 @@ public class Metadata extends AbstractCollection<IndexMetadata> implements Diffa
         return true;
     }
 
+    public boolean indicesLookupInitialized() {
+        return indicesLookup != null;
+    }
+
     public SortedMap<String, IndexAbstraction> getIndicesLookup() {
         SortedMap<String, IndexAbstraction> lookup = indicesLookup;
         if (lookup == null) {

server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

@@ -332,8 +332,7 @@ public class MasterService extends AbstractLifecycleComponent {
                     logger.debug("publishing cluster state version [{}]", newClusterState.version());
                     // initialize routing nodes and the indices lookup concurrently, we will need both of them for the cluster state
                     // application and can compute them while we wait for the other nodes during publication
-                    threadPool.generic().execute(newClusterState::getRoutingNodes);
-                    threadPool.generic().execute(newClusterState.metadata()::getIndicesLookup);
+                    newClusterState.initializeAsync(threadPool.generic());
                     publish(
                         clusterStatePublicationEvent,
                         new CompositeTaskAckListener(

test/framework/src/test/java/org/elasticsearch/cluster/service/FakeThreadPoolMasterServiceTests.java

@@ -142,8 +142,7 @@ public class FakeThreadPoolMasterServiceTests extends ESTestCase {
 
         runnableTasks.remove(0).run(); // schedule again
 
-        // run tasks for computing routing nodes and indices lookup
-        runnableTasks.remove(0).run();
+        // run task for computing missing indices lookup
         runnableTasks.remove(0).run();
 
         runnableTasks.remove(0).run(); // publish again