CoordinatorTests sometimes needs three term bumps

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

@@ -236,6 +236,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
@@ -1165,6 +1167,7 @@ protected void onSendRequest(
                 final ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
                 clusterApplierService = new DisruptableClusterApplierService(
                     localNode.getId(),
+                    localNode.getEphemeralId(),
                     settings,
                     clusterSettings,
                     deterministicTaskQueue,
@@ -1612,20 +1615,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,
+            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 -> {
@@ -1648,7 +1654,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,27 +7,40 @@
  */
 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) {
-        super(name, 0, 1, 0L, TimeUnit.MILLISECONDS, r -> new Thread() {
+    public MockSinglePrioritizingExecutor(
+        String nodeName,
+        String nodeId,
+        DeterministicTaskQueue deterministicTaskQueue,
+        ThreadPool threadPool
+    ) {
+        super(nodeName, 0, 1, 0L, TimeUnit.MILLISECONDS, r -> new Thread() {
             @Override
             public void start() {
                 deterministicTaskQueue.scheduleNow(new Runnable() {
                     @Override
                     public void run() {
-                        try {
+                        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) {