Trigger search replica sync when scaling down to search-only

server/src/internalClusterTest/java/org/opensearch/action/admin/indices/scale/searchonly/ScaleIndexIT.java

@@ -16,9 +16,11 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.routing.IndexRoutingTable;
 import org.opensearch.cluster.routing.IndexShardRoutingTable;
+import org.opensearch.cluster.routing.Preference;
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.rest.RestStatus;
+import org.opensearch.index.IndexSettings;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
 import org.opensearch.test.InternalTestCluster;
@@ -52,6 +54,61 @@ public void testFullLifecycleWithoutSearchReplicas() throws Exception {
         testFullLifecycle(0);
     }
 
+    public void testScaleDownSearchReplicaCatchesUpWithFinalRemoteStoreState() throws Exception {
+        internalCluster().startClusterManagerOnlyNode();
+        internalCluster().startDataOnlyNode();
+        internalCluster().startSearchOnlyNode();
+
+        Settings specificSettings = Settings.builder()
+            .put(indexSettings())
+            .put(SETTING_NUMBER_OF_SHARDS, 1)
+            .put(SETTING_NUMBER_OF_REPLICAS, 0)
+            .put(SETTING_NUMBER_OF_SEARCH_REPLICAS, 1)
+            .build();
+
+        createIndex(TEST_INDEX, specificSettings);
+        ensureGreen(TEST_INDEX);
+
+        client().prepareIndex(TEST_INDEX)
+            .setId("baseline")
+            .setSource("field1", "baseline")
+            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+            .get();
+
+        assertBusy(() -> {
+            assertHitCount(client().prepareSearch(TEST_INDEX).setPreference(Preference.PRIMARY.type()).setSize(0).get(), 1);
+            assertHitCount(client().prepareSearch(TEST_INDEX).setPreference(Preference.SEARCH_REPLICA.type()).setSize(0).get(), 1);
+        });
+
+        assertAcked(
+            client().admin()
+                .indices()
+                .prepareUpdateSettings(TEST_INDEX)
+                .setSettings(Settings.builder().put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), "-1"))
+                .get()
+        );
+
+        client().prepareIndex(TEST_INDEX)
+            .setId("latest")
+            .setSource("field1", "latest")
+            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+            .get();
+
+        assertHitCount(client().prepareSearch(TEST_INDEX).setPreference(Preference.PRIMARY.type()).setSize(0).get(), 2);
+        assertHitCount(client().prepareSearch(TEST_INDEX).setPreference(Preference.SEARCH_REPLICA.type()).setSize(0).get(), 1);
+
+        assertAcked(client().admin().indices().prepareScaleSearchOnly(TEST_INDEX, true).get());
+        ensureGreen(TEST_INDEX);
+
+        assertBusy(
+            () -> {
+                assertHitCount(client().prepareSearch(TEST_INDEX).setPreference(Preference.SEARCH_REPLICA.type()).setSize(0).get(), 2);
+            },
+            10,
+            TimeUnit.SECONDS
+        );
+    }
+
     /**
      * Tests the full lifecycle of scaling an index down to search-only mode,
      * scaling search replicas while in search-only mode, verifying cluster health in

server/src/main/java/org/opensearch/index/IndexService.java

@@ -1214,6 +1214,10 @@ public void addMetadataListener(Consumer<IndexMetadata> listener) {
 
     @Override
     public synchronized void updateMetadata(final IndexMetadata currentIndexMetadata, final IndexMetadata newIndexMetadata) {
+        final boolean wasSearchOnly = currentIndexMetadata != null
+            && IndexMetadata.INDEX_BLOCKS_SEARCH_ONLY_SETTING.get(currentIndexMetadata.getSettings());
+        final boolean isSearchOnly = IndexMetadata.INDEX_BLOCKS_SEARCH_ONLY_SETTING.get(newIndexMetadata.getSettings());
+        final boolean becameSearchOnly = wasSearchOnly == false && isSearchOnly;
         final boolean updateIndexSettings = indexSettings.updateIndexMetadata(newIndexMetadata);
 
         if (Assertions.ENABLED && currentIndexMetadata != null) {
@@ -1245,18 +1249,21 @@ public synchronized void updateMetadata(final IndexMetadata currentIndexMetadata
             }
             onRefreshIntervalChange();
             updateFsyncTaskIfNecessary();
-            updateReplicationTask();
+            updateReplicationTask(becameSearchOnly);
             updatePublishReferencedSegmentsTask();
         }
 
         metadataListeners.forEach(c -> c.accept(newIndexMetadata));
     }
 
-    private void updateReplicationTask() {
+    private void updateReplicationTask(boolean forceSync) {
         try {
             asyncReplicationTask.close();
         } finally {
             asyncReplicationTask = new AsyncReplicationTask(this);
+            if (forceSync) {
+                asyncReplicationTask.forceSyncSegments();
+            }
         }
     }
 
@@ -1624,6 +1631,25 @@ protected void runInternal() {
             }
         }
 
+        void forceSyncSegments() {
+            if (mustReschedule() == false) {
+                return;
+            }
+            threadPool.executor(getThreadPool()).execute(new AbstractRunnable() {
+                @Override
+                public void onFailure(Exception e) {
+                    logger.warn(() -> new ParameterizedMessage("failed to run forced task {}", AsyncReplicationTask.this), e);
+                }
+
+                @Override
+                protected void doRun() {
+                    if (mustReschedule()) {
+                        indexService.maybeSyncSegments(true);
+                    }
+                }
+            });
+        }
+
         @Override
         protected String getThreadPool() {
             return ThreadPool.Names.GENERIC;