Handle publish requests without attached joins

Author: DaveCTurner

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

@@ -887,7 +887,17 @@ protected void onJoin(Join join) {
             if (join.getTerm() == getCurrentTerm()) {
                 handleJoin(join);
             }
-            // TODO: what to do on missing 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) {
+                updateMaxTermSeen(publishRequest.getAcceptedState().term() + 1);
+            }
         }
 
         @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,14 @@ 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;
                     onJoin(join);
-                });
+                } else {
+                    onMissingJoin(discoveryNode);
+                }
 
                 assert state == PublicationTargetState.SENT_PUBLISH_REQUEST : state + " -> " + PublicationTargetState.WAITING_FOR_QUORUM;
                 state = PublicationTargetState.WAITING_FOR_QUORUM;

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

@@ -101,6 +101,7 @@ abstract class MockPublication extends Publication {
         Map<DiscoveryNode, ActionListener<PublishWithJoinResponse>> pendingPublications = new HashMap<>();
         Map<DiscoveryNode, ActionListener<TransportResponse.Empty>> pendingCommits = new HashMap<>();
         Map<DiscoveryNode, Join> joins = new HashMap<>();
+        Set<DiscoveryNode> missingJoins = new HashSet<>();
 
         MockPublication(Settings settings, PublishRequest publishRequest, Discovery.AckListener ackListener,
                                LongSupplier currentTimeSupplier) {
@@ -120,6 +121,11 @@ protected void onJoin(Join join) {
             assertNull(joins.put(join.getSourceNode(), join));
         }
 
+        @Override
+        protected void onMissingJoin(DiscoveryNode discoveryNode) {
+            assertTrue(missingJoins.add(discoveryNode));
+        }
+
         @Override
         protected void sendPublishRequest(DiscoveryNode destination, PublishRequest publishRequest,
                                           ActionListener<PublishWithJoinResponse> responseActionListener) {
@@ -182,14 +188,16 @@ public void testSimpleClusterStatePublishing() throws InterruptedException {
             assertNotEquals(processedNode1PublishResponse.get(), publication.pendingCommits.isEmpty());
             assertFalse(publication.joins.containsKey(e.getKey()));
             PublishWithJoinResponse publishWithJoinResponse = new PublishWithJoinResponse(publishResponse,
-                randomBoolean() ? Optional.empty() : Optional.of(new Join(e.getKey(), randomFrom(n1, n2, n3), randomNonNegativeLong(),
+                randomBoolean() ? Optional.empty() : Optional.of(new Join(e.getKey(), randomFrom(n1, n2, n3), publishResponse.getTerm(),
                     randomNonNegativeLong(), randomNonNegativeLong())));
             e.getValue().onResponse(publishWithJoinResponse);
             if (publishWithJoinResponse.getJoin().isPresent()) {
                 assertTrue(publication.joins.containsKey(e.getKey()));
+                assertFalse(publication.missingJoins.contains(e.getKey()));
                 assertEquals(publishWithJoinResponse.getJoin().get(), publication.joins.get(e.getKey()));
             } else {
                 assertFalse(publication.joins.containsKey(e.getKey()));
+                assertTrue(publication.missingJoins.contains(e.getKey()));
             }
             if (e.getKey().equals(n1)) {
                 processedNode1PublishResponse.set(true);