Validate cluster UUID when joining Zen1 cluster (#41063)

Today we fail to join a Zen2 cluster if the cluster UUID does not match our
own, but we do not perform the same validation when joining a Zen1 cluster.
This means that a Zen2 node will pass join validation and be added to a Zen1
cluster but will reject all cluster states from the master.

Relates #37775

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

@@ -153,6 +153,14 @@ public ClusterTasksResult<JoinTaskExecutor.Task> execute(ClusterState currentSta
         transportService.registerRequestHandler(MembershipAction.DISCOVERY_JOIN_VALIDATE_ACTION_NAME,
             ValidateJoinRequest::new, ThreadPool.Names.GENERIC,
             (request, channel, task) -> {
+                final ClusterState localState = currentStateSupplier.get();
+                if (localState.metaData().clusterUUIDCommitted() &&
+                    localState.metaData().clusterUUID().equals(request.getState().metaData().clusterUUID()) == false) {
+                    throw new CoordinationStateRejectedException("mixed-version cluster join validation on cluster state" +
+                        " with a different cluster uuid " + request.getState().metaData().clusterUUID() +
+                        " than local cluster uuid " + localState.metaData().clusterUUID()
+                        + ", rejecting");
+                }
                 joinValidators.forEach(action -> action.accept(transportService.getLocalNode(), request.getState()));
                 channel.sendResponse(Empty.INSTANCE);
             });

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

@@ -20,12 +20,19 @@
 
 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.discovery.zen.MembershipAction;
 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 +42,7 @@
 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 +139,48 @@ public void testFailedJoinAttemptLogLevel() {
                 new RemoteTransportException("caused by NotMasterException",
                         new NotMasterException("test"))), is(Level.DEBUG));
     }
+
+    public void testZen1JoinValidationRejectsMismatchedClusterUUID() {
+        assertJoinValidationRejectsMismatchedClusterUUID(MembershipAction.DISCOVERY_JOIN_VALIDATE_ACTION_NAME,
+            "mixed-version cluster join validation on cluster state with a different cluster uuid");
+    }
+
+    public void testJoinValidationRejectsMismatchedClusterUUID() {
+        assertJoinValidationRejectsMismatchedClusterUUID(JoinHelper.VALIDATE_JOIN_ACTION_NAME,
+            "join validation on cluster state with a different cluster uuid");
+    }
+
+    private void assertJoinValidationRejectsMismatchedClusterUUID(String actionName, String expectedMessage) {
+        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, actionName,
+            new ValidateJoinRequest(otherClusterState),
+            new ActionListenerResponseHandler<>(future, in -> TransportResponse.Empty.INSTANCE));
+        deterministicTaskQueue.runAllTasks();
+
+        final CoordinationStateRejectedException coordinationStateRejectedException
+            = expectThrows(CoordinationStateRejectedException.class, future::actionGet);
+        assertThat(coordinationStateRejectedException.getMessage(), containsString(expectedMessage));
+        assertThat(coordinationStateRejectedException.getMessage(), containsString(localClusterState.metaData().clusterUUID()));
+        assertThat(coordinationStateRejectedException.getMessage(), containsString(otherClusterState.metaData().clusterUUID()));
+    }
 }