Clean up awaitClusterState overloads (#133550)

The `logger` parameter is unused, it doesn't throw any checked
exceptions, and there's no need for `protected` instance methods when
they're also available as `public static`.

Backport of #132529 to 9.1

Author: DaveCTurner

qa/smoke-test-http/src/internalClusterTest/java/org/elasticsearch/http/snapshots/RestGetSnapshotsIT.java

@@ -204,7 +204,7 @@ public void testSortAndPaginateWithInProgress() throws Exception {
             inProgressSnapshots.add(AbstractSnapshotIntegTestCase.startFullSnapshot(logger, repoName, snapshotName, false));
         }
         AbstractSnapshotIntegTestCase.awaitNumberOfSnapshotsInProgress(logger, inProgressCount);
-        AbstractSnapshotIntegTestCase.awaitClusterState(logger, state -> {
+        AbstractSnapshotIntegTestCase.awaitClusterState(state -> {
             final var snapshotsInProgress = SnapshotsInProgress.get(state);
             boolean firstIndexSuccessfullySnapshot = snapshotsInProgress.asStream()
                 .flatMap(s -> s.shards().entrySet().stream())

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

@@ -213,7 +213,6 @@ public void runRepeatedlyWhileChangingMaster(Runnable runnable) throws Exception
 
         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)
         );

server/src/internalClusterTest/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java

@@ -1077,7 +1077,6 @@ public void onRequestSent(
         final ActionFuture<AcknowledgedResponse> deleteResponse = startDeleteSnapshot(repoName, snapshotName);
 
         awaitClusterState(
-            logger,
             otherDataNode,
             state -> SnapshotsInProgress.get(state)
                 .forRepo(repoName)

test/framework/src/main/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java

@@ -557,7 +557,7 @@ protected void addBwCFailedSnapshot(String repoName, String snapshotName, Map<St
         );
     }
 
-    protected void awaitNDeletionsInProgress(int count) throws Exception {
+    protected void awaitNDeletionsInProgress(int count) {
         logger.info("--> wait for [{}] deletions to show up in the cluster state", count);
         awaitClusterState(state -> SnapshotDeletionsInProgress.get(state).getEntries().size() == count);
     }
@@ -569,7 +569,6 @@ protected void awaitNoMoreRunningOperations() throws Exception {
     protected void awaitNoMoreRunningOperations(String viaNode) throws Exception {
         logger.info("--> verify no more operations in the cluster state");
         awaitClusterState(
-            logger,
             viaNode,
             state -> SnapshotsInProgress.get(state).isEmpty() && SnapshotDeletionsInProgress.get(state).hasDeletionsInProgress() == false
         );
@@ -604,13 +603,13 @@ public static ActionFuture<CreateSnapshotResponse> startFullSnapshot(
             .execute();
     }
 
-    protected void awaitNumberOfSnapshotsInProgress(int count) throws Exception {
+    protected void awaitNumberOfSnapshotsInProgress(int count) {
         awaitNumberOfSnapshotsInProgress(logger, count);
     }
 
-    public static void awaitNumberOfSnapshotsInProgress(Logger logger, int count) throws Exception {
+    public static void awaitNumberOfSnapshotsInProgress(Logger logger, int count) {
         logger.info("--> wait for [{}] snapshots to show up in the cluster state", count);
-        awaitClusterState(logger, state -> SnapshotsInProgress.get(state).count() == count);
+        awaitClusterState(state -> SnapshotsInProgress.get(state).count() == count);
     }
 
     protected SnapshotInfo assertSuccessful(ActionFuture<CreateSnapshotResponse> future) throws Exception {

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

@@ -8,17 +8,14 @@
  */
 package org.elasticsearch.test;
 
-import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.core.util.Throwables;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
-import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.ClusterStatePublicationEvent;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.NodeConnectionsService;
@@ -37,14 +34,12 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.telemetry.tracing.Tracer;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
 
@@ -219,33 +214,8 @@ public static void setAllElapsedMillis(ClusterStatePublicationEvent clusterState
         clusterStatePublicationEvent.setMasterApplyElapsedMillis(0L);
     }
 
-    public static void awaitClusterState(Logger logger, Predicate<ClusterState> statePredicate, ClusterService clusterService)
-        throws Exception {
-        final PlainActionFuture<Void> future = new PlainActionFuture<>();
-        ClusterStateObserver.waitForState(
-            clusterService,
-            clusterService.getClusterApplierService().threadPool().getThreadContext(),
-            new ClusterStateObserver.Listener() {
-                @Override
-                public void onNewClusterState(ClusterState state) {
-                    future.onResponse(null);
-                }
-
-                @Override
-                public void onClusterServiceClose() {
-                    future.onFailure(new NodeClosedException(clusterService.localNode()));
-                }
-
-                @Override
-                public void onTimeout(TimeValue timeout) {
-                    assert false : "onTimeout called with no timeout set";
-                }
-            },
-            statePredicate,
-            null,
-            logger
-        );
-        future.get(30L, TimeUnit.SECONDS);
+    public static void awaitClusterState(Predicate<ClusterState> statePredicate, ClusterService clusterService) {
+        ESTestCase.safeAwait(addTemporaryStateListener(clusterService, statePredicate, TimeValue.THIRTY_SECONDS), TimeValue.THIRTY_SECONDS);
     }
 
     public static void awaitNoPendingTasks(ClusterService clusterService) {

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

@@ -17,7 +17,6 @@
 import com.carrotsearch.randomizedtesting.generators.RandomPicks;
 
 import org.apache.http.HttpHost;
-import org.apache.logging.log4j.Logger;
 import org.apache.lucene.search.Sort;
 import org.apache.lucene.search.TotalHits;
 import org.apache.lucene.tests.util.LuceneTestCase;
@@ -1207,16 +1206,12 @@ public static PendingClusterTasksResponse getClusterPendingTasks(Client client)
         }
     }
 
-    protected void awaitClusterState(Predicate<ClusterState> statePredicate) throws Exception {
-        awaitClusterState(logger, internalCluster().getMasterName(), statePredicate);
+    public static void awaitClusterState(Predicate<ClusterState> statePredicate) {
+        awaitClusterState(internalCluster().getMasterName(), statePredicate);
     }
 
-    public static void awaitClusterState(Logger logger, Predicate<ClusterState> statePredicate) throws Exception {
-        awaitClusterState(logger, internalCluster().getMasterName(), statePredicate);
-    }
-
-    public static void awaitClusterState(Logger logger, String viaNode, Predicate<ClusterState> statePredicate) throws Exception {
-        ClusterServiceUtils.awaitClusterState(logger, statePredicate, internalCluster().getInstance(ClusterService.class, viaNode));
+    public static void awaitClusterState(String viaNode, Predicate<ClusterState> statePredicate) {
+        ClusterServiceUtils.awaitClusterState(statePredicate, internalCluster().getInstance(ClusterService.class, viaNode));
     }
 
     public static String getNodeId(String nodeName) {

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

@@ -2064,7 +2064,7 @@ public String getMasterName(@Nullable String viaNode) {
             throw new AssertionError("Unable to get master name, no node found");
         }
         try {
-            ClusterServiceUtils.awaitClusterState(logger, state -> state.nodes().getMasterNode() != null, clusterService(viaNode));
+            ClusterServiceUtils.awaitClusterState(state -> state.nodes().getMasterNode() != null, clusterService(viaNode));
             final ClusterState state = client(viaNode).admin().cluster().prepareState(TEST_REQUEST_TIMEOUT).setLocal(true).get().getState();
             final DiscoveryNode masterNode = state.nodes().getMasterNode();
             if (masterNode == null) {

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/IndexLifecycleRunnerTests.java

@@ -488,7 +488,6 @@ public void doTestRunPolicyWithFailureToReadPolicy(boolean asyncAction, boolean
 
         // The cluster state can take a few extra milliseconds to update after the steps are executed
         ClusterServiceUtils.awaitClusterState(
-            logger,
             s -> s.metadata().getProject(state.projectId()).index(indexMetadata.getIndex()).getLifecycleExecutionState().stepInfo() != null,
             clusterService
         );

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlSingleNodeTestCase.java

@@ -121,11 +121,7 @@ public void tearDown() throws Exception {
 
     protected void waitForMlTemplates() throws Exception {
         // block until the templates are installed
-        ClusterServiceUtils.awaitClusterState(
-            logger,
-            MachineLearning::criticalTemplatesInstalled,
-            getInstanceFromNode(ClusterService.class)
-        );
+        ClusterServiceUtils.awaitClusterState(MachineLearning::criticalTemplatesInstalled, getInstanceFromNode(ClusterService.class));
     }
 
     protected <T> void blockingCall(Consumer<ActionListener<T>> function, AtomicReference<T> response, AtomicReference<Exception> error)

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/support/BaseMlIntegTestCase.java

@@ -169,9 +169,9 @@ protected Collection<Class<? extends Plugin>> getMockPlugins() {
     }
 
     @Before
-    public void ensureTemplatesArePresent() throws Exception {
+    public void ensureTemplatesArePresent() {
         if (cluster().size() > 0) {
-            awaitClusterState(logger, MachineLearning::criticalTemplatesInstalled);
+            awaitClusterState(MachineLearning::criticalTemplatesInstalled);
         }
     }