Remove test dependencies on cluster state API master waiting (#129118)

As preparation for running the cluster state API on the local node, we
need to update these tests that currently depend on that API running on
(and waiting for) the master node.

Relates #127212

plugins/discovery-azure-classic/src/internalClusterTest/java/org/elasticsearch/discovery/azure/classic/AzureSimpleTests.java

@@ -13,7 +13,6 @@ import org.elasticsearch.cloud.azure.classic.AbstractAzureComputeServiceTestCase
 import org.elasticsearch.cloud.azure.classic.management.AzureComputeService.Discovery;
 import org.elasticsearch.cloud.azure.classic.management.AzureComputeService.Management;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESIntegTestCase;
 
 import static org.hamcrest.Matchers.containsString;
@@ -28,16 +27,7 @@ public class AzureSimpleTests extends AbstractAzureComputeServiceTestCase {
 
         final String node1 = internalCluster().startNode(settings);
         registerAzureNode(node1);
-        assertNotNull(
+        awaitMasterNode();
-            client().admin()
-                .cluster()
-                .prepareState(TEST_REQUEST_TIMEOUT)
-                .setMasterNodeTimeout(TimeValue.timeValueSeconds(1))
-                .get()
-                .getState()
-                .nodes()
-                .getMasterNodeId()
-        );
 
         // We expect having 1 node as part of the cluster, let's test that
         assertNumberOfNodes(1);
@@ -50,16 +40,7 @@ public class AzureSimpleTests extends AbstractAzureComputeServiceTestCase {
 
         final String node1 = internalCluster().startNode(settings);
         registerAzureNode(node1);
-        assertNotNull(
+        awaitMasterNode();
-            client().admin()
-                .cluster()
-                .prepareState(TEST_REQUEST_TIMEOUT)
-                .setMasterNodeTimeout(TimeValue.timeValueSeconds(1))
-                .get()
-                .getState()
-                .nodes()
-                .getMasterNodeId()
-        );
 
         // We expect having 1 node as part of the cluster, let's test that
         assertNumberOfNodes(1);

plugins/discovery-azure-classic/src/internalClusterTest/java/org/elasticsearch/discovery/azure/classic/AzureTwoStartedNodesTests.java

@@ -13,7 +13,6 @@ import org.elasticsearch.cloud.azure.classic.AbstractAzureComputeServiceTestCase
 import org.elasticsearch.cloud.azure.classic.management.AzureComputeService.Discovery;
 import org.elasticsearch.cloud.azure.classic.management.AzureComputeService.Management;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESIntegTestCase;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 0)
@@ -31,30 +30,12 @@ public class AzureTwoStartedNodesTests extends AbstractAzureComputeServiceTestCa
         logger.info("--> start first node");
         final String node1 = internalCluster().startNode(settings);
         registerAzureNode(node1);
-        assertNotNull(
+        awaitMasterNode();
-            client().admin()
-                .cluster()
-                .prepareState(TEST_REQUEST_TIMEOUT)
-                .setMasterNodeTimeout(TimeValue.timeValueSeconds(1))
-                .get()
-                .getState()
-                .nodes()
-                .getMasterNodeId()
-        );
 
         logger.info("--> start another node");
         final String node2 = internalCluster().startNode(settings);
         registerAzureNode(node2);
-        assertNotNull(
+        awaitMasterNode();
-            client().admin()
-                .cluster()
-                .prepareState(TEST_REQUEST_TIMEOUT)
-                .setMasterNodeTimeout(TimeValue.timeValueSeconds(1))
-                .get()
-                .getState()
-                .nodes()
-                .getMasterNodeId()
-        );
 
         // We expect having 2 nodes as part of the cluster, let's test that
         assertNumberOfNodes(2);

server/src/internalClusterTest/java/org/elasticsearch/discovery/DiscoveryDisruptionIT.java

@@ -146,6 +146,7 @@ public class DiscoveryDisruptionIT extends AbstractDisruptionTestCase {
 
         isolateAllNodes.stopDisrupting();
 
+        awaitMasterNode();
         final ClusterState state = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState();
         if (state.metadata().getProject().hasIndex("test") == false) {
             fail("index 'test' was lost. current cluster state: " + state);

server/src/internalClusterTest/java/org/elasticsearch/readiness/ReadinessClusterIT.java

@@ -8,7 +8,6 @@
  */
 package org.elasticsearch.readiness;
 
-import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
@@ -18,9 +17,7 @@ import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Strings;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
-import org.elasticsearch.discovery.MasterNotDiscoveredException;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.reservedstate.service.FileSettingsService;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -48,7 +45,6 @@ import static org.elasticsearch.test.NodeRoles.masterNode;
 import static org.elasticsearch.test.NodeRoles.nonDataNode;
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.notNullValue;
 
@@ -110,20 +106,6 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         return Collections.unmodifiableList(plugins);
     }
 
-    private void assertMasterNode(Client client, String node) {
-        assertThat(
-            client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNode().getName(),
-            equalTo(node)
-        );
-    }
-
-    private void expectMasterNotFound() {
-        expectThrows(
-            MasterNotDiscoveredException.class,
-            clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).setMasterNodeTimeout(TimeValue.timeValueMillis(100))
-        );
-    }
-
     @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/108613")
     public void testReadinessDuringRestarts() throws Exception {
         internalCluster().setBootstrapMasterNodeIndex(0);
@@ -131,23 +113,23 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         logger.info("--> start data node / non master node");
         String dataNode = internalCluster().startNode(Settings.builder().put(dataOnlyNode()).put("discovery.initial_state_timeout", "1s"));
 
-        expectMasterNotFound();
+        awaitMasterNotFound();
         assertFalse(internalCluster().getInstance(ReadinessService.class, dataNode).ready());
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
 
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         tcpReadinessProbeTrue(internalCluster().getInstance(ReadinessService.class, dataNode));
         tcpReadinessProbeTrue(internalCluster().getInstance(ReadinessService.class, masterNode));
 
         final var masterReadinessService = internalCluster().getInstance(ReadinessService.class, masterNode);
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
 
         logger.info("--> stop master node");
         Settings masterDataPathSettings = internalCluster().dataPathSettings(internalCluster().getMasterName());
         internalCluster().stopCurrentMasterNode();
-        expectMasterNotFound();
+        awaitMasterNotFound();
 
         tcpReadinessProbeFalse(masterReadinessService);
 
@@ -156,8 +138,8 @@ public class ReadinessClusterIT extends ESIntegTestCase {
             Settings.builder().put(nonDataNode(masterNode())).put(masterDataPathSettings)
         );
 
-        assertMasterNode(internalCluster().nonMasterClient(), nextMasterEligibleNodeName);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), nextMasterEligibleNodeName);
-        assertMasterNode(internalCluster().masterClient(), nextMasterEligibleNodeName);
+        awaitMasterNode(internalCluster().getMasterName(), nextMasterEligibleNodeName);
         tcpReadinessProbeTrue(internalCluster().getInstance(ReadinessService.class, nextMasterEligibleNodeName));
     }
 
