Gather votes from all nodes (#34335)

Today we accept that some nodes may vote for the wrong master in an election.
This is mostly fine because they do end up joining the correct master in the
end, but the lack of a vote from every follower may prevent a future desirable
reconfiguration from taking place.

The solution is to hold another election in a yet-higher term in order to
collect a complete set of votes. Elections are somewhat disruptive so we should
think carefully about when this election should take place. One option is to
wait as late as possible (on the grounds that it might not ever be necessary).
This unfortunately makes it harder to predict how an
apparently-smoothly-running cluster will react to nodes leaving and joining.
Instead we prefer to perform the election as soon as possible in the leader's
term, adding "votes from all followers" to the invariants that we expect to
hold in a stable cluster. The start of a leader's term is already a somewhat
disrupted time for the cluster, so performing another election at this point
does not materially change the cluster's behaviour.

This change implements the logic needed to trigger a new election in order to
satisfy this extra stabilisation condition.

Author: DaveCTurner

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

@@ -222,8 +222,10 @@ PublishWithJoinResponse handlePublishRequest(PublishRequest publishRequest) {
             ensureTermAtLeast(sourceNode, publishRequest.getAcceptedState().term());
             final PublishResponse publishResponse = coordinationState.get().handlePublishRequest(publishRequest);
 
-            if (sourceNode.equals(getLocalNode()) == false) {
-                becomeFollower("handlePublishRequest", sourceNode);
+            if (sourceNode.equals(getLocalNode())) {
+                preVoteCollector.update(getPreVoteResponse(), getLocalNode());
+            } else {
+                becomeFollower("handlePublishRequest", sourceNode); // also updates preVoteCollector
             }
 
             return new PublishWithJoinResponse(publishResponse,
@@ -254,27 +256,31 @@ private void closePrevotingAndElectionScheduler() {
     }
 
     private void updateMaxTermSeen(final long term) {
-        maxTermSeen.updateAndGet(oldMaxTerm -> Math.max(oldMaxTerm, term));
-        // TODO if we are leader here, and there is no publication in flight, then we should bump our term
-        // (if we are leader and there _is_ a publication in flight then doing so would cancel the publication, so don't do that, but
-        // do check for this after the publication completes)
+        final long updatedMaxTermSeen = maxTermSeen.updateAndGet(oldMaxTerm -> Math.max(oldMaxTerm, term));
+        synchronized (mutex) {
+            if (mode == Mode.LEADER && publicationInProgress() == false && updatedMaxTermSeen > getCurrentTerm()) {
+                // Bump our term. However if there is a publication in flight then doing so would cancel the publication, so don't do that
+                // since we check whether a term bump is needed at the end of the publication too.
+                ensureTermAtLeast(getLocalNode(), updatedMaxTermSeen);
+                startElection();
+            }
+        }
     }
 
-    // TODO: make private again after removing term-bump workaround
-    void startElection() {
+    private void startElection() {
         synchronized (mutex) {
             // The preVoteCollector is only active while we are candidate, but it does not call this method with synchronisation, so we have
             // to check our mode again here.
             if (mode == Mode.CANDIDATE) {
                 final StartJoinRequest startJoinRequest
                     = new StartJoinRequest(getLocalNode(), Math.max(getCurrentTerm(), maxTermSeen.get()) + 1);
+                logger.debug("starting election with {}", startJoinRequest);
                 getDiscoveredNodes().forEach(node -> joinHelper.sendStartJoinRequest(startJoinRequest, node));
             }
         }
     }
 
-    // TODO: make private again after removing term-bump workaround
-    Optional<Join> ensureTermAtLeast(DiscoveryNode sourceNode, long targetTerm) {
+    private Optional<Join> ensureTermAtLeast(DiscoveryNode sourceNode, long targetTerm) {
         assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
         if (getCurrentTerm() < targetTerm) {
             return Optional.of(joinLeaderInTerm(new StartJoinRequest(sourceNode, targetTerm)));
@@ -289,9 +295,10 @@ private Join joinLeaderInTerm(StartJoinRequest startJoinRequest) {
             lastJoin = Optional.of(join);
             peerFinder.setCurrentTerm(getCurrentTerm());
             if (mode != Mode.CANDIDATE) {
-                becomeCandidate("joinLeaderInTerm"); // updates followersChecker
+                becomeCandidate("joinLeaderInTerm"); // updates followersChecker and preVoteCollector
             } else {
                 followersChecker.updateFastResponseState(getCurrentTerm(), mode);
+                preVoteCollector.update(getPreVoteResponse(), null);
             }
             return join;
         }
@@ -485,6 +492,8 @@ public void invariant() {
                 assert becomingMaster || getStateForMasterService().nodes().getMasterNodeId() != null : getStateForMasterService();
                 assert leaderCheckScheduler == null : leaderCheckScheduler;
                 assert applierState.nodes().getMasterNodeId() == null || getLocalNode().equals(applierState.nodes().getMasterNode());
+                assert preVoteCollector.getLeader() == getLocalNode() : preVoteCollector;
+                assert preVoteCollector.getPreVoteResponse().equals(getPreVoteResponse()) : preVoteCollector;
 
                 final boolean activePublication = currentPublication.map(CoordinatorPublication::isActiveForCurrentLeader).orElse(false);
                 if (becomingMaster && activePublication == false) {
@@ -517,6 +526,8 @@ public void invariant() {
                 assert leaderCheckScheduler != null;
                 assert followersChecker.getKnownFollowers().isEmpty();
                 assert currentPublication.map(Publication::isCommitted).orElse(true);
+                assert preVoteCollector.getLeader().equals(lastKnownLeader.get()) : preVoteCollector;
+                assert preVoteCollector.getPreVoteResponse().equals(getPreVoteResponse()) : preVoteCollector;
             } else {
                 assert mode == Mode.CANDIDATE;
                 assert joinAccumulator instanceof JoinHelper.CandidateJoinAccumulator;
@@ -528,6 +539,8 @@ public void invariant() {
                 assert followersChecker.getKnownFollowers().isEmpty();
                 assert applierState.nodes().getMasterNodeId() == null;
                 assert currentPublication.map(Publication::isCommitted).orElse(true);
+                assert preVoteCollector.getLeader() == null : preVoteCollector;
+                assert preVoteCollector.getPreVoteResponse().equals(getPreVoteResponse()) : preVoteCollector;
             }
         }
     }
@@ -537,7 +550,7 @@ boolean hasJoinVoteFrom(DiscoveryNode localNode) {
         return coordinationState.get().containsJoinVoteFor(localNode);
     }
 
-    void handleJoin(Join join) {
+    private void handleJoin(Join join) {
         synchronized (mutex) {
             ensureTermAtLeast(getLocalNode(), join.getTerm()).ifPresent(this::handleJoin);
 
@@ -547,7 +560,7 @@ void handleJoin(Join join) {
                 try {
                     coordinationState.get().handleJoin(join);
                 } catch (CoordinationStateRejectedException e) {
-                    logger.debug("failed to add join, ignoring", e);
+                    logger.debug(new ParameterizedMessage("failed to add {} - ignoring", join), e);
                 }
             } else {
                 coordinationState.get().handleJoin(join); // this might fail and bubble up the exception
@@ -753,6 +766,11 @@ class CoordinatorPublication extends Publication {
         private final AckListener ackListener;
         private final ActionListener<Void> publishListener;
 
+        // We may not have accepted our own state before receiving a join from another node, causing its join to be rejected (we cannot
+        // safely accept a join whose last-accepted term/version is ahead of ours), so store them up and process them at the end.
+        private final List<Join> receivedJoins = new ArrayList<>();
+        private boolean receivedJoinsProcessed;
+
         CoordinatorPublication(PublishRequest publishRequest, ListenableFuture<Void> localNodeAckEvent, AckListener ackListener,
                                ActionListener<Void> publishListener) {
             super(Coordinator.this.settings, publishRequest,
@@ -790,6 +808,7 @@ private void removePublicationAndPossiblyBecomeCandidate(String reason) {
 
             assert currentPublication.get() == this;
             currentPublication = Optional.empty();
+            logger.debug("publication ended unsuccessfully: {}", this);
 
             // check if node has not already switched modes (by bumping term)
             if (isActiveForCurrentLeader()) {
@@ -812,6 +831,10 @@ public void onResponse(Void ignore) {
                     assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
                     assert committed;
 
+                    receivedJoins.forEach(CoordinatorPublication.this::handleAssociatedJoin);
+                    assert receivedJoinsProcessed == false;
+                    receivedJoinsProcessed = true;
+
                     clusterApplier.onNewClusterState(CoordinatorPublication.this.toString(), () -> applierState,
                         new ClusterApplyListener() {
                             @Override
@@ -828,6 +851,7 @@ public void onSuccess(String source) {
                                 synchronized (mutex) {
                                     assert currentPublication.get() == CoordinatorPublication.this;
                                     currentPublication = Optional.empty();
+                                    logger.debug("publication ended successfully: {}", CoordinatorPublication.this);
                                     // trigger term bump if new term was found during publication
                                     updateMaxTermSeen(getCurrentTerm());
                                 }
@@ -850,6 +874,13 @@ public void onFailure(Exception e) {
             }, EsExecutors.newDirectExecutorService());
         }
 
+        private void handleAssociatedJoin(Join join) {
+            if (join.getTerm() == getCurrentTerm() && hasJoinVoteFrom(join.getSourceNode()) == false) {
+                logger.trace("handling {}", join);
+                handleJoin(join);
+            }
+        }
+
         @Override
         protected boolean isPublishQuorum(CoordinationState.VoteCollection votes) {
             assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
@@ -867,10 +898,26 @@ protected Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourc
         @Override
         protected void onJoin(Join join) {
             assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
-            if (join.getTerm() == getCurrentTerm()) {
-                handleJoin(join);
+            if (receivedJoinsProcessed) {
+                // a late response may arrive after the state has been locally applied, meaning that receivedJoins has already been
+                // processed, so we have to handle this late response here.
+                handleAssociatedJoin(join);
+            } else {
+                receivedJoins.add(join);
+            }
+        }
+
+        @Override
+        protected void onMissingJoin(DiscoveryNode discoveryNode) {
+            assert Thread.holdsLock(mutex) : "Coordinator mutex not held";
+            // The remote node did not include a join vote in its publish response. We do not persist joins, so it could be that the remote
+            // node voted for us and then rebooted, or it could be that it voted for a different node in this term. If we don't have a copy
+            // of a join from this node then we assume the latter and bump our term to obtain a vote from this node.
+            if (hasJoinVoteFrom(discoveryNode) == false) {
+                final long term = publishRequest.getAcceptedState().term();
+                logger.debug("onMissingJoin: no join vote from {}, bumping term to exceed {}", discoveryNode, term);
+                updateMaxTermSeen(term + 1);
             }
-            // TODO: what to do on missing join?
         }
 
         @Override

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

@@ -79,19 +79,33 @@ public Releasable start(final ClusterState clusterState, final Iterable<Discover
         return preVotingRound;
     }
 
+    // only for testing
+    PreVoteResponse getPreVoteResponse() {
+        return state.v2();
+    }
+
+    // only for testing
+    @Nullable
+    DiscoveryNode getLeader() {
+        return state.v1();
+    }
+
     public void update(final PreVoteResponse preVoteResponse, @Nullable final DiscoveryNode leader) {
         logger.trace("updating with preVoteResponse={}, leader={}", preVoteResponse, leader);
         state = new Tuple<>(leader, preVoteResponse);
     }
 
     private PreVoteResponse handlePreVoteRequest(final PreVoteRequest request) {
-        // TODO if we are a leader and the max term seen exceeds our term then we need to bump our term
+        updateMaxTermSeen.accept(request.getCurrentTerm());
 
         Tuple<DiscoveryNode, PreVoteResponse> state = this.state;
+        assert state != null : "received pre-vote request before fully initialised";
+
         final DiscoveryNode leader = state.v1();
+        final PreVoteResponse response = state.v2();
 
         if (leader == null) {
-            return state.v2();
+            return response;
         }
 
         if (leader.equals(request.getSourceNode())) {
@@ -100,7 +114,7 @@ private PreVoteResponse handlePreVoteRequest(final PreVoteRequest request) {
             // major drawback in offering a join to our old leader. The advantage of this is that it makes it slightly more likely that the
             // leader won't change, and also that its re-election will happen more quickly than if it had to wait for a quorum of followers
             // to also detect its failure.
-            return state.v2();
+            return response;
         }
 
         throw new CoordinationStateRejectedException("rejecting " + request + " as there is already a leader");
@@ -141,11 +155,7 @@ public void handleResponse(PreVoteResponse response) {
 
                     @Override
                     public void handleException(TransportException exp) {
-                        if (exp.getRootCause() instanceof CoordinationStateRejectedException) {
-                            logger.debug("{} failed: {}", this, exp.getRootCause().getMessage());
-                        } else {
-                            logger.debug(new ParameterizedMessage("{} failed", this), exp);
-                        }
+                        logger.debug(new ParameterizedMessage("{} failed", this), exp);
                     }
 
                     @Override

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

@@ -172,6 +172,8 @@ private void onPossibleCommitFailure() {
 
     protected abstract void onJoin(Join join);
 
+    protected abstract void onMissingJoin(DiscoveryNode discoveryNode);
+
     protected abstract void sendPublishRequest(DiscoveryNode destination, PublishRequest publishRequest,
                                                ActionListener<PublishWithJoinResponse> responseActionListener);
 
@@ -301,10 +303,16 @@ public void onResponse(PublishWithJoinResponse response) {
                     return;
                 }
 
-                response.getJoin().ifPresent(join -> {
+                if (response.getJoin().isPresent()) {
+                    final Join join = response.getJoin().get();
                     assert discoveryNode.equals(join.getSourceNode());
+                    assert join.getTerm() == response.getPublishResponse().getTerm() : response;
+                    logger.trace("handling join within publish response: {}", join);
                     onJoin(join);
-                });
+                } else {
+                    logger.trace("publish response from {} contained no join", discoveryNode);
+                    onMissingJoin(discoveryNode);
+                }
 
                 assert state == PublicationTargetState.SENT_PUBLISH_REQUEST : state + " -> " + PublicationTargetState.WAITING_FOR_QUORUM;
                 state = PublicationTargetState.WAITING_FOR_QUORUM;

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

@@ -30,8 +30,8 @@
  */
 public class PublishResponse implements Writeable {
 
-    protected final long term;
-    protected final long version;
+    private final long term;
+    private final long version;
 
     public PublishResponse(long term, long version) {
         assert term >= 0;

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

@@ -25,10 +25,12 @@
 import org.elasticsearch.cluster.ClusterState.VotingConfiguration;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNode.Role;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.cluster.coordination.CoordinationState.PersistedState;
+import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.test.ESTestCase;
@@ -37,6 +39,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
@@ -82,7 +85,11 @@ public void setupNodes() {
     }
 
     public static DiscoveryNode createNode(String id) {
-        return new DiscoveryNode(id, buildNewFakeTransportAddress(), Version.CURRENT);
+        final TransportAddress address = buildNewFakeTransportAddress();
+        return new DiscoveryNode("", id,
+            UUIDs.randomBase64UUID(random()), // generated deterministically for repeatable tests
+            address.address().getHostString(), address.getAddress(), address, Collections.emptyMap(),
+            EnumSet.allOf(Role.class), Version.CURRENT);
     }
 
     public void testSetInitialState() {