Test that join validation checks the cluster UUID (#41250)

This is a forward-port of parts of #41063 to `master`, adding a test to show
that join validation does indeed verify that the cluster UUIDs match.

Relates #37775

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

@@ -20,12 +20,18 @@ package org.elasticsearch.cluster.coordination;
 
 import org.apache.logging.log4j.Level;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListenerResponseHandler;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.NotMasterException;
+import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.transport.CapturingTransport;
 import org.elasticsearch.test.transport.CapturingTransport.CapturedRequest;
+import org.elasticsearch.test.transport.MockTransport;
 import org.elasticsearch.transport.RemoteTransportException;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportResponse;
@@ -35,6 +41,7 @@ import java.util.Collections;
 import java.util.Optional;
 
 import static org.elasticsearch.node.Node.NODE_NAME_SETTING;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.core.Is.is;
 
@@ -131,4 +138,40 @@ public class JoinHelperTests extends ESTestCase {
                 new RemoteTransportException("caused by NotMasterException",
                         new NotMasterException("test"))), is(Level.DEBUG));
     }
+
+    public void testJoinValidationRejectsMismatchedClusterUUID() {
+        DeterministicTaskQueue deterministicTaskQueue = new DeterministicTaskQueue(
+            Settings.builder().put(NODE_NAME_SETTING.getKey(), "node0").build(), random());
+        MockTransport mockTransport = new MockTransport();
+        DiscoveryNode localNode = new DiscoveryNode("node0", buildNewFakeTransportAddress(), Version.CURRENT);
+
+        final ClusterState localClusterState = ClusterState.builder(ClusterName.DEFAULT).metaData(MetaData.builder()
+            .generateClusterUuidIfNeeded().clusterUUIDCommitted(true)).build();
+
+        TransportService transportService = mockTransport.createTransportService(Settings.EMPTY,
+            deterministicTaskQueue.getThreadPool(), TransportService.NOOP_TRANSPORT_INTERCEPTOR,
+            x -> localNode, null, Collections.emptySet());
+        new JoinHelper(Settings.EMPTY, null, null, transportService, () -> 0L, () -> localClusterState,
+            (joinRequest, joinCallback) -> { throw new AssertionError(); }, startJoinRequest -> { throw new AssertionError(); },
+            Collections.emptyList()); // registers request handler
+        transportService.start();
+        transportService.acceptIncomingRequests();
+
+        final ClusterState otherClusterState = ClusterState.builder(ClusterName.DEFAULT).metaData(MetaData.builder()
+            .generateClusterUuidIfNeeded()).build();
+
+        final PlainActionFuture<TransportResponse.Empty> future = new PlainActionFuture<>();
+        transportService.sendRequest(localNode, JoinHelper.VALIDATE_JOIN_ACTION_NAME,
+            new ValidateJoinRequest(otherClusterState),
+            new ActionListenerResponseHandler<>(future, in -> TransportResponse.Empty.INSTANCE));
+        deterministicTaskQueue.runAllTasks();
+
+        final CoordinationStateRejectedException coordinationStateRejectedException
+            = expectThrows(CoordinationStateRejectedException.class, future::actionGet);
+        assertThat(coordinationStateRejectedException.getMessage(),
+            containsString("join validation on cluster state with a different cluster uuid"));
+        assertThat(coordinationStateRejectedException.getMessage(), containsString(localClusterState.metaData().clusterUUID()));
+        assertThat(coordinationStateRejectedException.getMessage(), containsString(otherClusterState.metaData().clusterUUID()));
+    }
+
 }