diff --git a/server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java b/server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java index aaf2517b800d5..0ad2b511c4417 100644 --- a/server/src/main/java/org/opensearch/cluster/coordination/JoinTaskExecutor.java +++ b/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 remoteDN = existingNodes.stream().filter(DiscoveryNode::isRemoteStoreNode).findFirst(); remoteDN.ifPresent(discoveryNode -> ensureRemoteStoreNodesCompatibility(joiningNode, discoveryNode)); diff --git a/server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java b/server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java index 76644e0309d18..8dc7f3a472b11 100644 --- a/server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java +++ b/server/src/test/java/org/opensearch/cluster/coordination/JoinTaskExecutorTests.java @@ -77,6 +77,7 @@ 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.maxCompatibleVersion; import static org.opensearch.test.VersionUtils.randomCompatibleVersion; import static org.opensearch.test.VersionUtils.randomVersion; @@ -967,6 +968,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 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 {