[Zen2] Logging improvements in CoordinatorTests

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

@@ -158,13 +158,21 @@ public void setInitialState(ClusterState initialState) {
      */
     public Join handleStartJoin(StartJoinRequest startJoinRequest) {
         if (startJoinRequest.getTerm() <= getCurrentTerm()) {
-            logger.debug("handleStartJoin: ignored as term provided [{}] not greater than current term [{}]",
-                startJoinRequest.getTerm(), getCurrentTerm());
+            logger.debug("handleStartJoin: ignoring [{}] as term provided is not greater than current term [{}]",
+                startJoinRequest, getCurrentTerm());
             throw new CoordinationStateRejectedException("incoming term " + startJoinRequest.getTerm() +
                 " not greater than current term " + getCurrentTerm());
         }
 
-        logger.debug("handleStartJoin: updating term from [{}] to [{}]", getCurrentTerm(), startJoinRequest.getTerm());
+        logger.debug("handleStartJoin: leaving term [{}] due to {}", getCurrentTerm(), startJoinRequest);
+
+        if (joinVotes.isEmpty() == false) {
+            if (electionWon == false) {
+                logger.debug("handleStartJoin: failed election, discarding {}", joinVotes);
+            } else if (startJoinRequest.getSourceNode().equals(localNode) == false) {
+                logger.debug("handleStartJoin: standing down as leader, discarding {}", joinVotes);
+            }
+        }
 
         persistedState.setCurrentTerm(startJoinRequest.getTerm());
         assert getCurrentTerm() == startJoinRequest.getTerm();

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlocks;
+import org.elasticsearch.cluster.coordination.JoinHelper.InitialJoinAccumulator;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
@@ -99,7 +100,7 @@ public Coordinator(Settings settings, TransportService transportService, Allocat
         this.persistedStateSupplier = persistedStateSupplier;
         this.lastKnownLeader = Optional.empty();
         this.lastJoin = Optional.empty();
-        this.joinAccumulator = joinHelper.new CandidateJoinAccumulator();
+        this.joinAccumulator = new InitialJoinAccumulator();
         this.publishTimeout = PUBLISH_TIMEOUT_SETTING.get(settings);
         this.electionSchedulerFactory = new ElectionSchedulerFactory(settings, random, transportService.getThreadPool());
         this.preVoteCollector = new PreVoteCollector(settings, transportService, this::startElection, this::updateMaxTermSeen);

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

@@ -202,7 +202,6 @@ interface JoinAccumulator {
         void handleJoinRequest(DiscoveryNode sender, JoinCallback joinCallback);
 
         default void close(Mode newMode) {
-
         }
     }
 
@@ -220,6 +219,19 @@ public String toString() {
         }
     }
 
+    static class InitialJoinAccumulator implements JoinAccumulator {
+        @Override
+        public void handleJoinRequest(DiscoveryNode sender, JoinCallback joinCallback) {
+            assert false : "unexpected join from " + sender + " during initialisation";
+            joinCallback.onFailure(new CoordinationStateRejectedException("join target is not initialised yet"));
+        }
+
+        @Override
+        public String toString() {
+            return "InitialJoinAccumulator";
+        }
+    }
+
     static class FollowerJoinAccumulator implements JoinAccumulator {
         @Override
         public void handleJoinRequest(DiscoveryNode sender, JoinCallback joinCallback) {
@@ -265,13 +277,14 @@ public void close(Mode newMode) {
                 });
                 masterService.submitStateUpdateTasks(stateUpdateSource, pendingAsTasks, ClusterStateTaskConfig.build(Priority.URGENT),
                     joinTaskExecutor);
-            } else if (newMode == Mode.FOLLOWER) {
+            } else {
+                assert newMode == Mode.FOLLOWER : newMode;
                 joinRequestAccumulator.values().forEach(joinCallback -> joinCallback.onFailure(
                     new CoordinationStateRejectedException("became follower")));
-            } else {
-                assert newMode == Mode.CANDIDATE;
-                assert joinRequestAccumulator.isEmpty() : joinRequestAccumulator.keySet();
             }
+
+            // CandidateJoinAccumulator is only closed when becoming leader or follower, otherwise it accumulates all joins received
+            // regardless of term.
         }
 
         @Override

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

@@ -76,6 +76,8 @@ public void testCanUpdateClusterStateAfterStabilisation() {
 
         final ClusterNode leader = cluster.getAnyLeader();
         long finalValue = randomLong();
+
+        logger.info("--> submitting value [{}] to [{}]", finalValue, leader);
         leader.submitValue(finalValue);
         cluster.stabilise(); // TODO this should only need a short stabilisation
 
@@ -96,6 +98,7 @@ class Cluster {
 
         final List<ClusterNode> clusterNodes;
         final DeterministicTaskQueue deterministicTaskQueue = new DeterministicTaskQueue(
+            // TODO does ThreadPool need a node name any more?
             Settings.builder().put(NODE_NAME_SETTING.getKey(), "deterministic-task-queue").build());
         private final VotingConfiguration initialConfiguration;
 
@@ -267,17 +270,27 @@ boolean isLeader() {
             }
 
             void submitValue(final long value) {
-                masterService.submitStateUpdateTask("new value [" + value + "]", new ClusterStateUpdateTask() {
-                    @Override
-                    public ClusterState execute(ClusterState currentState) {
-                        return setValue(currentState, value);
-                    }
+                onNode(localNode, new Runnable() {
+                        @Override
+                        public void run() {
+                            masterService.submitStateUpdateTask("new value [" + value + "]", new ClusterStateUpdateTask() {
+                                @Override
+                                public ClusterState execute(ClusterState currentState) {
+                                    return setValue(currentState, value);
+                                }
+
+                                @Override
+                                public void onFailure(String source, Exception e) {
+                                    logger.debug(() -> new ParameterizedMessage("failed to publish: [{}]", source), e);
+                                }
+                            });
+                        }
 
                     @Override
-                    public void onFailure(String source, Exception e) {
-                        logger.debug(() -> new ParameterizedMessage("failed to publish: [{}]", source), e);
+                    public String toString() {
+                        return "submitStateUpdateTask: new value [" + value + "]";
                     }
-                });
+                }).run();
             }
         }