Don't fail on first node-join publication failure

Extends the `Coordinator.handleJoinRequest` `onFailure` method in the
callback to only fail if the node is not in already in the
`ClusterState`. This avoids a rare corner case where a master node's
`ClusterState` already has a node in it that is attempting to join the
cluster, and it logs an incorrect error message

Author: joshua-adams-1

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

@@ -672,28 +672,32 @@ public void onResponse(Releasable response) {
 
             @Override
             public void onFailure(Exception e) {
-                logger.warn(
-                    () -> format(
-                        "received join request from [%s] but could not connect back to the joining node",
-                        joinRequest.getSourceNode()
-                    ),
-                    e
-                );
-
-                joinListener.onFailure(
-                    // NodeDisconnectedException mainly to suppress uninteresting stack trace
-                    new NodeDisconnectedException(
-                        joinRequest.getSourceNode(),
-                        String.format(
-                            Locale.ROOT,
-                            "failure when opening connection back from [%s] to [%s]",
-                            getLocalNode().descriptionWithoutAttributes(),
-                            joinRequest.getSourceNode().descriptionWithoutAttributes()
+                // The failure of the first node-join publication does not imply that the join failed,
+                // because it may be retried and eventually succeed
+                if (!applierState.nodes().nodeExists(joinRequest.getSourceNode())) {
+                    logger.warn(
+                        () -> format(
+                            "received join request from [%s] but could not connect back to the joining node",
+                            joinRequest.getSourceNode()
                         ),
-                        JoinHelper.JOIN_ACTION_NAME,
                         e
-                    )
-                );
+                    );
+
+                    joinListener.onFailure(
+                        // NodeDisconnectedException mainly to suppress uninteresting stack trace
+                        new NodeDisconnectedException(
+                            joinRequest.getSourceNode(),
+                            String.format(
+                                Locale.ROOT,
+                                "failure when opening connection back from [%s] to [%s]",
+                                getLocalNode().descriptionWithoutAttributes(),
+                                joinRequest.getSourceNode().descriptionWithoutAttributes()
+                            ),
+                            JoinHelper.JOIN_ACTION_NAME,
+                            e
+                        )
+                    );
+                }
             }
         });
     }

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

@@ -1474,6 +1474,120 @@ public void assertMatched() {
         }
     }
 
+    // TODO:
+    // 1: Refactor this test and above
+    // 2: Add a test where it fails to begin with, then the handshake block is removed and it succeeds
+    @TestLogging(
+        reason = "test includes assertions about logging",
+        value = "org.elasticsearch.cluster.coordination.Coordinator:WARN,org.elasticsearch.cluster.coordination.JoinHelper:INFO"
+    )
+    public void testDoesNotReportConnectBackProblemsDuringJoiningIfNodeIsInClusterState() throws Exception {
+        try (var cluster = new Cluster(3)) {
+            cluster.runRandomly();
+            cluster.stabilise();
+
+            final var partitionedNode = cluster.getAnyNode();
+            partitionedNode.disconnect();
+            cluster.stabilise();
+
+            logger.info("--> removed [{}] but adding to master's cluster state", partitionedNode);
+            final ClusterNode leader = cluster.getAnyLeader();
+            leader.submitUpdateTask("updating cluster state",
+                cs -> {
+                    ClusterState cs2 = ClusterState.builder(cs)
+                        .nodes(DiscoveryNodes.builder(cs.nodes())
+                            .add(partitionedNode.getLocalNode())
+                            .build())
+                        .build();
+                    // Insert breakpoint here
+                    return cs2;
+                },
+                (e) -> {}
+            );
+
+            logger.info("--> healing [{}] but blocking handshakes", partitionedNode);
+            partitionedNode.heal();
+            leader.addActionBlock(TransportService.HANDSHAKE_ACTION_NAME);
+
+            try (var mockLog = MockLog.capture(Coordinator.class, JoinHelper.class)) {
+
+                // Since the node is in the cluster state, we do not expect this log
+                mockLog.addExpectation(
+                    new MockLog.UnseenEventExpectation(
+                        "connect-back failure",
+                        Coordinator.class.getCanonicalName(),
+                        Level.WARN,
+                        "*received join request from ["
+                            + partitionedNode.getLocalNode().descriptionWithoutAttributes()
+                            + "] but could not connect back to the joining node"
+                    )
+                );
+
+                // We do not expect an info log from JoinHelper about the handshake failure
+                mockLog.addExpectation(new MockLog.LoggingExpectation() {
+                    boolean matched = false;
+
+                    @Override
+                    public void match(LogEvent event) {
+                        if (event.getLevel() != Level.INFO) {
+                            return;
+                        }
+                        if (event.getLoggerName().equals(JoinHelper.class.getCanonicalName()) == false) {
+                            return;
+                        }
+
+                        var cause = event.getThrown();
+                        if (cause == null) {
+                            return;
+                        }
+                        cause = cause.getCause();
+                        if (cause == null) {
+                            return;
+                        }
+                        if (Regex.simpleMatch(
+                            "* failure when opening connection back from ["
+                                + leader.getLocalNode().descriptionWithoutAttributes()
+                                + "] to ["
+                                + partitionedNode.getLocalNode().descriptionWithoutAttributes()
+                                + "]",
+                            cause.getMessage()
+                        ) == false) {
+                            return;
+                        }
+                        if (cause.getStackTrace() != null && cause.getStackTrace().length != 0) {
+                            return;
+                        }
+                        matched = true;
+                    }
+
+                    @Override
+                    public void assertMatched() {
+                        assertFalse(matched);
+                    }
+                });
+
+                cluster.runFor(
+                    // This expects 8 tasks to be executed after PeerFinder handling wakeup:
+                    //
+                    // * connectToRemoteMasterNode[0.0.0.0:11]
+                    // * [internal:transport/handshake] from {node1} to {node2}
+                    // * response to [internal:transport/handshake] from {node1} to {node2}
+                    // * [internal:discovery/request_peers] from {node1} to
+                    // * response to [internal:discovery/request_peers] from {node1} to {node2}
+                    // * [internal:cluster/coordination/join] from {node1} to {node2}
+                    // * [internal:transport/handshake] from {node2} to {node1} (rejected due to action block)
+                    // * error response to [internal:cluster/coordination/join] from {node1} to {node2}
+                    //
+                    defaultMillis(DISCOVERY_FIND_PEERS_INTERVAL_SETTING) + 8 * DEFAULT_DELAY_VARIABILITY,
+                    "allowing time for join attempt"
+                );
+                mockLog.assertAllExpectationsMatched();
+            }
+
+            leader.clearActionBlocks();
+        }
+    }
+
     public void testFollowerRemovedIfUnableToSendRequestsToMaster() {
         try (Cluster cluster = new Cluster(3)) {
             cluster.runRandomly();