Disable logging in ClusterFormationFailureHelper on shutdown. (elastic#125244)

Modifies Coordinator to enable logging in ClusterFormationFailureHelper
when started and disables logging in ClusterFormationFailureHelper when
stopped.  The warning scheduler handling and invariant check in the
Coordinator are left as is, with the logging boolean set independently,
eliminating the need to hold the mutex in doStop() when Coordinator.stop()
is called when the Node is shutdown.

Closes elastic#105559.

* Update docs/changelog/125244.yaml

Author: JeremyDahlgren

docs/changelog/125244.yaml

@@ -0,0 +1,6 @@
+pr: 125244
+summary: Disable logging in `ClusterFormationFailureHelper` on shutdown
+area: Cluster Coordination
+type: bug
+issues:
+ - 105559

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

@@ -68,6 +68,7 @@ public class ClusterFormationFailureHelper {
     private final Runnable logLastFailedJoinAttempt;
     @Nullable // if no warning is scheduled
     private volatile WarningScheduler warningScheduler;
+    private volatile boolean loggingEnabled;
 
     /**
      * Works with the {@link JoinHelper} to log the latest node-join attempt failure and cluster state debug information. Must call
@@ -90,6 +91,11 @@ public ClusterFormationFailureHelper(
         this.clusterCoordinationExecutor = threadPool.executor(Names.CLUSTER_COORDINATION);
         this.clusterFormationWarningTimeout = DISCOVERY_CLUSTER_FORMATION_WARNING_TIMEOUT_SETTING.get(settings);
         this.logLastFailedJoinAttempt = logLastFailedJoinAttempt;
+        this.loggingEnabled = true;
+    }
+
+    public void setLoggingEnabled(boolean enabled) {
+        this.loggingEnabled = enabled;
     }
 
     public boolean isRunning() {
@@ -98,7 +104,7 @@ public boolean isRunning() {
 
     /**
      * Schedules a warning debug message to be logged in 'clusterFormationWarningTimeout' time, and periodically thereafter, until
-     * {@link ClusterFormationState#stop()} has been called.
+     * {@link ClusterFormationFailureHelper#stop()} has been called.
      */
     public void start() {
         assert warningScheduler == null;
@@ -125,7 +131,7 @@ public void onFailure(Exception e) {
 
                 @Override
                 protected void doRun() {
-                    if (isActive()) {
+                    if (isActive() && loggingEnabled) {
                         logLastFailedJoinAttempt.run();
                         logger.warn(
                             "{}; for troubleshooting guidance, see {}",

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

@@ -1104,6 +1104,7 @@ protected void doStart() {
             applierState = initialState;
             clusterApplier.setInitialState(initialState);
         }
+        clusterFormationFailureHelper.setLoggingEnabled(true);
     }
 
     public DiscoveryStats stats() {
@@ -1126,6 +1127,7 @@ public void startInitialJoin() {
     protected void doStop() {
         configuredHostsResolver.stop();
         joinValidationService.stop();
+        clusterFormationFailureHelper.setLoggingEnabled(false);
     }
 
     @Override

server/src/test/java/org/elasticsearch/cluster/coordination/ClusterFormationFailureHelperTests.java

@@ -171,6 +171,45 @@ public void testScheduling() {
 
         assertThat(warningCount.get(), is(5L));
         assertThat(logLastFailedJoinAttemptWarningCount.get(), is(5L));
+
+        // Temporarily disable logging and verify we don't get incremented logging counts.
+        clusterFormationFailureHelper.setLoggingEnabled(false);
+        warningCount.set(0);
+        logLastFailedJoinAttemptWarningCount.set(0);
+        clusterFormationFailureHelper.start();
+        clusterFormationFailureHelper.stop();
+        clusterFormationFailureHelper.start();
+        final long thirdStartTimeMillis = deterministicTaskQueue.getCurrentTimeMillis();
+
+        while (deterministicTaskQueue.getCurrentTimeMillis() - thirdStartTimeMillis < 5 * expectedDelayMillis) {
+            assertTrue(clusterFormationFailureHelper.isRunning());
+            if (deterministicTaskQueue.hasRunnableTasks()) {
+                deterministicTaskQueue.runRandomTask();
+            } else {
+                deterministicTaskQueue.advanceTime();
+            }
+        }
+
+        assertThat(warningCount.get(), is(0L));
+        assertThat(logLastFailedJoinAttemptWarningCount.get(), is(0L));
+
+        // Re-enable logging and verify the logging counts again.
+        clusterFormationFailureHelper.stop();
+        clusterFormationFailureHelper.start();
+        clusterFormationFailureHelper.setLoggingEnabled(true);
+        final long fourthStartTimeMillis = deterministicTaskQueue.getCurrentTimeMillis();
+
+        while (warningCount.get() < 5) {
+            assertTrue(clusterFormationFailureHelper.isRunning());
+            if (deterministicTaskQueue.hasRunnableTasks()) {
+                deterministicTaskQueue.runRandomTask();
+            } else {
+                deterministicTaskQueue.advanceTime();
+            }
+        }
+        assertThat(deterministicTaskQueue.getCurrentTimeMillis() - fourthStartTimeMillis, equalTo(5 * expectedDelayMillis));
+        assertThat(warningCount.get(), is(5L));
+        assertThat(logLastFailedJoinAttemptWarningCount.get(), is(5L));
     }
 
     public void testDescriptionOnMasterIneligibleNodes() {