Fix NullPointerExceptions failing TransportClusterStateActionDisruptionIT (elastic#127523)

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 elastic#127466
Closes elastic#127443
Closes elastic#127424
Closes elastic#127423
Closes elastic#127422

Author: JeremyDahlgren

muted-tests.yml

@@ -429,18 +429,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
 - class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeIT
   method: test
   issue: https://github.com/elastic/elasticsearch/issues/127536

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

@@ -28,17 +28,16 @@
 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 +211,12 @@ 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()));
+        awaitClusterState(
+            logger,
+            nonMasterNode,
+            state -> Optional.ofNullable(state.nodes().getMasterNode()).map(m -> m.getName().equals(masterName) == false).orElse(false)
+        );
 
         shutdown.set(true);
         assertingThread.join();