diff --git a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/scale/searchonly/ScaleIndexIT.java b/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/scale/searchonly/ScaleIndexIT.java index 97d0f8565394d..d14d7d56562dd 100644 --- a/server/src/internalClusterTest/java/org/opensearch/action/admin/indices/scale/searchonly/ScaleIndexIT.java +++ b/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 diff --git a/server/src/main/java/org/opensearch/index/IndexService.java b/server/src/main/java/org/opensearch/index/IndexService.java index 65bcfdcc565c5..fdbee2e5f944e 100644 --- a/server/src/main/java/org/opensearch/index/IndexService.java +++ b/server/src/main/java/org/opensearch/index/IndexService.java @@ -1214,6 +1214,10 @@ public void addMetadataListener(Consumer 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;