@@ -168,7 +150,7 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         String masterNode = internalCluster().startMasterOnlyNode();
         internalCluster().validateClusterFormed();
 
-        assertMasterNode(internalCluster().masterClient(), masterNode);
+        awaitMasterNode(internalCluster().getMasterName(), masterNode);
 
         logger.info("--> start 2 data nodes");
         List<String> dataNodes = internalCluster().startDataOnlyNodes(2);
@@ -196,7 +178,7 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         internalCluster().restartNode(masterNode, new InternalTestCluster.RestartCallback() {
             @Override
             public Settings onNodeStopped(String nodeName) throws Exception {
-                expectMasterNotFound();
+                awaitMasterNotFound();
 
                 logger.info("--> master node [{}] stopped", nodeName);
 
@@ -271,7 +253,7 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         final String masterNode = internalCluster().startMasterOnlyNode(
             Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
         );
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
@@ -294,14 +276,14 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         String dataNode = internalCluster().startNode(Settings.builder().put(dataOnlyNode()).put("discovery.initial_state_timeout", "1s"));
         String masterNode = internalCluster().startMasterOnlyNode();
 
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         assertBusy(() -> assertTrue("master node ready", internalCluster().getInstance(ReadinessService.class, masterNode).ready()));
         assertBusy(() -> assertTrue("data node ready", internalCluster().getInstance(ReadinessService.class, dataNode).ready()));
 
         logger.info("--> stop master node");
         Settings masterDataPathSettings = internalCluster().dataPathSettings(internalCluster().getMasterName());
         internalCluster().stopCurrentMasterNode();
-        expectMasterNotFound();
+        awaitMasterNotFound();
 
         logger.info("--> write bad file settings before restarting master node");
         writeFileSettings(testErrorJSON);
@@ -309,7 +291,7 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         logger.info("--> restart master node");
         String nextMasterNode = internalCluster().startNode(Settings.builder().put(nonDataNode(masterNode())).put(masterDataPathSettings));
 
-        assertMasterNode(internalCluster().nonMasterClient(), nextMasterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), nextMasterNode);
 
         var savedClusterState = setupClusterStateListenerForError(nextMasterNode);
         assertTrue(savedClusterState.await(20, TimeUnit.SECONDS));
@@ -324,7 +306,7 @@ public class ReadinessClusterIT extends ESIntegTestCase {
         final String masterNode = internalCluster().startMasterOnlyNode(
             Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
         );
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListener(masterNode);
 
         // we need this after we setup the listener above, in case the node started and processed
@@ -382,7 +364,7 @@ public class ReadinessClusterIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var readinessProbeListening = setupReadinessProbeListener(masterNode);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);

server/src/internalClusterTest/java/org/elasticsearch/reservedstate/service/ComponentTemplatesFileSettingsIT.java

@@ -17,7 +17,6 @@ import org.elasticsearch.action.admin.indices.template.get.GetComposableIndexTem
 import org.elasticsearch.action.admin.indices.template.put.PutComponentTemplateAction;
 import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
 import org.elasticsearch.action.admin.indices.template.reservedstate.ReservedComposableIndexTemplateAction;
-import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.metadata.ComponentTemplate;
@@ -35,7 +34,6 @@ import java.io.ByteArrayInputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -46,7 +44,6 @@ import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.anEmptyMap;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.notNullValue;
 
@@ -355,13 +352,6 @@ public class ComponentTemplatesFileSettingsIT extends ESIntegTestCase {
              }
         }""";
 
-    private void assertMasterNode(Client client, String node) throws ExecutionException, InterruptedException {
-        assertThat(
-            client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).execute().get().getState().nodes().getMasterNode().getName(),
-            equalTo(node)
-        );
-    }
-
     private void writeJSONFile(String node, String json) throws Exception {
         FileSettingsServiceIT.writeJSONFile(node, json, logger, versionCounter.incrementAndGet());
     }
@@ -543,7 +533,7 @@ public class ComponentTemplatesFileSettingsIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
 
         assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2());
 
@@ -611,7 +601,7 @@ public class ComponentTemplatesFileSettingsIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         writeJSONFile(masterNode, testErrorJSON);

server/src/internalClusterTest/java/org/elasticsearch/reservedstate/service/FileSettingsServiceIT.java

@@ -14,7 +14,6 @@ import org.apache.lucene.tests.util.LuceneTestCase;
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
-import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.metadata.ReservedStateErrorMetadata;
@@ -129,13 +128,6 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
              }
         }""";
 
-    private void assertMasterNode(Client client, String node) {
-        assertThat(
-            client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNode().getName(),
-            equalTo(node)
-        );
-    }
-
     public static void writeJSONFile(String node, String json, Logger logger, Long version) throws Exception {
         FileSettingsService fileSettingsService = internalCluster().getInstance(FileSettingsService.class, node);
         writeJSONFile(node, json, logger, version, fileSettingsService.watchedFile());
@@ -254,7 +246,7 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListener(masterNode, versionCounter.incrementAndGet());
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
@@ -281,7 +273,7 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
 
@@ -297,7 +289,7 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
         final String masterNode = internalCluster().startMasterOnlyNode(
             Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
         );
-        assertMasterNode(internalCluster().masterClient(), masterNode);
+        awaitMasterNode(internalCluster().getMasterName(), masterNode);
         var savedClusterState = setupClusterStateListener(masterNode, versionCounter.incrementAndGet());
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
@@ -375,7 +367,7 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
         final String masterNode = internalCluster().startMasterOnlyNode(
             Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
         );
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
@@ -399,7 +391,7 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
         final String masterNode = internalCluster().startMasterOnlyNode(
             Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
         );
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);
@@ -435,7 +427,7 @@ public class FileSettingsServiceIT extends ESIntegTestCase {
         final String masterNode = internalCluster().startMasterOnlyNode(
             Settings.builder().put(INITIAL_STATE_TIMEOUT_SETTING.getKey(), "0s").build()
         );
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);

server/src/internalClusterTest/java/org/elasticsearch/reservedstate/service/RepositoriesFileSettingsIT.java

@@ -16,7 +16,6 @@ import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryRequ
 import org.elasticsearch.action.admin.cluster.repositories.put.TransportPutRepositoryAction;
 import org.elasticsearch.action.admin.cluster.repositories.reservedstate.ReservedRepositoryAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
-import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
@@ -33,7 +32,6 @@ import org.elasticsearch.xcontent.XContentParserConfiguration;
 import java.io.ByteArrayInputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
@@ -43,7 +41,6 @@ import static org.elasticsearch.xcontent.XContentType.JSON;
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.notNullValue;
 
@@ -94,13 +91,6 @@ public class RepositoriesFileSettingsIT extends ESIntegTestCase {
              }
         }""";
 
-    private void assertMasterNode(Client client, String node) throws ExecutionException, InterruptedException {
-        assertThat(
-            client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).execute().get().getState().nodes().getMasterNode().getName(),
-            equalTo(node)
-        );
-    }
-
     private void writeJSONFile(String node, String json) throws Exception {
         FileSettingsServiceIT.writeJSONFile(node, json, logger, versionCounter.incrementAndGet());
     }
@@ -164,7 +154,7 @@ public class RepositoriesFileSettingsIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
 
         assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2());
     }
@@ -220,7 +210,7 @@ public class RepositoriesFileSettingsIT extends ESIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         writeJSONFile(masterNode, testErrorJSON);

test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java

@@ -962,6 +962,16 @@ public abstract class ESIntegTestCase extends ESTestCase {
         safeAwait(listener, TEST_REQUEST_TIMEOUT);
     }
 
+    /**
+     * Waits for all nodes in the cluster to have a consistent view of which node is currently the master.
+     */
+    public void awaitMasterNode() {
+        // The cluster health API always runs on the master node, and the master only completes cluster state publication when all nodes
+        // in the cluster have accepted the new cluster state. By waiting for all events to have finished on the master node, we ensure
+        // that the whole cluster has a consistent view of which node is the master.
+        clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT).setTimeout(TEST_REQUEST_TIMEOUT).setWaitForEvents(Priority.LANGUID).get();
+    }
+
     /**
      * Waits for a random node in the cluster to not see a master node in the cluster state.
      * Note that this does not guarantee that all other nodes in the cluster are on the same cluster state version already.
@@ -1360,8 +1370,14 @@ public abstract class ESIntegTestCase extends ESTestCase {
     }
 
     protected final void doEnsureClusterStateConsistency(NamedWriteableRegistry namedWriteableRegistry) {
+        // This check has very little value in external test clusters and there is no guaranteed method of obtaining the master cluster
+        // state in those clusters.
+        if (isInternalCluster() == false) {
+            return;
+        }
         final PlainActionFuture<Void> future = new PlainActionFuture<>();
         final List<SubscribableListener<ClusterStateResponse>> localStates = new ArrayList<>(cluster().size());
+        final var masterName = internalCluster().getMasterName();
         for (Client client : cluster().getClients()) {
             localStates.add(
                 SubscribableListener.newForked(
@@ -1371,7 +1387,7 @@ public abstract class ESIntegTestCase extends ESTestCase {
         }
         try (RefCountingListener refCountingListener = new RefCountingListener(future)) {
             SubscribableListener.<ClusterStateResponse>newForked(
-                l -> client().admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).all().execute(l)
+                l -> client(masterName).admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).all().execute(l)
             ).andThenAccept(masterStateResponse -> {
                 byte[] masterClusterStateBytes = ClusterState.Builder.toBytes(masterStateResponse.getState());
                 // remove local node reference

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlNativeIntegTestCase.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.ml.integration;
 
 import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateAction;
 import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
@@ -55,6 +56,7 @@ import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ExternalTestCluster;
 import org.elasticsearch.test.SecuritySettingsSourceField;
 import org.elasticsearch.test.TestCluster;
+import org.elasticsearch.test.XContentTestUtils;
 import org.elasticsearch.transport.netty4.Netty4Plugin;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.autoscaling.Autoscaling;
@@ -103,6 +105,7 @@ import org.elasticsearch.xpack.ml.autoscaling.MlScalingReason;
 import org.elasticsearch.xpack.slm.SnapshotLifecycle;
 import org.elasticsearch.xpack.slm.history.SnapshotLifecycleTemplateRegistry;
 import org.elasticsearch.xpack.transform.Transform;
+import org.junit.After;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -343,8 +346,11 @@ abstract class MlNativeIntegTestCase extends ESIntegTestCase {
         return messages;
     }
 
-    @Override
+    /**
-    protected void ensureClusterStateConsistency() throws IOException {
+     * Asserts that all ML named writeables pass a cluster state round-trip (de)serialization.
+     */
+    @After
+    protected void assertClusterRoundTrip() throws IOException {
         if (cluster() != null && cluster().size() > 0) {
             List<NamedWriteableRegistry.Entry> entries = new ArrayList<>(ClusterModule.getNamedWriteables());
             entries.addAll(new SearchModule(Settings.EMPTY, Collections.emptyList()).getNamedWriteables());
@@ -431,7 +437,29 @@ abstract class MlNativeIntegTestCase extends ESIntegTestCase {
             );
             entries.add(new NamedWriteableRegistry.Entry(NamedDiff.class, ModelRegistryMetadata.TYPE, ModelRegistryMetadata::readDiffFrom));
 
-            doEnsureClusterStateConsistency(new NamedWriteableRegistry(entries));
+            // Retrieve the cluster state from a random node, and serialize and deserialize it.
+            final ClusterStateResponse clusterStateResponse = client().admin()
+                .cluster()
+                .prepareState(TEST_REQUEST_TIMEOUT)
+                .all()
+                .get(TEST_REQUEST_TIMEOUT);
+            byte[] clusterStateBytes = ClusterState.Builder.toBytes(clusterStateResponse.getState());
+            final ClusterState parsedClusterState = ClusterState.Builder.fromBytes(
+                clusterStateBytes,
+                clusterStateResponse.getState().nodes().getLocalNode(),
+                new NamedWriteableRegistry(entries)
+            );
+            final var responseMap = XContentTestUtils.convertToMap(clusterStateResponse.getState());
+            final var parsedMap = XContentTestUtils.convertToMap(parsedClusterState);
+            final var diff = XContentTestUtils.differenceBetweenMapsIgnoringArrayOrder(responseMap, parsedMap);
+            if (diff != null) {
+                logger.error(
+                    "Cluster state response:\n{}\nParsed cluster state:\n{}",
+                    clusterStateResponse.getState().toString(),
+                    parsedClusterState.toString()
+                );
+                assertNull("cluster state JSON serialization does not match", diff);
+            }
         }
     }
 

x-pack/plugin/slm/src/internalClusterTest/java/org/elasticsearch/xpack/slm/SLMFileSettingsIT.java

@@ -12,7 +12,6 @@ import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequ
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
-import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.metadata.ReservedStateErrorMetadata;
@@ -149,13 +148,6 @@ public class SLMFileSettingsIT extends AbstractSnapshotIntegTestCase {
         return Settings.builder().put(LifecycleSettings.SLM_HISTORY_INDEX_ENABLED, false).build();
     }
 
-    private void assertMasterNode(Client client, String node) {
-        assertThat(
-            client.admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNode().getName(),
-            equalTo(node)
-        );
-    }
-
     private void writeJSONFile(String node, String json) throws Exception {
         long version = versionCounter.incrementAndGet();
 
@@ -238,7 +230,7 @@ public class SLMFileSettingsIT extends AbstractSnapshotIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
 
         assertClusterStateSaveOK(savedClusterState.v1(), savedClusterState.v2());
 
@@ -352,7 +344,7 @@ public class SLMFileSettingsIT extends AbstractSnapshotIntegTestCase {
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
-        assertMasterNode(internalCluster().nonMasterClient(), masterNode);
+        awaitMasterNode(internalCluster().getMasterName(), masterNode);
         var savedClusterState = setupClusterStateListenerForError(masterNode);
 
         writeJSONFile(masterNode, testErrorJSON);

x-pack/plugin/voting-only-node/src/internalClusterTest/java/org/elasticsearch/cluster/coordination/votingonly/VotingOnlyNodePluginTests.java

@@ -15,12 +15,9 @@ import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.blobstore.BlobStore;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
-import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.discovery.MasterNotDiscoveredException;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.indices.recovery.RecoverySettings;
@@ -53,7 +50,6 @@ import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.nullValue;
 
 @ESIntegTestCase.ClusterScope(scope = Scope.TEST, numDataNodes = 0, autoManageMasterNodes = false)
 public class VotingOnlyNodePluginTests extends ESIntegTestCase {
@@ -110,7 +106,7 @@ public class VotingOnlyNodePluginTests extends ESIntegTestCase {
         final String originalMaster = internalCluster().getMasterName();
 
         internalCluster().stopCurrentMasterNode();
-        clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT).setWaitForEvents(Priority.LANGUID).get();
+        awaitMasterNode();
         assertNotEquals(originalMaster, internalCluster().getMasterName());
         assertThat(
             VotingOnlyNodePlugin.isVotingOnlyNode(
@@ -129,6 +125,7 @@ public class VotingOnlyNodePluginTests extends ESIntegTestCase {
                 equalTo(3)
             )
         );
+        awaitMasterNode();
         assertThat(
             VotingOnlyNodePlugin.isVotingOnlyNode(
                 clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNode()
@@ -146,7 +143,8 @@ public class VotingOnlyNodePluginTests extends ESIntegTestCase {
                 .build()
         );
         internalCluster().startNode();
-        assertBusy(() -> assertThat(clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState().getNodes().getSize(), equalTo(2)));
+        ensureStableCluster(2);
+        awaitMasterNode();
         assertThat(
             VotingOnlyNodePlugin.isVotingOnlyNode(
                 clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNode()
@@ -166,22 +164,11 @@ public class VotingOnlyNodePluginTests extends ESIntegTestCase {
                 equalTo(3)
             )
         );
+        awaitMasterNode();
         final String oldMasterId = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState().nodes().getMasterNodeId();
 
         internalCluster().stopCurrentMasterNode();
-
+        awaitMasterNotFound();
-        expectThrows(
-            MasterNotDiscoveredException.class,
-            () -> assertThat(
-                clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT)
-                    .setMasterNodeTimeout(TimeValue.timeValueMillis(100))
-                    .get()
-                    .getState()
-                    .nodes()
-                    .getMasterNodeId(),
-                nullValue()
-            )
-        );
 
         // start a fresh full master node, which will be brought into the cluster as master by the voting-only nodes
         final String newMaster = internalCluster().startNode();