introduce a setting to disable full cluster state from remote on term…

… mismatch

server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateTermVersionIT.java

@@ -40,6 +40,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING_KEY;
 import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.hamcrest.Matchers.is;
@@ -72,6 +73,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
             )
             .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, REMOTE_ROUTING_TABLE_REPO)
             .put(REMOTE_PUBLICATION_SETTING_KEY, true)
+            .put(REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING_KEY, true)
             .build();
     }
 

server/src/main/java/org/opensearch/action/support/clustermanager/term/TransportGetTermVersionAction.java

@@ -98,7 +98,7 @@ private GetTermVersionResponse buildResponse(GetTermVersionRequest request, Clus
         ClusterStateTermVersion termVersion = new ClusterStateTermVersion(state);
         if (discovery instanceof Coordinator) {
             Coordinator coordinator = (Coordinator) discovery;
-            if (coordinator.isRemotePublicationEnabled()) {
+            if (coordinator.canDownloadFullStateFromRemote()) {
                 return new GetTermVersionResponse(termVersion, coordinator.isRemotePublicationEnabled());
             }
         }

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

@@ -1906,4 +1906,13 @@ public boolean isRemotePublicationEnabled() {
         }
         return false;
     }
+
+    public boolean canDownloadFullStateFromRemote() {
+        if (remoteClusterStateService != null) {
+            return remoteClusterStateService.isRemotePublicationEnabled()
+                && remoteClusterStateService.canDownloadFromRemoteOnTermMismatch();
+        }
+        return false;
+    }
+
 }

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -738,6 +738,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING,
                 RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING,
                 RemoteClusterStateService.REMOTE_PUBLICATION_SETTING,
+                RemoteClusterStateService.REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING,
+
                 INDEX_METADATA_UPLOAD_TIMEOUT_SETTING,
                 GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING,
                 METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING,

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

@@ -129,6 +129,7 @@ public class RemoteClusterStateService implements Closeable {
      * Gates the functionality of remote publication.
      */
     public static final String REMOTE_PUBLICATION_SETTING_KEY = "cluster.remote_store.publication.enabled";
+    public static final String REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING_KEY = "cluster.remote_publication.download.term.mismatch.enabled";
 
     public static final Setting<Boolean> REMOTE_PUBLICATION_SETTING = Setting.boolSetting(
         REMOTE_PUBLICATION_SETTING_KEY,
@@ -137,6 +138,13 @@ public class RemoteClusterStateService implements Closeable {
         Property.Dynamic
     );
 
+    public static final Setting<Boolean> REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING = Setting.boolSetting(
+        REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING_KEY,
+        false,
+        Property.NodeScope,
+        Property.Dynamic
+    );
+
     /**
      * Used to specify if cluster state metadata should be published to remote store
      */
@@ -235,6 +243,9 @@ public static RemoteClusterStateValidationMode parseString(String mode) {
         + "indices, coordination metadata updated : [{}], settings metadata updated : [{}], templates metadata "
         + "updated : [{}], custom metadata updated : [{}], indices routing updated : [{}]";
     private volatile AtomicBoolean isPublicationEnabled;
+
+    private volatile AtomicBoolean downloadFromRemoteOnTermMismatch;
+
     private final String remotePathPrefix;
 
     private final RemoteClusterStateCache remoteClusterStateCache;
@@ -281,6 +292,8 @@ public RemoteClusterStateService(
                 && RemoteStoreNodeAttribute.isRemoteRoutingTableConfigured(settings)
         );
         clusterSettings.addSettingsUpdateConsumer(REMOTE_PUBLICATION_SETTING, this::setRemotePublicationSetting);
+        this.downloadFromRemoteOnTermMismatch = new AtomicBoolean(clusterSettings.get(REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING));
+        clusterSettings.addSettingsUpdateConsumer(REMOTE_DOWNLOAD_TERM_MISMATCH_SETTING, this::setRemoteDownloadOnTermMismatchSetting);
         this.remotePathPrefix = CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX.get(settings);
         this.remoteRoutingTableService = RemoteRoutingTableServiceFactory.getService(
             repositoriesService,
@@ -1124,6 +1137,14 @@ private void setRemotePublicationSetting(boolean remotePublicationSetting) {
         }
     }
 
+    private void setRemoteDownloadOnTermMismatchSetting(boolean remotePublicationSetting) {
+        this.downloadFromRemoteOnTermMismatch.set(remotePublicationSetting);
+    }
+
+    public boolean canDownloadFromRemoteOnTermMismatch() {
+        return this.downloadFromRemoteOnTermMismatch.get();
+    }
+
     // Package private for unit test
     RemoteRoutingTableService getRemoteRoutingTableService() {
         return this.remoteRoutingTableService;