Fix NullPointerExceptions in TransportClusterStateActionDisruptionIT

Replaces the use of InternalTestCluster.getMasterName() with a
ClusterServiceUtils.addTemporaryStateListener() call that waits
for a new master node other than the previous master node.
InternalTestCluster.getMasterName() is not safe to use in
unstable clusters, per PR 127213.

Closes:
#127466
#127443
#127424
#127423
#127422

Author: JeremyDahlgren

muted-tests.yml

@@ -432,18 +432,9 @@ tests:
 - class: org.elasticsearch.xpack.esql.heap_attack.HeapAttackIT
   method: testLookupExplosionNoFetch
   issue: https://github.com/elastic/elasticsearch/issues/127365
-- class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
-  method: testNonLocalRequestAlwaysFindsMasterAndWaitsForMetadata
-  issue: https://github.com/elastic/elasticsearch/issues/127422
 - class: org.elasticsearch.xpack.esql.qa.single_node.PushQueriesIT
   method: testPushCaseInsensitiveEqualityOnDefaults
   issue: https://github.com/elastic/elasticsearch/issues/127431
-- class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
-  method: testLocalRequestAlwaysSucceeds
-  issue: https://github.com/elastic/elasticsearch/issues/127423
-- class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionDisruptionIT
-  method: testLocalRequestWaitsForMetadata
-  issue: https://github.com/elastic/elasticsearch/issues/127466
 
 # Examples:
 #

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateActionDisruptionIT.java

@@ -21,24 +21,24 @@
 import org.elasticsearch.discovery.MasterNotDiscoveredException;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.not;
 
 @ESIntegTestCase.ClusterScope(numDataNodes = 0, scope = ESIntegTestCase.Scope.TEST)
 public class TransportClusterStateActionDisruptionIT extends ESIntegTestCase {
@@ -212,11 +212,13 @@ public void runRepeatedlyWhileChangingMaster(Runnable runnable) throws Exception
             }
         }
 
-        assertBusy(() -> {
-            final String nonMasterNode = randomValueOtherThan(masterName, () -> randomFrom(internalCluster().getNodeNames()));
-            final String claimedMasterName = internalCluster().getMasterName(nonMasterNode);
-            assertThat(claimedMasterName, not(equalTo(masterName)));
-        });
+        final String nonMasterNode = randomValueOtherThan(masterName, () -> randomFrom(internalCluster().getNodeNames()));
+        var newMasterNodeListener = ClusterServiceUtils.addTemporaryStateListener(
+            internalCluster().clusterService(nonMasterNode),
+            state -> Optional.ofNullable(state.nodes().getMasterNode()).map(m -> m.getName().equals(masterName) == false).orElse(false),
+            TEST_REQUEST_TIMEOUT
+        );
+        safeAwait(newMasterNodeListener, TEST_REQUEST_TIMEOUT);
 
         shutdown.set(true);
         assertingThread.join();