CoordinatorTests sometimes needs three term bumps

Today we require the cluster to stabilise in a time period that allows
time for the first election to encounter conflicts. However on very rare
occasions there might be an election conflict in the second election
too. This commit extends the stabilisation timeout to allow for this.

Similar to elastic#64462
Closes elastic#78370

Author: DaveCTurner

server/src/test/java/org/elasticsearch/cluster/InternalClusterInfoServiceSchedulingTests.java

@@ -59,7 +59,7 @@ public void testScheduling() {
         final ClusterApplierService clusterApplierService = new ClusterApplierService("test", settings, clusterSettings, threadPool) {
             @Override
             protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
-                return new MockSinglePrioritizingExecutor("mock-executor", deterministicTaskQueue, threadPool);
+                return new MockSinglePrioritizingExecutor("mock-executor", "", deterministicTaskQueue, threadPool);
             }
         };
 

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -1628,7 +1628,7 @@ private final class TestClusterNode {
                     new ClusterApplierService(node.getName(), settings, clusterSettings, threadPool) {
                         @Override
                         protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
-                            return new MockSinglePrioritizingExecutor(node.getName(), deterministicTaskQueue, threadPool);
+                            return new MockSinglePrioritizingExecutor(node.getName(), node.getId(), deterministicTaskQueue, threadPool);
                         }
 
                         @Override

test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java

@@ -235,6 +235,8 @@ protected static int defaultInt(Setting<Integer> setting) {
             + DEFAULT_ELECTION_DELAY
             // perhaps there is an election collision requiring another publication (which times out) and a term bump
             + defaultMillis(PUBLISH_TIMEOUT_SETTING) + DEFAULT_ELECTION_DELAY
+            // very rarely there is another election collision requiring another publication (which times out) and a term bump
+            + defaultMillis(PUBLISH_TIMEOUT_SETTING) + DEFAULT_ELECTION_DELAY
             // then wait for the new leader to notice that the old leader is unresponsive
             + (defaultMillis(FOLLOWER_CHECK_INTERVAL_SETTING) + defaultMillis(FOLLOWER_CHECK_TIMEOUT_SETTING))
             * defaultInt(FOLLOWER_CHECK_RETRY_COUNT_SETTING)
@@ -1029,8 +1031,13 @@ protected void onSendRequest(
                 masterService = new AckedFakeThreadPoolMasterService(localNode.getId(), "test", threadPool,
                     runnable -> deterministicTaskQueue.scheduleNow(onNode(runnable)));
                 final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
-                clusterApplierService = new DisruptableClusterApplierService(localNode.getId(), settings, clusterSettings,
-                    deterministicTaskQueue, threadPool);
+                clusterApplierService = new DisruptableClusterApplierService(
+                    localNode.getId(),
+                    localNode.getEphemeralId(),
+                    settings,
+                    clusterSettings,
+                    deterministicTaskQueue,
+                    threadPool);
                 clusterService = new ClusterService(settings, clusterSettings, masterService, clusterApplierService);
                 clusterService.setNodeConnectionsService(
                     new NodeConnectionsService(clusterService.getSettings(), threadPool, transportService));
@@ -1442,15 +1449,23 @@ public void onNodeAck(DiscoveryNode node, Exception e) {
 
     static class DisruptableClusterApplierService extends ClusterApplierService {
         private final String nodeName;
+        private final String nodeId;
         private final DeterministicTaskQueue deterministicTaskQueue;
         private final ThreadPool threadPool;
         ClusterStateApplyResponse clusterStateApplyResponse = ClusterStateApplyResponse.SUCCEED;
         private boolean applicationMayFail;
 
-        DisruptableClusterApplierService(String nodeName, Settings settings, ClusterSettings clusterSettings,
-                                         DeterministicTaskQueue deterministicTaskQueue, ThreadPool threadPool) {
+        DisruptableClusterApplierService(
+            String nodeName,
+            String nodeId,
+            Settings settings,
+            ClusterSettings clusterSettings,
+            DeterministicTaskQueue deterministicTaskQueue,
+            ThreadPool threadPool
+        ) {
             super(nodeName, settings, clusterSettings, threadPool);
             this.nodeName = nodeName;
+            this.nodeId = nodeId;
             this.deterministicTaskQueue = deterministicTaskQueue;
             this.threadPool = threadPool;
             addStateApplier(event -> {
@@ -1470,7 +1485,7 @@ static class DisruptableClusterApplierService extends ClusterApplierService {
 
         @Override
         protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
-            return new MockSinglePrioritizingExecutor(nodeName, deterministicTaskQueue, threadPool);
+            return new MockSinglePrioritizingExecutor(nodeName, nodeId, deterministicTaskQueue, threadPool);
         }
 
         @Override

test/framework/src/main/java/org/elasticsearch/cluster/coordination/MockSinglePrioritizingExecutor.java

@@ -7,22 +7,30 @@
  */
 package org.elasticsearch.cluster.coordination;
 
+import org.apache.logging.log4j.CloseableThreadContext;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.concurrent.TimeUnit;
 
+import static org.elasticsearch.common.util.concurrent.DeterministicTaskQueue.NODE_ID_LOG_CONTEXT_KEY;
+
 /**
  * Mock single threaded {@link PrioritizedEsThreadPoolExecutor} based on {@link DeterministicTaskQueue},
  * simulating the behaviour of an executor returned by {@link EsExecutors#newSinglePrioritizing}.
  */
 public class MockSinglePrioritizingExecutor extends PrioritizedEsThreadPoolExecutor {
 
-    public MockSinglePrioritizingExecutor(String name, DeterministicTaskQueue deterministicTaskQueue, ThreadPool threadPool) {
+    public MockSinglePrioritizingExecutor(
+        String nodeName,
+        String nodeId,
+        DeterministicTaskQueue deterministicTaskQueue,
+        ThreadPool threadPool
+    ) {
         super(
-            name,
+            nodeName,
             0,
             1,
             0L,
@@ -34,7 +42,12 @@ public void start() {
                         @Override
                         public void run() {
                             try {
-                                r.run();
+                                try (CloseableThreadContext.Instance ignored = CloseableThreadContext.put(
+                                    NODE_ID_LOG_CONTEXT_KEY,
+                                    '{' + nodeName + "}{" + nodeId + '}'
+                                )) {
+                                    r.run();
+                                }
                             } catch (KillWorkerError kwe) {
                                 // hacks everywhere
                             }

test/framework/src/test/java/org/elasticsearch/cluster/coordination/MockSinglePrioritizingExecutorTests.java

@@ -20,7 +20,12 @@ public class MockSinglePrioritizingExecutorTests extends ESTestCase {
     public void testPrioritizedEsThreadPoolExecutor() {
         final DeterministicTaskQueue taskQueue = new DeterministicTaskQueue();
 
-        final PrioritizedEsThreadPoolExecutor executor = new MockSinglePrioritizingExecutor("test", taskQueue, taskQueue.getThreadPool());
+        final PrioritizedEsThreadPoolExecutor executor = new MockSinglePrioritizingExecutor(
+            "test",
+            "",
+            taskQueue,
+            taskQueue.getThreadPool()
+        );
         final AtomicBoolean called1 = new AtomicBoolean();
         final AtomicBoolean called2 = new AtomicBoolean();
         executor.execute(new PrioritizedRunnable(Priority.NORMAL) {