Fix race in FileSettingsServiceIT.testSettingsAppliedOnStart (#134368) (#135196)

This was failing very very rarely due to unfortunate timing conditions.

Cluster state changes are applied to all nodes prior to being published
on the master node itself. However, the cluster state listener was
previously attached to the data node, allowing for a very short time
window where the state update wasn't visible on the master node itself
when checking in `assertClusterStateSaveOK`.

This changes the test to attach the listener to the master node itself
preventing above condition. I was initially worried it might be attached
too late in cases, but I couldn't reproduce any more issues this way.

> According to the dashboard, this started to fail on Monday (13/07). It
definitely does not look like a test failure, so I'm assigning a medium
priority, which we could raise if we discover this is a new bug.

I couldn't find any related commit that might have caused this. Still
wondering why this started failing around that time 🤔

Fixes #131210

(cherry picked from commit a29392c)

# Conflicts:
#	muted-tests.yml

Author: mosche

server/src/internalClusterTest/java/org/elasticsearch/reservedstate/service/FileSettingsServiceIT.java

@@ -188,18 +188,23 @@ public void clusterChanged(ClusterChangedEvent event) {
         return new Tuple<>(savedClusterState, metadataVersion);
     }
 
-    private Tuple<CountDownLatch, AtomicLong> setupClusterStateListener(String node, long version) {
+    private Tuple<CountDownLatch, AtomicLong> setupClusterStateListener(String node, long fileSettingsVersion) {
         ClusterService clusterService = internalCluster().clusterService(node);
         CountDownLatch savedClusterState = new CountDownLatch(1);
         AtomicLong metadataVersion = new AtomicLong(-1);
         clusterService.addListener(new ClusterStateListener() {
             @Override
             public void clusterChanged(ClusterChangedEvent event) {
                 ReservedStateMetadata reservedState = event.state().metadata().reservedStateMetadata().get(FileSettingsService.NAMESPACE);
-                if (reservedState != null && reservedState.version() == version) {
+                if (reservedState != null && reservedState.version() == fileSettingsVersion) {
                     clusterService.removeListener(this);
                     metadataVersion.set(event.state().metadata().version());
                     savedClusterState.countDown();
+                    logger.info(
+                        "done waiting for file settings [version: {}, metadata version: {}]",
+                        event.state().version(),
+                        event.state().metadata().version()
+                    );
                 }
             }
         });
@@ -265,15 +270,16 @@ public void testSettingsAppliedOnStart() throws Exception {
         FileSettingsService dataFileSettingsService = internalCluster().getInstance(FileSettingsService.class, dataNode);
 
         assertFalse(dataFileSettingsService.watching());
-        var savedClusterState = setupClusterStateListener(dataNode, versionCounter.incrementAndGet());
+        long expectedVersion = versionCounter.incrementAndGet();
 
         // In internal cluster tests, the nodes share the config directory, so when we write with the data node path
         // the master will pick it up on start
-        writeJSONFile(dataNode, testJSON, logger, versionCounter.get());
+        writeJSONFile(dataNode, testJSON, logger, expectedVersion);
 
         logger.info("--> start master node");
         final String masterNode = internalCluster().startMasterOnlyNode();
         awaitMasterNode(internalCluster().getNonMasterNodeName(), masterNode);
+        var savedClusterState = setupClusterStateListener(masterNode, expectedVersion);
 
         FileSettingsService masterFileSettingsService = internalCluster().getInstance(FileSettingsService.class, masterNode);