TESTS: Real Coordinator in SnapshotServiceTests (#37162)

* TESTS: Real Coordinator in SnapshotServiceTests

* Introduce real coordinator in SnapshotServiceTests to be able to test network disruptions realistically
  * Make adjustments to cluster applier service so that we can pass a mocked single threaded executor for tests

Author: original-brownbear

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -56,6 +56,8 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.DiscoverySettings;
 import org.elasticsearch.discovery.DiscoveryStats;
@@ -872,12 +874,7 @@ public void publish(ClusterChangedEvent clusterChangedEvent, ActionListener<Void
                     return;
                 }
 
-                // there is no equals on cluster state, so we just serialize it to XContent and compare JSON representation
-                assert clusterChangedEvent.previousState() == coordinationState.get().getLastAcceptedState() ||
-                    Strings.toString(clusterChangedEvent.previousState()).equals(
-                        Strings.toString(clusterStateWithNoMasterBlock(coordinationState.get().getLastAcceptedState())))
-                    : Strings.toString(clusterChangedEvent.previousState()) + " vs "
-                    + Strings.toString(clusterStateWithNoMasterBlock(coordinationState.get().getLastAcceptedState()));
+                assert assertPreviousStateConsistency(clusterChangedEvent);
 
                 final ClusterState clusterState = clusterChangedEvent.state();
 
@@ -917,6 +914,22 @@ public String toString() {
         }
     }
 
+    // there is no equals on cluster state, so we just serialize it to XContent and compare Maps
+    // deserialized from the resulting JSON
+    private boolean assertPreviousStateConsistency(ClusterChangedEvent event) {
+        assert event.previousState() == coordinationState.get().getLastAcceptedState() ||
+            XContentHelper.convertToMap(
+                JsonXContent.jsonXContent, Strings.toString(event.previousState()), false
+            ).equals(
+                XContentHelper.convertToMap(
+                    JsonXContent.jsonXContent,
+                    Strings.toString(clusterStateWithNoMasterBlock(coordinationState.get().getLastAcceptedState())),
+                    false))
+            : Strings.toString(event.previousState()) + " vs "
+            + Strings.toString(clusterStateWithNoMasterBlock(coordinationState.get().getLastAcceptedState()));
+        return true;
+    }
+
     private <T> ActionListener<T> wrapWithMutex(ActionListener<T> listener) {
         return new ActionListener<T>() {
             @Override

server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java

@@ -89,7 +89,7 @@ public class ClusterApplierService extends AbstractLifecycleComponent implements
 
     private final Collection<ClusterStateListener> clusterStateListeners = new CopyOnWriteArrayList<>();
     private final Collection<TimeoutClusterStateListener> timeoutClusterStateListeners =
-        Collections.newSetFromMap(new ConcurrentHashMap<TimeoutClusterStateListener, Boolean>());
+        Collections.newSetFromMap(new ConcurrentHashMap<>());
 
     private final LocalNodeMasterListeners localNodeMasterListeners;
 
@@ -134,11 +134,15 @@ protected synchronized void doStart() {
         Objects.requireNonNull(nodeConnectionsService, "please set the node connection service before starting");
         Objects.requireNonNull(state.get(), "please set initial state before starting");
         addListener(localNodeMasterListeners);
-        threadPoolExecutor = EsExecutors.newSinglePrioritizing(
-                nodeName + "/" + CLUSTER_UPDATE_THREAD_NAME,
-                daemonThreadFactory(nodeName, CLUSTER_UPDATE_THREAD_NAME),
-                threadPool.getThreadContext(),
-                threadPool.scheduler());
+        threadPoolExecutor = createThreadPoolExecutor();
+    }
+
+    protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
+        return EsExecutors.newSinglePrioritizing(
+            nodeName + "/" + CLUSTER_UPDATE_THREAD_NAME,
+            daemonThreadFactory(nodeName, CLUSTER_UPDATE_THREAD_NAME),
+            threadPool.getThreadContext(),
+            threadPool.scheduler());
     }
 
     class UpdateTask extends SourcePrioritizedRunnable implements Function<ClusterState, ClusterState> {

server/src/main/java/org/elasticsearch/cluster/service/ClusterService.java

@@ -71,8 +71,13 @@ public class ClusterService extends AbstractLifecycleComponent {
 
     private final String nodeName;
 
-    public ClusterService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool,
-        MasterService masterService) {
+    public ClusterService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) {
+        this(settings, clusterSettings, new MasterService(Node.NODE_NAME_SETTING.get(settings), settings, threadPool),
+            new ClusterApplierService(Node.NODE_NAME_SETTING.get(settings), settings, clusterSettings, threadPool));
+    }
+
+    public ClusterService(Settings settings, ClusterSettings clusterSettings, MasterService masterService,
+        ClusterApplierService clusterApplierService) {
         super(settings);
         this.settings = settings;
         this.nodeName = Node.NODE_NAME_SETTING.get(settings);
@@ -84,11 +89,7 @@ public ClusterService(Settings settings, ClusterSettings clusterSettings, Thread
             this::setSlowTaskLoggingThreshold);
         // Add a no-op update consumer so changes are logged
         this.clusterSettings.addAffixUpdateConsumer(USER_DEFINED_META_DATA, (first, second) -> {}, (first, second) -> {});
-        this.clusterApplierService = new ClusterApplierService(nodeName, settings, clusterSettings, threadPool);
-    }
-
-    public ClusterService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) {
-        this(settings, clusterSettings, threadPool, new MasterService(Node.NODE_NAME_SETTING.get(settings), settings, threadPool));
+        this.clusterApplierService = clusterApplierService;
     }
 
     private void setSlowTaskLoggingThreshold(TimeValue slowTaskLoggingThreshold) {