Prevent version upgrade during remote migration (opensearch-project#1…

…3185)

Signed-off-by: Lakshya Taragi <lakshya.taragi@gmail.com>
(cherry picked from commit 2aad499)

server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java

@@ -31,6 +31,7 @@
 
 package org.opensearch.cluster.coordination;
 
+import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.LegacyESVersion;
 import org.opensearch.Version;
@@ -59,6 +60,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -68,6 +70,7 @@
 import static org.opensearch.cluster.decommission.DecommissionHelper.nodeCommissioned;
 import static org.opensearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.CompatibilityMode;
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.CompatibilityMode.MIXED;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.CompatibilityMode.STRICT;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
 
@@ -80,7 +83,7 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut
 
     private final AllocationService allocationService;
 
-    private final Logger logger;
+    private static Logger logger = LogManager.getLogger(JoinTaskExecutor.class);
     private final RerouteService rerouteService;
     private final TransportService transportService;
 
@@ -146,7 +149,7 @@ public JoinTaskExecutor(
         RemoteStoreNodeService remoteStoreNodeService
     ) {
         this.allocationService = allocationService;
-        this.logger = logger;
+        JoinTaskExecutor.logger = logger;
         this.rerouteService = rerouteService;
         this.transportService = transportService;
         this.remoteStoreNodeService = remoteStoreNodeService;
@@ -582,7 +585,18 @@ private static void ensureRemoteStoreNodesCompatibility(DiscoveryNode joiningNod
                 }
             }
         } else {
-            if (remoteStoreCompatibilityMode == CompatibilityMode.MIXED) {
+            if (MIXED.equals(remoteStoreCompatibilityMode)) {
+                if (joiningNode.getVersion().after(currentNodes.getMaxNodeVersion())) {
+                    String reason = String.format(
+                        Locale.ROOT,
+                        "remote migration : a node [%s] of higher version [%s] is not allowed to join a cluster with maximum version [%s]",
+                        joiningNode,
+                        joiningNode.getVersion(),
+                        currentNodes.getMaxNodeVersion()
+                    );
+                    logger.warn(reason);
+                    throw new IllegalStateException(reason);
+                }
                 if (joiningNode.isRemoteStoreNode()) {
                     Optional<DiscoveryNode> remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst();
                     remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode));

server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java

@@ -77,6 +77,8 @@
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
+import static org.opensearch.test.VersionUtils.allOpenSearchVersions;
+import static org.opensearch.test.VersionUtils.allVersions;
 import static org.opensearch.test.VersionUtils.maxCompatibleVersion;
 import static org.opensearch.test.VersionUtils.randomCompatibleVersion;
 import static org.opensearch.test.VersionUtils.randomVersion;
@@ -967,6 +969,64 @@ public void testUpdatesClusterStateWithMultiNodeClusterAndSameRepository() throw
         validateRepositoryMetadata(result.resultingState, clusterManagerNode, 2);
     }
 
+    public void testNodeJoinInMixedMode() {
+        Settings nodeSettings = Settings.builder().put(REMOTE_STORE_MIGRATION_EXPERIMENTAL, "true").build();
+        FeatureFlags.initializeFeatureFlags(nodeSettings);
+
+        List<Version> versions = allOpenSearchVersions();
+        assert versions.size() >= 2 : "test requires at least two open search versions";
+        Version baseVersion = versions.get(versions.size() - 2);
+        Version higherVersion = versions.get(versions.size() - 1);
+
+        DiscoveryNode currentNode1 = new DiscoveryNode(UUIDs.base64UUID(), buildNewFakeTransportAddress(), baseVersion);
+        DiscoveryNode currentNode2 = new DiscoveryNode(UUIDs.base64UUID(), buildNewFakeTransportAddress(), baseVersion);
+        DiscoveryNodes currentNodes = DiscoveryNodes.builder()
+            .add(currentNode1)
+            .localNodeId(currentNode1.getId())
+            .add(currentNode2)
+            .localNodeId(currentNode2.getId())
+            .build();
+
+        Settings mixedModeCompatibilitySettings = Settings.builder()
+            .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), RemoteStoreNodeService.CompatibilityMode.MIXED)
+            .build();
+
+        Metadata metadata = Metadata.builder().persistentSettings(mixedModeCompatibilitySettings).build();
+
+        // joining node of a higher version than the current nodes
+        DiscoveryNode joiningNode1 = new DiscoveryNode(
+            randomAlphaOfLength(10),
+            randomAlphaOfLength(10),
+            buildNewFakeTransportAddress(),
+            remoteStoreNodeAttributes(SEGMENT_REPO, TRANSLOG_REPO),
+            Collections.singleton(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE),
+            higherVersion
+        );
+        final IllegalStateException exception = expectThrows(
+            IllegalStateException.class,
+            () -> JoinTaskExecutor.ensureNodesCompatibility(joiningNode1, currentNodes, metadata)
+        );
+        String reason = String.format(
+            Locale.ROOT,
+            "remote migration : a node [%s] of higher version [%s] is not allowed to join a cluster with maximum version [%s]",
+            joiningNode1,
+            joiningNode1.getVersion(),
+            currentNodes.getMaxNodeVersion()
+        );
+        assertEquals(reason, exception.getMessage());
+
+        // joining node of the same version as the current nodes
+        DiscoveryNode joiningNode2 = new DiscoveryNode(
+            randomAlphaOfLength(10),
+            randomAlphaOfLength(10),
+            buildNewFakeTransportAddress(),
+            remoteStoreNodeAttributes(SEGMENT_REPO, TRANSLOG_REPO),
+            Collections.singleton(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE),
+            baseVersion
+        );
+        JoinTaskExecutor.ensureNodesCompatibility(joiningNode2, currentNodes, metadata);
+    }
+
     private void validateRepositoryMetadata(ClusterState updatedState, DiscoveryNode existingNode, int expectedRepositories)
         throws Exception {