[Zen2] Simulate scheduling delays (#34181)

Today we schedule tasks (both immediate and future ones) exactly when
requested. In fact it is more realistic to allow for a small amount of delay in
the scheduling of tasks, and this helps to exercise more interleavings of
actions and therefore to improve test coverage.

This change adds to the DeterministicTaskQueue the ability to add a random
delay to the scheduling of tasks.

This change also provides more explicit timeouts for stabilisation in the
CoordinatorTests.

Using the randomised scheduling feature in the CoordinatorTests also found a
situation in which we could become a leader, then a candidate, and then a
leader again very quickly, causing a clash of the _BECOME_MASTER_ and
_FINISH_ELECTION_ tasks. We change their behaviour to not consider these
duplicates to be problematic.

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/ClusterStateTaskExecutor.java

@@ -58,7 +58,7 @@ default void clusterStatePublished(ClusterChangedEvent clusterChangedEvent) {
      * This allows groupd task description but the submitting source.
      */
     default String describeTasks(List<T> tasks) {
-        return String.join(", ", tasks.stream().map(t -> (CharSequence)t.toString()).filter(t -> t.length() == 0)::iterator);
+        return String.join(", ", tasks.stream().map(t -> (CharSequence)t.toString()).filter(t -> t.length() > 0)::iterator);
     }
 
     /**

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

@@ -417,8 +417,8 @@ public void invariant() {
         synchronized (mutex) {
             final Optional<DiscoveryNode> peerFinderLeader = peerFinder.getLeader();
             assert peerFinder.getCurrentTerm() == getCurrentTerm();
-            assert followersChecker.getFastResponseState().term == getCurrentTerm();
-            assert followersChecker.getFastResponseState().mode == getMode();
+            assert followersChecker.getFastResponseState().term == getCurrentTerm() : followersChecker.getFastResponseState();
+            assert followersChecker.getFastResponseState().mode == getMode() : followersChecker.getFastResponseState();
             if (mode == Mode.LEADER) {
                 final boolean becomingMaster = getStateForMasterService().term() != getCurrentTerm();
 

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

@@ -287,9 +287,9 @@ public void close(Mode newMode) {
 
                 final String stateUpdateSource = "elected-as-master ([" + pendingAsTasks.size() + "] nodes joined)";
 
-                pendingAsTasks.put(JoinTaskExecutor.BECOME_MASTER_TASK, (source, e) -> {
+                pendingAsTasks.put(JoinTaskExecutor.newBecomeMasterTask(), (source, e) -> {
                 });
-                pendingAsTasks.put(JoinTaskExecutor.FINISH_ELECTION_TASK, (source, e) -> {
+                pendingAsTasks.put(JoinTaskExecutor.newFinishElectionTask(), (source, e) -> {
                 });
                 masterService.submitStateUpdateTasks(stateUpdateSource, pendingAsTasks, ClusterStateTaskConfig.build(Priority.URGENT),
                     joinTaskExecutor);

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

@@ -63,6 +63,17 @@ public String reason() {
         public String toString() {
             return node != null ? node + " " + reason : reason;
         }
+
+        public boolean isBecomeMasterTask() {
+            return reason.equals(BECOME_MASTER_TASK_REASON);
+        }
+
+        public boolean isFinishElectionTask() {
+            return reason.equals(FINISH_ELECTION_TASK_REASON);
+        }
+
+        private static final String BECOME_MASTER_TASK_REASON = "_BECOME_MASTER_TASK_";
+        private static final String FINISH_ELECTION_TASK_REASON = "_FINISH_ELECTION_";
     }
 
     public JoinTaskExecutor(AllocationService allocationService, Logger logger) {
@@ -78,10 +89,11 @@ public ClusterTasksResult<Task> execute(ClusterState currentState, List<Task> jo
         boolean nodesChanged = false;
         ClusterState.Builder newState;
 
-        if (joiningNodes.size() == 1 && joiningNodes.get(0).equals(FINISH_ELECTION_TASK)) {
+        if (joiningNodes.size() == 1 && joiningNodes.get(0).isFinishElectionTask()) {
             return results.successes(joiningNodes).build(currentState);
-        } else if (currentNodes.getMasterNode() == null && joiningNodes.contains(BECOME_MASTER_TASK)) {
-            assert joiningNodes.contains(FINISH_ELECTION_TASK) : "becoming a master but election is not finished " + joiningNodes;
+        } else if (currentNodes.getMasterNode() == null && joiningNodes.stream().anyMatch(Task::isBecomeMasterTask)) {
+            assert joiningNodes.stream().anyMatch(Task::isFinishElectionTask)
+                : "becoming a master but election is not finished " + joiningNodes;
             // use these joins to try and become the master.
             // Note that we don't have to do any validation of the amount of joining nodes - the commit
             // during the cluster state publishing guarantees that we have enough
@@ -104,7 +116,7 @@ public ClusterTasksResult<Task> execute(ClusterState currentState, List<Task> jo
         final boolean enforceMajorVersion = currentState.getBlocks().hasGlobalBlock(STATE_NOT_RECOVERED_BLOCK) == false;
         // processing any joins
         for (final Task joinTask : joiningNodes) {
-            if (joinTask.equals(BECOME_MASTER_TASK) || joinTask.equals(FINISH_ELECTION_TASK)) {
+            if (joinTask.isBecomeMasterTask() || joinTask.isFinishElectionTask()) {
                 // noop
             } else if (currentNodes.nodeExists(joinTask.node())) {
                 logger.debug("received a join request for an existing node [{}]", joinTask.node());
@@ -146,7 +158,7 @@ private ClusterState.Builder becomeMasterAndTrimConflictingNodes(ClusterState cu
         nodesBuilder.masterNodeId(currentState.nodes().getLocalNodeId());
 
         for (final Task joinTask : joiningNodes) {
-            if (joinTask.equals(BECOME_MASTER_TASK) || joinTask.equals(FINISH_ELECTION_TASK)) {
+            if (joinTask.isBecomeMasterTask() || joinTask.isFinishElectionTask()) {
                 // noop
             } else {
                 final DiscoveryNode joiningNode = joinTask.node();
@@ -180,16 +192,17 @@ public boolean runOnlyOnMaster() {
         return false;
     }
 
-    /**
-     * a task indicated that the current node should become master, if no current master is known
-     */
-    public static final Task BECOME_MASTER_TASK = new Task(null, "_BECOME_MASTER_TASK_");
+    public static Task newBecomeMasterTask() {
+        return new Task(null, Task.BECOME_MASTER_TASK_REASON);
+    }
 
     /**
      * a task that is used to signal the election is stopped and we should process pending joins.
-     * it may be use in combination with {@link JoinTaskExecutor#BECOME_MASTER_TASK}
+     * it may be used in combination with {@link JoinTaskExecutor#newBecomeMasterTask()}
      */
-    public static final Task FINISH_ELECTION_TASK = new Task(null, "_FINISH_ELECTION_");
+    public static Task newFinishElectionTask() {
+        return new Task(null, Task.FINISH_ELECTION_TASK_REASON);
+    }
 
     /**
      * Ensures that all indices are compatible with the given node version. This will ensure that all indices in the given metadata

server/src/main/java/org/elasticsearch/discovery/HandshakingTransportAddressConnector.java

@@ -22,6 +22,8 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -69,7 +71,9 @@ protected void doRun() throws Exception {
 
                 // TODO if transportService is already connected to this address then skip the handshaking
 
-                final DiscoveryNode targetNode = new DiscoveryNode(transportAddress.toString(), transportAddress, emptyMap(),
+                final DiscoveryNode targetNode = new DiscoveryNode("", transportAddress.toString(),
+                    UUIDs.randomBase64UUID(Randomness.get()), // generated deterministically for reproducible tests
+                    transportAddress.address().getHostString(), transportAddress.getAddress(), transportAddress, emptyMap(),
                     emptySet(), Version.CURRENT.minimumCompatibilityVersion());
 
                 logger.trace("[{}] opening probe connection", this);

server/src/main/java/org/elasticsearch/discovery/PeerFinder.java

@@ -411,7 +411,7 @@ public void handleResponse(PeersResponse response) {
                     @Override
                     public void handleException(TransportException exp) {
                         peersRequestInFlight = false;
-                        logger.debug(new ParameterizedMessage("{} peers request failed", this), exp);
+                        logger.debug(new ParameterizedMessage("{} peers request failed", Peer.this), exp);
                     }
 
                     @Override

server/src/main/java/org/elasticsearch/discovery/zen/NodeJoinController.java

@@ -276,16 +276,16 @@ assert isEnoughPendingJoins(getPendingMasterJoinsCount()) : "becoming a master b
             final String source = "zen-disco-elected-as-master ([" + tasks.size() + "] nodes joined)";
 
             // noop listener, the election finished listener determines result
-            tasks.put(JoinTaskExecutor.BECOME_MASTER_TASK, (source1, e) -> {});
-            tasks.put(JoinTaskExecutor.FINISH_ELECTION_TASK, electionFinishedListener);
+            tasks.put(JoinTaskExecutor.newBecomeMasterTask(), (source1, e) -> {});
+            tasks.put(JoinTaskExecutor.newFinishElectionTask(), electionFinishedListener);
             masterService.submitStateUpdateTasks(source, tasks, ClusterStateTaskConfig.build(Priority.URGENT), joinTaskExecutor);
         }
 
         public synchronized void closeAndProcessPending(String reason) {
             innerClose();
             Map<JoinTaskExecutor.Task, ClusterStateTaskListener> tasks = getPendingAsTasks(reason);
             final String source = "zen-disco-election-stop [" + reason + "]";
-            tasks.put(JoinTaskExecutor.FINISH_ELECTION_TASK, electionFinishedListener);
+            tasks.put(JoinTaskExecutor.newFinishElectionTask(), electionFinishedListener);
             masterService.submitStateUpdateTasks(source, tasks, ClusterStateTaskConfig.build(Priority.URGENT), joinTaskExecutor);
         }