fix comments

Author: idegtiarenko

server/src/internalClusterTest/java/org/elasticsearch/cluster/coordination/RareClusterStateIT.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
@@ -28,6 +29,7 @@
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
@@ -37,7 +39,6 @@
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.disruption.BlockClusterStateProcessing;
-import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportSettings;
 
 import java.util.List;
@@ -133,20 +134,43 @@ private <Req extends ActionRequest, Res extends ActionResponse> ActionFuture<Res
         // Wait for no publication in progress to not accidentally cancel a publication different from the one triggered by the given
         // request.
         final Coordinator masterCoordinator = internalCluster().getCurrentMasterNodeInstance(Coordinator.class);
-        assertBusy(() -> {
-            assertFalse(masterCoordinator.publicationInProgress());
-            final long applierVersion = masterCoordinator.getApplierState().version();
-            for (Coordinator instance : internalCluster().getInstances(Coordinator.class)) {
-                assertEquals(instance.getApplierState().version(), applierVersion);
-            }
-        });
+
+        ensureNoPendingMasterTasks().actionGet(TimeValue.timeValueSeconds(30));
         ActionFuture<Res> future = req.execute();
-        // cancel all commit publications produced by the request
-        internalCluster().getCurrentMasterNodeInstance(ThreadPool.class).generic().execute(() -> {
-            while (!future.isDone()) {
-                masterCoordinator.cancelCommittedPublication();
-            }
-        });
+
+        // cancel the first cluster state update produced by the request above
+        assertBusy(() -> assertTrue(masterCoordinator.cancelCommittedPublication()));
+        // await and cancel any other forked cluster state updates that might be produced by the request
+        var task = ensureNoPendingMasterTasks();
+        while (task.isDone() == false) {
+            masterCoordinator.cancelCommittedPublication();
+            Thread.onSpinWait();
+        }
+        task.actionGet(TimeValue.timeValueSeconds(30));
+
+        return future;
+    }
+
+    private PlainActionFuture<Void> ensureNoPendingMasterTasks() {
+        var future = new PlainActionFuture<Void>();
+        internalCluster().getCurrentMasterNodeInstance(ClusterService.class)
+            .submitUnbatchedStateUpdateTask("test", new ClusterStateUpdateTask(Priority.LANGUID, TimeValue.timeValueSeconds(30)) {
+
+                @Override
+                public ClusterState execute(ClusterState currentState) {
+                    return currentState;
+                }
+
+                @Override
+                public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
+                    future.onResponse(null);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    future.onFailure(e);
+                }
+            });
         return future;
     }