Add publication flag and remote routing table check

Signed-off-by: Himshikha Gupta <himshikh@amazon.com>

Author: Himshikha Gupta

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

@@ -52,6 +52,7 @@
 import java.util.Set;
 
 import static org.opensearch.cluster.coordination.Coordinator.ZEN1_BWC_TERM;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled;
 
 /**
@@ -78,7 +79,7 @@ public class CoordinationState {
     private long lastPublishedVersion;
     private VotingConfiguration lastPublishedConfiguration;
     private VoteCollection publishVotes;
-    private final boolean isRemoteStateEnabled;
+    private final boolean isRemotePublicationEnabled;
 
     public CoordinationState(
         DiscoveryNode localNode,
@@ -101,11 +102,11 @@ public CoordinationState(
             .getLastAcceptedState()
             .getLastAcceptedConfiguration();
         this.publishVotes = new VoteCollection();
-        this.isRemoteStateEnabled = isRemoteStoreClusterStateEnabled(settings);
+        this.isRemotePublicationEnabled = isRemoteStoreClusterStateEnabled(settings) && isRemoteRoutingTableEnabled(settings);
     }
 
-    public boolean isRemoteStateEnabled() {
-        return isRemoteStateEnabled;
+    public boolean isRemotePublicationEnabled() {
+        return isRemotePublicationEnabled;
     }
 
     public long getCurrentTerm() {
@@ -579,7 +580,7 @@ public void handlePrePublish(ClusterState clusterState) {
         // This is to ensure the remote store is the single source of truth for current state. Even if the current node
         // goes down after sending the cluster state to other nodes, we should be able to read the remote state and
         // recover the cluster.
-        if (isRemoteStateEnabled) {
+        if (isRemotePublicationEnabled) {
             assert persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE) != null : "Remote state has not been initialized";
             persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).setLastAcceptedState(clusterState);
         }
@@ -590,7 +591,7 @@ public void handlePrePublish(ClusterState clusterState) {
      */
     public void handlePreCommit() {
         // Publishing the committed state to remote store before sending apply commit to other nodes.
-        if (isRemoteStateEnabled) {
+        if (isRemotePublicationEnabled) {
             assert persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE) != null : "Remote state has not been initialized";
             persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE).markLastAcceptedStateAsCommitted();
         }

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

@@ -1323,7 +1323,8 @@ assert getLocalNode().equals(clusterState.getNodes().get(getLocalNode().getId())
                     + clusterState;
 
                 final PublicationTransportHandler.PublicationContext publicationContext = publicationHandler.newPublicationContext(
-                    clusterChangedEvent, coordinationState.get().isRemoteStateEnabled()
+                    clusterChangedEvent,
+                    coordinationState.get().isRemotePublicationEnabled()
                 );
 
                 final PublishRequest publishRequest = coordinationState.get().handleClientValue(clusterState);

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

@@ -279,8 +279,8 @@ private PublishWithJoinResponse acceptState(ClusterState incomingState) {
         return handlePublishRequest.apply(new PublishRequest(incomingState));
     }
 
-    public PublicationContext newPublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemoteStateEnabled) {
-        final PublicationContext publicationContext = new PublicationContext(clusterChangedEvent, isRemoteStateEnabled);
+    public PublicationContext newPublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemotePublicationEnabled) {
+        final PublicationContext publicationContext = new PublicationContext(clusterChangedEvent, isRemotePublicationEnabled);
 
         // Build the serializations we expect to need now, early in the process, so that an error during serialization fails the publication
         // straight away. This isn't watertight since we send diffs on a best-effort basis and may fall back to sending a full state (and
@@ -327,12 +327,12 @@ public class PublicationContext {
         private final Map<Version, BytesReference> serializedDiffs = new HashMap<>();
         private final boolean sendRemoteState;
 
-        PublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemoteStateEnabled) {
+        PublicationContext(ClusterChangedEvent clusterChangedEvent, boolean isRemotePublicationEnabled) {
             discoveryNodes = clusterChangedEvent.state().nodes();
             newState = clusterChangedEvent.state();
             previousState = clusterChangedEvent.previousState();
             sendFullVersion = previousState.getBlocks().disableStatePersistence();
-            sendRemoteState = isRemoteStateEnabled;
+            sendRemoteState = isRemotePublicationEnabled;
         }
 
         void buildDiffAndSerializeStates() {

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

@@ -8,11 +8,12 @@
 
 package org.opensearch.cluster.coordination;
 
-import java.io.IOException;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 
+import java.io.IOException;
+
 public class RemotePublishRequest extends TermVersionRequest {
 
     // todo Do we need cluster name and UUID ?

server/src/main/java/org/opensearch/cluster/node/DiscoveryNode.java

@@ -61,7 +61,9 @@
 import java.util.stream.Stream;
 
 import static org.opensearch.node.NodeRoleSettings.NODE_ROLES_SETTING;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_NODE_ATTRIBUTE_KEY_PREFIX;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
 
 /**
  * A discovery node represents a node that is part of the cluster.
@@ -470,6 +472,25 @@ public boolean isRemoteStoreNode() {
         return this.getAttributes().keySet().stream().anyMatch(key -> key.startsWith(REMOTE_STORE_NODE_ATTRIBUTE_KEY_PREFIX));
     }
 
+    /**
+     * Returns whether the node is a remote cluster state enabled node.
+     * @return true if the node contains remote cluster state and remote routing table node attributes, false otherwise
+     */
+    public boolean isRemoteStateNode() {
+        return isRemoteClusterStateEnabled() && isRemoteRoutingTableEnabled();
+    }
+
+    private boolean isRemoteClusterStateEnabled() {
+        return this.getAttributes()
+            .keySet()
+            .stream()
+            .anyMatch(key -> (key.equals(REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY)));
+    }
+
+    private boolean isRemoteRoutingTableEnabled() {
+        return this.getAttributes().keySet().stream().anyMatch(key -> key.equals(REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY));
+    }
+
     /**
      * Returns a set of all the roles that the node has. The roles are returned in sorted order by the role name.
      * <p>