From 7157c119b6fc08dba2eba254a0069220b80b5f27 Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Wed, 26 Aug 2026 11:28:35 +0000 Subject: [PATCH 01/10] Restore size-based periodic flush on remote-store shards On remote-store enabled shards the translog based periodic flush condition (index.translog.flush_threshold_size) is ineffective: RemoteFsTranslog.getMinUnreferencedSeqNoInSegments() returns minSeqNoToKeep, which is advanced by every successful segments upload, so the referenced translog size never grows. Shards can therefore go arbitrarily long without a commit, pinning the soft-deletes retention policy to a stale commit point and accumulating delete tombstones through merges. This restores an equivalent size-based flush signal scoped to the broken code path only. RemoteStoreRefreshListener already builds a map of post-refresh local segment file names to sizes for every upload cycle; after a successful segments sync (the same callback that trims the translog) it now publishes the total size of segment files not yet referenced by the last commit point to the engine, stamped with the commit generation it was computed against. The engine flushes once that size breaches the threshold. A stale stamp (e.g. right after a flush) can never re-trigger a flush, and the computation adds no I/O on either the write path (plain volatile read) or the refresh path (in-memory set diff over the existing map). Introduces two dynamic index settings: - index.remote_store.flush_on_uncommitted_segments.enabled (default true) controls the condition; when disabled the accounting is neither published nor consulted - index.remote_store.flush_on_uncommitted_segments.threshold_size defaults to the current value of the index's index.translog.flush_threshold_size via setting fallback Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> --- .../common/settings/IndexScopedSettings.java | 2 + .../org/opensearch/index/IndexSettings.java | 67 ++++++++++++++++ .../index/engine/InternalEngine.java | 77 +++++++++++++++++++ .../shard/RemoteStoreRefreshListener.java | 8 ++ .../index/engine/InternalEngineTests.java | 60 +++++++++++++++ .../remotestore/RemoteStoreCoreTestCase.java | 24 ++++++ 6 files changed, 238 insertions(+) diff --git a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java index b59e67749782b..ef7f4ed324e7d 100644 --- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java @@ -250,6 +250,8 @@ public final class IndexScopedSettings extends AbstractScopedSettings { // Settings for remote translog IndexSettings.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING, IndexSettings.INDEX_REMOTE_TRANSLOG_KEEP_EXTRA_GEN_SETTING, + IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING, + IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING, // Settings for remote store enablement IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING, diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java index 4b9ca27890e97..854c2b919938f 100644 --- a/server/src/main/java/org/opensearch/index/IndexSettings.java +++ b/server/src/main/java/org/opensearch/index/IndexSettings.java @@ -881,6 +881,34 @@ public static IndexMergePolicy fromString(String text) { Property.IndexScope ); + /** + * Controls whether a remote-store shard periodically flushes based on the total size of segment bytes that are + * refreshed and uploaded to the remote store but not yet referenced by the last commit point. On remote-store + * shards the translog based flush threshold ({@code index.translog.flush_threshold_size}) is ineffective because + * uploaded translog generations are trimmed continuously; this condition restores an equivalent size-based flush + * signal driven by uncommitted segment bytes. + */ + public static final Setting INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING = Setting.boolSetting( + "index.remote_store.flush_on_uncommitted_segments.enabled", + true, + Property.Dynamic, + Property.IndexScope + ); + + /** + * The minimum total size of segment bytes not yet referenced by the last commit point which triggers a flush on a + * remote-store shard. Only takes effect when + * {@link #INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING} is enabled. Defaults to the current + * value of {@code index.translog.flush_threshold_size} of the index. + */ + public static final Setting INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING = Setting + .byteSizeSetting( + "index.remote_store.flush_on_uncommitted_segments.threshold_size", + INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING, + Property.Dynamic, + Property.IndexScope + ); + public static final Setting INDEX_CONTEXT_CREATED_VERSION = Setting.longSetting( "index.context.created_version", 0, @@ -943,6 +971,8 @@ public static IndexMergePolicy fromString(String text) { // For warm index we would partially store files in local. private final boolean isWarmIndex; private volatile TimeValue remoteTranslogUploadBufferInterval; + private volatile boolean flushOnUncommittedSegmentsEnabled; + private volatile ByteSizeValue flushOnUncommittedSegmentsThresholdSize; private volatile String remoteStoreTranslogRepository; private volatile String remoteStoreRepository; private volatile String remoteStoreSegmentPathPrefix; @@ -1173,6 +1203,10 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti remoteStoreTranslogRepository = settings.get(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY); remoteTranslogUploadBufferInterval = INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING.get(settings); + flushOnUncommittedSegmentsEnabled = scopedSettings.get(INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING); + flushOnUncommittedSegmentsThresholdSize = scopedSettings.get( + INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING + ); remoteStoreRepository = settings.get(IndexMetadata.SETTING_REMOTE_SEGMENT_STORE_REPOSITORY); this.remoteTranslogKeepExtraGen = INDEX_REMOTE_TRANSLOG_KEEP_EXTRA_GEN_SETTING.get(settings); String rawPrefix = IndexMetadata.INDEX_REMOTE_STORE_SEGMENT_PATH_PREFIX.get(settings); @@ -1389,6 +1423,14 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti this::setRemoteTranslogUploadBufferInterval ); scopedSettings.addSettingsUpdateConsumer(INDEX_REMOTE_TRANSLOG_KEEP_EXTRA_GEN_SETTING, this::setRemoteTranslogKeepExtraGen); + scopedSettings.addSettingsUpdateConsumer( + INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING, + this::setFlushOnUncommittedSegmentsEnabled + ); + scopedSettings.addSettingsUpdateConsumer( + INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING, + this::setFlushOnUncommittedSegmentsThresholdSize + ); this.autoForcemergeEnabled = scopedSettings.get(INDEX_AUTO_FORCE_MERGES_ENABLED); scopedSettings.addSettingsUpdateConsumer(INDEX_AUTO_FORCE_MERGES_ENABLED, this::setAutoForcemergeEnabled); scopedSettings.addSettingsUpdateConsumer(INDEX_DOC_ID_FUZZY_SET_ENABLED_SETTING, this::setEnableFuzzySetForDocId); @@ -1774,6 +1816,31 @@ public int getRemoteTranslogExtraKeep() { return remoteTranslogKeepExtraGen; } + /** + * Returns whether the periodic flush condition based on uncommitted segment bytes is enabled for this + * (remote-store) index. + */ + public boolean isFlushOnUncommittedSegmentsEnabled() { + return flushOnUncommittedSegmentsEnabled; + } + + /** + * Returns the threshold size of segment bytes not yet referenced by the last commit point when to forcefully + * flush a remote-store shard. Only takes effect when {@link #isFlushOnUncommittedSegmentsEnabled()}. Defaults to + * the current value of {@code index.translog.flush_threshold_size} of this index. + */ + public ByteSizeValue getFlushOnUncommittedSegmentsThresholdSize() { + return flushOnUncommittedSegmentsThresholdSize; + } + + private void setFlushOnUncommittedSegmentsEnabled(boolean flushOnUncommittedSegmentsEnabled) { + this.flushOnUncommittedSegmentsEnabled = flushOnUncommittedSegmentsEnabled; + } + + private void setFlushOnUncommittedSegmentsThresholdSize(ByteSizeValue flushOnUncommittedSegmentsThresholdSize) { + this.flushOnUncommittedSegmentsThresholdSize = flushOnUncommittedSegmentsThresholdSize; + } + /** * Returns true iff the remote translog buffer interval setting exists or in other words is explicitly set. */ diff --git a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java index 997161ffb60c2..f8c4ef458d6bc 100644 --- a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java +++ b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java @@ -37,6 +37,7 @@ import org.apache.lucene.document.NumericDocValuesField; import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.IndexCommit; +import org.apache.lucene.index.IndexFileNames; import org.apache.lucene.index.IndexWriter; import org.apache.lucene.index.IndexWriterConfig; import org.apache.lucene.index.LeafReaderContext; @@ -118,6 +119,7 @@ import java.io.IOException; import java.util.Arrays; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; @@ -161,6 +163,24 @@ public class InternalEngine extends Engine { protected final AtomicBoolean shouldPeriodicallyFlushAfterBigMerge = new AtomicBoolean(false); protected final NumericDocValuesField softDeletesField = Lucene.newSoftDeletesField(); + /** + * Size of segment bytes not yet referenced by the last commit point on a remote-store shard, published by the + * remote segment upload path after every successful segments sync (see RemoteStoreRefreshListener). Stamped with + * the commit generation it was computed against: a value whose generation does not match the current last commit + * is stale (e.g. right after a flush) and is ignored until the next sync republishes. + */ + private volatile UncommittedSegmentBytes uncommittedSegmentBytes; + + private static final class UncommittedSegmentBytes { + private final long bytes; + private final long committedInfosGeneration; + + private UncommittedSegmentBytes(long bytes, long committedInfosGeneration) { + this.bytes = bytes; + this.committedInfosGeneration = committedInfosGeneration; + } + } + // A uid (in the form of BytesRef) to the version map // we use the hashed variant since we iterate over it and check removal and additions on existing keys protected final LiveVersionMap versionMap = new LiveVersionMap(); @@ -1496,6 +1516,9 @@ public boolean shouldPeriodicallyFlush() { if (shouldPeriodicallyFlushAfterBigMerge.get()) { return true; } + if (shouldFlushOnUncommittedSegmentBytes()) { + return true; + } final long localCheckpointOfLastCommit = Long.parseLong( lastCommittedSegmentInfos.userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY) ); @@ -1505,6 +1528,60 @@ public boolean shouldPeriodicallyFlush() { ); } + /** + * Updates the uncommitted segment bytes accounting for this (remote-store) shard. Invoked by the remote segment + * upload path after a successful segments sync, passing the post-refresh local segment file sizes. The uncommitted + * bytes are computed as the total size of local segment files that are not referenced by the last commit point, + * which includes newly written segments as well as updated per-segment files (e.g. live docs and doc-values + * updates) of already committed segments. The value is stamped with the generation of the commit point it was + * computed against, so a commit implicitly invalidates it. + * + * @param localSegmentsSizeMap post-refresh local segment file names mapped to their sizes in bytes + */ + public void updateUncommittedSegmentBytes(Map localSegmentsSizeMap) { + final SegmentInfos committedInfos = this.lastCommittedSegmentInfos; + if (committedInfos == null) { + return; + } + final Set committedFiles; + try { + committedFiles = new HashSet<>(committedInfos.files(false)); + } catch (IOException e) { + // best-effort accounting, a failed computation must never affect the segment upload path + logger.debug("failed to compute uncommitted segment bytes", e); + return; + } + long bytes = 0; + for (Map.Entry file : localSegmentsSizeMap.entrySet()) { + if (committedFiles.contains(file.getKey()) == false && file.getKey().startsWith(IndexFileNames.SEGMENTS) == false) { + bytes += file.getValue(); + } + } + uncommittedSegmentBytes = new UncommittedSegmentBytes(bytes, committedInfos.getGeneration()); + } + + /** + * Checks whether the uncommitted segment bytes published by the remote segment upload path breach + * {@link IndexSettings#INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING}. Only ever + * effective on remote-store shards since the accounting is only published there, and only when + * {@link IndexSettings#INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING} is enabled; the stamped + * commit generation must match the current last commit so that stale values (e.g. right after a flush, before the + * next successful segments sync) can never re-trigger a flush. + */ + private boolean shouldFlushOnUncommittedSegmentBytes() { + final UncommittedSegmentBytes current = this.uncommittedSegmentBytes; + if (current == null) { + return false; + } + final IndexSettings indexSettings = config().getIndexSettings(); + if (indexSettings.isFlushOnUncommittedSegmentsEnabled() == false) { + return false; + } + return current.bytes > 0 + && current.bytes >= indexSettings.getFlushOnUncommittedSegmentsThresholdSize().getBytes() + && current.committedInfosGeneration == lastCommittedSegmentInfos.getGeneration(); + } + @Override public void flush(boolean force, boolean waitIfOngoing) throws EngineException { ensureOpen(); diff --git a/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java b/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java index a7a4406e37abd..30e2158ae4d0d 100644 --- a/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java +++ b/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java @@ -436,6 +436,14 @@ private void onSuccessfulSegmentsSync( resetBackOffDelayIterator(); // Set the minimum sequence number for keeping translog indexShard.getIndexer().translogManager().setMinSeqNoToKeep(lastRefreshedCheckpoint + 1); + // The above trimming makes the translog-size based periodic flush condition ineffective on remote-store + // shards, hence publish the size of segment bytes not yet referenced by the last commit point so that the + // engine can flush once they breach the configured threshold. + if (indexShard.indexSettings().isFlushOnUncommittedSegmentsEnabled() + && indexShard.getIndexer() instanceof EngineBackedIndexer engineBacked + && engineBacked.getEngine() instanceof InternalEngine internalEngine) { + internalEngine.updateUncommittedSegmentBytes(localFileSizeMap); + } // Publishing the new checkpoint which is used for remote store + segrep indexes checkpointPublisher.publish(indexShard, checkpoint); logger.debug("onSuccessfulSegmentsSync lastRefreshedCheckpoint={} checkpoint={}", lastRefreshedCheckpoint, checkpoint); diff --git a/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java b/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java index 6cefab864c960..e4fa8bf92f28f 100644 --- a/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java +++ b/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java @@ -6812,6 +6812,66 @@ public void testShouldPeriodicallyFlushAfterMerge() throws Exception { assertThat(engine.shouldPeriodicallyFlush(), equalTo(false)); } + public void testShouldPeriodicallyFlushOnUncommittedSegmentBytes() throws Exception { + assertThat("Empty engine does not need flushing", engine.shouldPeriodicallyFlush(), equalTo(false)); + ParsedDocument doc = testParsedDocument("0", null, testDocumentWithTextField(), SOURCE, null); + engine.index(indexForDoc(doc)); + engine.refresh("test"); + assertThat("Nothing published yet", engine.shouldPeriodicallyFlush(), equalTo(false)); + + // simulate the remote segment upload path publishing the post-refresh local file sizes + final Map localSegmentsSizeMap = new HashMap<>(); + try (GatedCloseable snapshot = engine.getSegmentInfosSnapshot()) { + for (String file : snapshot.get().files(false)) { + localSegmentsSizeMap.put(file, engine.store.directory().fileLength(file)); + } + } + engine.updateUncommittedSegmentBytes(localSegmentsSizeMap); + assertThat( + "Uncommitted bytes below the default threshold inherited from the translog flush threshold", + engine.shouldPeriodicallyFlush(), + equalTo(false) + ); + + final IndexSettings indexSettings = engine.config().getIndexSettings(); + updateIndexSettings( + indexSettings, + Settings.builder().put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + ); + assertThat("Uncommitted bytes breach the lowered threshold", engine.shouldPeriodicallyFlush(), equalTo(true)); + + updateIndexSettings( + indexSettings, + Settings.builder() + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING.getKey(), false) + ); + assertThat("Disabling the condition takes effect immediately", engine.shouldPeriodicallyFlush(), equalTo(false)); + + updateIndexSettings( + indexSettings, + Settings.builder() + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING.getKey(), true) + ); + assertThat("Re-enabling restores the condition", engine.shouldPeriodicallyFlush(), equalTo(true)); + + engine.flush(); + assertThat("Stale commit generation stamp is ignored after flush", engine.shouldPeriodicallyFlush(), equalTo(false)); + + // republishing against the new commit computes zero uncommitted bytes, hence no flush loop + engine.updateUncommittedSegmentBytes(localSegmentsSizeMap); + assertThat("All published files are committed now", engine.shouldPeriodicallyFlush(), equalTo(false)); + } + + private static void updateIndexSettings(IndexSettings indexSettings, Settings.Builder settingsBuilder) { + indexSettings.updateIndexMetadata( + IndexMetadata.builder(indexSettings.getIndexMetadata()) + .settings(Settings.builder().put(indexSettings.getSettings()).put(settingsBuilder.build())) + .build() + ); + } + public void testStressShouldPeriodicallyFlush() throws Exception { final long flushThreshold = randomLongBetween(120, 5000); final long generationThreshold = randomLongBetween(1000, 5000); diff --git a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java index d34db204a112f..54179dd0401c2 100644 --- a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java +++ b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java @@ -138,6 +138,30 @@ private void testPeerRecovery(int numberOfIterations, boolean invokeFlush) throw ); } + public void testPeriodicFlushOnUncommittedSegmentBytes() throws Exception { + String dataNode = internalCluster().startNodes(1).get(0); + // lower the uncommitted segment bytes threshold so that any refreshed but uncommitted segment breaches it; + // the condition itself is enabled by default + createIndex( + INDEX_NAME, + Settings.builder() + .put(remoteStoreIndexSettings(0)) + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .build() + ); + ensureGreen(INDEX_NAME); + IndexShard indexShard = getIndexShard(dataNode, INDEX_NAME); + assertEquals(0, indexShard.flushStats().getPeriodic()); + // each iteration drives one refresh (and hence one remote segments sync publishing the uncommitted bytes) + // followed by a write operation which polls the periodic flush condition and triggers the async flush + assertBusy(() -> { + indexSingleDoc(INDEX_NAME); + refresh(INDEX_NAME); + indexSingleDoc(INDEX_NAME); + assertThat(indexShard.flushStats().getPeriodic(), greaterThan(0L)); + }, 30, TimeUnit.SECONDS); + } + public void testRemoteStoreIndexCreationAndDeletionWithReferencedStore() throws InterruptedException, ExecutionException { String dataNode = internalCluster().startNodes(1).get(0); createIndex(INDEX_NAME, remoteStoreIndexSettings(0)); From 4f821321fb4799719fd69d1b9b9f9b0a9834820f Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Thu, 27 Aug 2026 09:18:14 +0000 Subject: [PATCH 02/10] Add integration test for update-heavy workloads Adds an end-to-end integration test reproducing the scenario from issue #22802: an update-heavy workload with regular refreshes on a remote-store shard. Asserts that the uncommitted-segment-bytes condition fires periodic flushes, the safe commit advances past the workload, and a non-flushing expunge-deletes merge can reclaim the soft-delete tombstones. The test pins index.soft_deletes.retention.operations=0 (the random index template may inject values that retain the whole workload) and uses a zero translog buffer interval plus a fast-tracking retention lease so each iteration completes in about 1.5 seconds. Verified across 50 randomized seeds. Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> --- .../remotestore/RemoteStoreCoreTestCase.java | 108 +++++++++++++++++- 1 file changed, 107 insertions(+), 1 deletion(-) diff --git a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java index 54179dd0401c2..5f747bc074d76 100644 --- a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java +++ b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java @@ -16,6 +16,7 @@ import org.opensearch.action.admin.indices.flush.FlushRequest; import org.opensearch.action.admin.indices.recovery.RecoveryResponse; import org.opensearch.action.admin.indices.settings.put.UpdateSettingsRequest; +import org.opensearch.action.bulk.BulkRequestBuilder; import org.opensearch.action.index.IndexResponse; import org.opensearch.action.search.SearchPhaseExecutionException; import org.opensearch.cluster.health.ClusterHealthStatus; @@ -27,7 +28,9 @@ import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.concurrent.BufferedAsyncIOProcessor; +import org.opensearch.index.IndexService; import org.opensearch.index.IndexSettings; +import org.opensearch.index.seqno.SequenceNumbers; import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.IndexShardClosedException; import org.opensearch.index.translog.Translog; @@ -41,6 +44,7 @@ import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.snapshots.SnapshotInfo; import org.opensearch.snapshots.SnapshotState; +import org.opensearch.test.InternalSettingsPlugin; import org.opensearch.test.InternalTestCluster; import org.opensearch.test.OpenSearchIntegTestCase; import org.opensearch.test.transport.MockTransportService; @@ -59,6 +63,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -75,7 +80,9 @@ import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount; import static org.hamcrest.Matchers.comparesEqualTo; import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.lessThan; import static org.hamcrest.Matchers.oneOf; @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) @@ -85,7 +92,8 @@ public class RemoteStoreCoreTestCase extends RemoteStoreBaseIntegTestCase { @Override protected Collection> nodePlugins() { - return Stream.concat(super.nodePlugins().stream(), Stream.of(MockTransportService.TestPlugin.class)).collect(Collectors.toList()); + return Stream.concat(super.nodePlugins().stream(), Stream.of(MockTransportService.TestPlugin.class, InternalSettingsPlugin.class)) + .collect(Collectors.toList()); } @Override @@ -162,6 +170,104 @@ public void testPeriodicFlushOnUncommittedSegmentBytes() throws Exception { }, 30, TimeUnit.SECONDS); } + /** + * End-to-end reproduction of the scenario in issue #22802: an update-heavy workload with regular refreshes on a + * remote-store shard. Every successful segments upload advances minSeqNoToKeep which suppresses the translog + * based flush condition, so without the uncommitted-segment-bytes condition no periodic flush ever happens, the + * safe commit never advances, and the soft-deleted documents produced by the updates stay protected from merge + * reclamation indefinitely. This test asserts the full recovery chain: periodic flushes fire, the safe commit + * advances, and a merge that does NOT flush (flush=false) is able to reclaim the tombstones. + */ + public void testPeriodicFlushReclaimsSoftDeletesUnderUpdateHeavyWorkload() throws Exception { + final int docCount = 20; + final int rounds = 8; + String dataNode = internalCluster().startNodes(1).get(0); + createIndex( + INDEX_NAME, + Settings.builder() + .put(remoteStoreIndexSettings(0)) + // refreshes are driven manually by the workload below + .put("index.refresh_interval", "-1") + // do not make writes wait on the translog upload batching window + .put(IndexSettings.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING.getKey(), "0ms") + // let the primary's own peer-recovery retention lease follow the global checkpoint at test speed; + // otherwise the renewal gate (half the 12h lease period) keeps retention pinned at the initial + // checkpoint for the whole test regardless of commits + .put(IndexSettings.INDEX_SOFT_DELETES_RETENTION_LEASE_PERIOD_SETTING.getKey(), "0ms") + .put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "1s") + // pin the (suppressed anyway) translog condition high so any observed periodic flush is attributable + // to the uncommitted-segment-bytes condition only + .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "1gb") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "5kb") + // any segment with at least one reclaimable delete qualifies for the expunge-deletes merge below, + // independent of the randomized merge policy defaults + .put("index.merge.policy.expunge_deletes_allowed", "0.0") + // the random index template may set retention.operations up to 1000 which would retain the whole + // workload below the global checkpoint no matter how far the safe commit advances + .put(IndexSettings.INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING.getKey(), 0) + .build() + ); + ensureGreen(INDEX_NAME); + IndexShard indexShard = getIndexShard(dataNode, INDEX_NAME); + assertEquals(0, indexShard.flushStats().getPeriodic()); + + // overwrite the same document ids round after round; every round turns the previous round's documents into + // soft-deleted tombstones, and the refresh drives a remote segments upload + for (int round = 0; round < rounds; round++) { + BulkRequestBuilder bulk = client().prepareBulk(); + for (int i = 0; i < docCount; i++) { + bulk.add( + client().prepareIndex(INDEX_NAME) + .setId(String.valueOf(i)) + .setSource("field", "value-" + round + "-" + randomAlphaOfLength(100)) + ); + } + assertFalse(bulk.get().hasFailures()); + refresh(INDEX_NAME); + } + + // the periodic flush condition is polled on write operations and the counter is republished on each + // successful upload, so nudge with write+refresh cycles until the flushes have carried the safe commit past + // the whole workload (the flush is async, so a single trip is not enough on its own) + final long maxSeqNoAfterWorkload = indexShard.seqNoStats().getMaxSeqNo(); + assertBusy(() -> { + client().prepareIndex(INDEX_NAME).setId("0").setSource("field", randomAlphaOfLength(100)).get(); + refresh(INDEX_NAME); + assertThat(indexShard.flushStats().getPeriodic(), greaterThan(0L)); + long committedCheckpoint = Long.parseLong(indexShard.commitStats().getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)); + assertThat(committedCheckpoint, greaterThanOrEqualTo(maxSeqNoAfterWorkload)); + }, 30, TimeUnit.SECONDS); + + // a merge WITHOUT a flush can now reclaim tombstones below the advanced safe commit. Without the fix the + // soft-deletes floor never moves off the initial commit, nothing is reclaimable, and the count can never + // drop below the full tombstone pile -- so ANY reclamation proves the chain end to end. Each retry keeps + // writing so the retention lease and the async flush pipeline keep advancing. + final long tombstonesCreated = (long) (rounds - 1) * docCount; + final AtomicInteger nudge = new AtomicInteger(); + assertBusy(() -> { + client().prepareIndex(INDEX_NAME).setId("nudge-" + nudge.incrementAndGet()).setSource("field", "x").get(); + refresh(INDEX_NAME); + assertEquals( + 0, + client().admin().indices().prepareForceMerge(INDEX_NAME).setOnlyExpungeDeletes(true).setFlush(false).get().getFailedShards() + ); + refresh(INDEX_NAME); + long deletedDocs = client().admin().indices().prepareStats(INDEX_NAME).get().getPrimaries().getDocs().getDeleted(); + assertThat( + "deleted=" + + deletedDocs + + " commitCkpt=" + + indexShard.commitStats().getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY) + + " gcp=" + + indexShard.getLastSyncedGlobalCheckpoint() + + " leases=" + + indexShard.getRetentionLeases(), + deletedDocs, + lessThan(tombstonesCreated) + ); + }, 30, TimeUnit.SECONDS); + } + public void testRemoteStoreIndexCreationAndDeletionWithReferencedStore() throws InterruptedException, ExecutionException { String dataNode = internalCluster().startNodes(1).get(0); createIndex(INDEX_NAME, remoteStoreIndexSettings(0)); From 3df1ff1ab981647fa0b0523867f8e9d3e9478c7f Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Thu, 27 Aug 2026 09:40:49 +0000 Subject: [PATCH 03/10] Address review: guard null commit infos in flush check Snapshot lastCommittedSegmentInfos into a local in shouldFlushOnUncommittedSegmentBytes() so the null check and the generation comparison observe the same commit point, matching the guard already present on the publish side. Also document why the publish-side race window is benign: the stamp and the file set come from one snapshot, so a flush landing between snapshot and publish only makes the stamp unmatchable (suppressing the trigger until the next successful segments sync), never a spurious flush. Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> --- .../org/opensearch/index/engine/InternalEngine.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java index f8c4ef458d6bc..db4bc20410150 100644 --- a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java +++ b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java @@ -1539,6 +1539,11 @@ public boolean shouldPeriodicallyFlush() { * @param localSegmentsSizeMap post-refresh local segment file names mapped to their sizes in bytes */ public void updateUncommittedSegmentBytes(Map localSegmentsSizeMap) { + // the file set and the generation stamp are both taken from this single snapshot, so the published value is + // always internally consistent. If a flush lands between this snapshot and the publish below, the stamp no + // longer matches the new last commit and shouldFlushOnUncommittedSegmentBytes() rejects the value -- the + // race can only suppress a flush trigger until the next successful segments sync republishes, never cause + // a spurious flush. final SegmentInfos committedInfos = this.lastCommittedSegmentInfos; if (committedInfos == null) { return; @@ -1577,9 +1582,12 @@ private boolean shouldFlushOnUncommittedSegmentBytes() { if (indexSettings.isFlushOnUncommittedSegmentsEnabled() == false) { return false; } - return current.bytes > 0 + // snapshot the volatile once so the null check and the generation comparison observe the same commit point + final SegmentInfos committedInfos = this.lastCommittedSegmentInfos; + return committedInfos != null + && current.bytes > 0 && current.bytes >= indexSettings.getFlushOnUncommittedSegmentsThresholdSize().getBytes() - && current.committedInfosGeneration == lastCommittedSegmentInfos.getGeneration(); + && current.committedInfosGeneration == committedInfos.getGeneration(); } @Override From 09889d02b78da9f8951b1e92250a48fb0e343e85 Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Thu, 27 Aug 2026 09:59:09 +0000 Subject: [PATCH 04/10] Bound flush_on_uncommitted_segments.threshold_size The threshold setting accepted any parseable byte size including 0b and -1, which would trigger a flush on every successful segments sync -- the opposite of what a user reaching for -1 (the common disable idiom) intends, and disablement already has its own explicit setting. Bound the value to [1b, Long.MAX_VALUE] via a new byteSizeSetting overload that combines a fallback setting with min/max validation, mirroring the existing bounded-default overload. Also adds settings-level test coverage for the fallback semantics: default inherited from index.translog.flush_threshold_size, dynamic updates of only the fallback reflected live, explicit value winning over the fallback, and rejection of zero/negative sizes. Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> --- .../opensearch/common/settings/Setting.java | 10 ++++ .../org/opensearch/index/IndexSettings.java | 4 ++ .../opensearch/index/IndexSettingsTests.java | 50 +++++++++++++++++++ 3 files changed, 64 insertions(+) diff --git a/server/src/main/java/org/opensearch/common/settings/Setting.java b/server/src/main/java/org/opensearch/common/settings/Setting.java index d31ab400223ea..8e9b55d9f5d40 100644 --- a/server/src/main/java/org/opensearch/common/settings/Setting.java +++ b/server/src/main/java/org/opensearch/common/settings/Setting.java @@ -2109,6 +2109,16 @@ public static Setting byteSizeSetting(String key, Setting(key, fallbackSetting, new ByteSizeValueParser(key), properties); } + public static Setting byteSizeSetting( + String key, + Setting fallbackSetting, + ByteSizeValue minValue, + ByteSizeValue maxValue, + Property... properties + ) { + return new Setting<>(key, fallbackSetting, new ByteSizeValueParser(minValue, maxValue, key), properties); + } + public static Setting byteSizeSetting(String key, Function defaultValue, Property... properties) { return new Setting<>(key, defaultValue, new ByteSizeValueParser(key), properties); } diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java index 854c2b919938f..72f49f1c8324d 100644 --- a/server/src/main/java/org/opensearch/index/IndexSettings.java +++ b/server/src/main/java/org/opensearch/index/IndexSettings.java @@ -905,6 +905,10 @@ public static IndexMergePolicy fromString(String text) { .byteSizeSetting( "index.remote_store.flush_on_uncommitted_segments.threshold_size", INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING, + // a zero or negative threshold would flush on every successful segments sync; disablement has its own + // explicit setting above + new ByteSizeValue(1, ByteSizeUnit.BYTES), + new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES), Property.Dynamic, Property.IndexScope ); diff --git a/server/src/test/java/org/opensearch/index/IndexSettingsTests.java b/server/src/test/java/org/opensearch/index/IndexSettingsTests.java index 5fb949cdbbaa2..ffab2d8ae956b 100644 --- a/server/src/test/java/org/opensearch/index/IndexSettingsTests.java +++ b/server/src/test/java/org/opensearch/index/IndexSettingsTests.java @@ -42,6 +42,7 @@ import org.opensearch.common.settings.Settings; import org.opensearch.common.settings.SettingsException; import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.unit.ByteSizeUnit; import org.opensearch.core.common.unit.ByteSizeValue; import org.opensearch.index.translog.Translog; import org.opensearch.indices.replication.common.ReplicationType; @@ -593,6 +594,55 @@ public void testTranslogFlushSizeThreshold() { assertEquals(actualNewTranslogFlushThresholdSize, settings.getFlushThresholdSize()); } + public void testFlushOnUncommittedSegmentsThresholdSize() { + // defaults to the index's translog flush threshold via setting fallback + IndexMetadata metadata = newIndexMeta( + "index", + Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) + .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "128mb") + .build() + ); + IndexSettings settings = new IndexSettings(metadata, Settings.EMPTY); + assertEquals(new ByteSizeValue(128, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize()); + // a dynamic update of only the fallback setting is reflected + settings.updateIndexMetadata( + newIndexMeta( + "index", + Settings.builder().put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "256mb").build() + ) + ); + assertEquals(new ByteSizeValue(256, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize()); + // an explicit value wins over the fallback + settings.updateIndexMetadata( + newIndexMeta( + "index", + Settings.builder() + .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "256mb") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "64mb") + .build() + ) + ); + assertEquals(new ByteSizeValue(64, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize()); + // zero and negative values are rejected: they would flush on every successful segments sync + for (String invalid : new String[] { "0b", "-1" }) { + IllegalArgumentException e = expectThrows( + IllegalArgumentException.class, + () -> new IndexSettings( + newIndexMeta( + "index", + Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), invalid) + .build() + ), + Settings.EMPTY + ) + ); + assertTrue(e.getMessage(), e.getMessage().contains("failed to parse value [" + invalid + "]")); + } + } + public void testTranslogGenerationSizeThreshold() { final ByteSizeValue size = new ByteSizeValue(Math.abs(randomInt())); final String key = IndexSettings.INDEX_TRANSLOG_GENERATION_THRESHOLD_SIZE_SETTING.getKey(); From 1d51a58457703dd843d22e879252831a8070cd79 Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Mon, 31 Aug 2026 10:59:00 +0000 Subject: [PATCH 05/10] Add tests for accounting exclusions and publish guard, document tests Asserts the uncommitted-segment-bytes accounting excludes committed files and segments_N (bounding the computed total from both sides), and that the listener publish guard skips cleanly when the setting is disabled or the indexer/engine is not the expected shape while the segments sync still completes. Adds Javadocs to all tests introduced by this change. Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> --- .../opensearch/index/IndexSettingsTests.java | 10 +- .../index/engine/InternalEngineTests.java | 75 ++++++++++- .../RemoteStoreRefreshListenerTests.java | 126 ++++++++++++++++++ .../remotestore/RemoteStoreCoreTestCase.java | 27 ++-- 4 files changed, 218 insertions(+), 20 deletions(-) diff --git a/server/src/test/java/org/opensearch/index/IndexSettingsTests.java b/server/src/test/java/org/opensearch/index/IndexSettingsTests.java index ffab2d8ae956b..1ecc458cdc1c4 100644 --- a/server/src/test/java/org/opensearch/index/IndexSettingsTests.java +++ b/server/src/test/java/org/opensearch/index/IndexSettingsTests.java @@ -594,8 +594,13 @@ public void testTranslogFlushSizeThreshold() { assertEquals(actualNewTranslogFlushThresholdSize, settings.getFlushThresholdSize()); } + /** + * Verifies {@code index.remote_store.flush_on_uncommitted_segments.threshold_size}: it defaults to the index's + * {@code index.translog.flush_threshold_size} via setting fallback, a dynamic update of only the fallback is + * reflected, an explicit value wins over the fallback, and zero or negative sizes are rejected since they + * would flush on every successful segments sync. + */ public void testFlushOnUncommittedSegmentsThresholdSize() { - // defaults to the index's translog flush threshold via setting fallback IndexMetadata metadata = newIndexMeta( "index", Settings.builder() @@ -605,7 +610,6 @@ public void testFlushOnUncommittedSegmentsThresholdSize() { ); IndexSettings settings = new IndexSettings(metadata, Settings.EMPTY); assertEquals(new ByteSizeValue(128, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize()); - // a dynamic update of only the fallback setting is reflected settings.updateIndexMetadata( newIndexMeta( "index", @@ -613,7 +617,6 @@ public void testFlushOnUncommittedSegmentsThresholdSize() { ) ); assertEquals(new ByteSizeValue(256, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize()); - // an explicit value wins over the fallback settings.updateIndexMetadata( newIndexMeta( "index", @@ -624,7 +627,6 @@ public void testFlushOnUncommittedSegmentsThresholdSize() { ) ); assertEquals(new ByteSizeValue(64, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize()); - // zero and negative values are rejected: they would flush on every successful segments sync for (String invalid : new String[] { "0b", "-1" }) { IllegalArgumentException e = expectThrows( IllegalArgumentException.class, diff --git a/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java b/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java index e4fa8bf92f28f..8de3fe5c16072 100644 --- a/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java +++ b/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java @@ -6812,6 +6812,12 @@ public void testShouldPeriodicallyFlushAfterMerge() throws Exception { assertThat(engine.shouldPeriodicallyFlush(), equalTo(false)); } + /** + * Verifies the engine flush condition on uncommitted segment bytes published by the remote segment upload path: + * nothing triggers before a publication, the threshold defaults to {@code index.translog.flush_threshold_size} + * via setting fallback, dynamic threshold and enabled-flag updates take effect immediately without a republish, + * and a stale commit-generation stamp after a flush can never re-trigger a flush (no flush loop). + */ public void testShouldPeriodicallyFlushOnUncommittedSegmentBytes() throws Exception { assertThat("Empty engine does not need flushing", engine.shouldPeriodicallyFlush(), equalTo(false)); ParsedDocument doc = testParsedDocument("0", null, testDocumentWithTextField(), SOURCE, null); @@ -6819,7 +6825,6 @@ public void testShouldPeriodicallyFlushOnUncommittedSegmentBytes() throws Except engine.refresh("test"); assertThat("Nothing published yet", engine.shouldPeriodicallyFlush(), equalTo(false)); - // simulate the remote segment upload path publishing the post-refresh local file sizes final Map localSegmentsSizeMap = new HashMap<>(); try (GatedCloseable snapshot = engine.getSegmentInfosSnapshot()) { for (String file : snapshot.get().files(false)) { @@ -6859,11 +6864,77 @@ public void testShouldPeriodicallyFlushOnUncommittedSegmentBytes() throws Except engine.flush(); assertThat("Stale commit generation stamp is ignored after flush", engine.shouldPeriodicallyFlush(), equalTo(false)); - // republishing against the new commit computes zero uncommitted bytes, hence no flush loop engine.updateUncommittedSegmentBytes(localSegmentsSizeMap); assertThat("All published files are committed now", engine.shouldPeriodicallyFlush(), equalTo(false)); } + /** + * Verifies the accounting inside {@code updateUncommittedSegmentBytes}: only segment files absent from the last + * commit point are summed, while committed files and {@code segments_N} entries are excluded. The total is + * asserted exactly by probing thresholds of the uncommitted size and one byte above it, and a publication + * holding only committed files computes zero bytes and can never trigger a flush. + */ + public void testUncommittedSegmentBytesExcludeCommittedAndSegmentsNFiles() throws Exception { + // establish a commit point holding the first segment + engine.index(indexForDoc(testParsedDocument("0", null, testDocumentWithTextField(), SOURCE, null))); + engine.flush(); + engine.refresh("test"); + final Set committedFiles; + try (GatedCloseable snapshot = engine.getSegmentInfosSnapshot()) { + committedFiles = new HashSet<>(snapshot.get().files(false)); + } + + // write a second, uncommitted segment + engine.index(indexForDoc(testParsedDocument("1", null, testDocumentWithTextField(), SOURCE, null))); + engine.refresh("test"); + + final Map localSegmentsSizeMap = new HashMap<>(); + long uncommittedBytes = 0; + try (GatedCloseable snapshot = engine.getSegmentInfosSnapshot()) { + for (String file : snapshot.get().files(false)) { + final long length = engine.store.directory().fileLength(file); + localSegmentsSizeMap.put(file, length); + if (committedFiles.contains(file) == false) { + uncommittedBytes += length; + } + } + } + localSegmentsSizeMap.put(IndexFileNames.SEGMENTS + "_99", Long.MAX_VALUE / 2); + assertThat("The workload must produce uncommitted segment files", uncommittedBytes, greaterThan(0L)); + + engine.updateUncommittedSegmentBytes(localSegmentsSizeMap); + + final IndexSettings indexSettings = engine.config().getIndexSettings(); + updateIndexSettings( + indexSettings, + Settings.builder() + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), uncommittedBytes + "b") + ); + assertThat("Exactly the uncommitted segment bytes are counted", engine.shouldPeriodicallyFlush(), equalTo(true)); + + updateIndexSettings( + indexSettings, + Settings.builder() + .put( + IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), + (uncommittedBytes + 1) + "b" + ) + ); + assertThat("Committed files and segments_N never contribute to the accounting", engine.shouldPeriodicallyFlush(), equalTo(false)); + + final Map committedOnlySizeMap = new HashMap<>(); + for (String file : committedFiles) { + committedOnlySizeMap.put(file, engine.store.directory().fileLength(file)); + } + committedOnlySizeMap.put(IndexFileNames.SEGMENTS + "_99", Long.MAX_VALUE / 2); + engine.updateUncommittedSegmentBytes(committedOnlySizeMap); + updateIndexSettings( + indexSettings, + Settings.builder().put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + ); + assertThat("Zero uncommitted bytes never trigger a flush", engine.shouldPeriodicallyFlush(), equalTo(false)); + } + private static void updateIndexSettings(IndexSettings indexSettings, Settings.Builder settingsBuilder) { indexSettings.updateIndexMetadata( IndexMetadata.builder(indexSettings.getIndexMetadata()) diff --git a/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java b/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java index 2e607efbbe8df..e71cd51203816 100644 --- a/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java +++ b/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java @@ -27,9 +27,13 @@ import org.opensearch.common.unit.TimeValue; import org.opensearch.core.action.ActionListener; import org.opensearch.core.index.shard.ShardId; +import org.opensearch.index.IndexSettings; +import org.opensearch.index.engine.Engine; +import org.opensearch.index.engine.EngineBackedIndexer; import org.opensearch.index.engine.InternalEngineFactory; import org.opensearch.index.engine.NRTReplicationEngineFactory; import org.opensearch.index.engine.exec.EngineBackedIndexerFactory; +import org.opensearch.index.engine.exec.Indexer; import org.opensearch.index.engine.exec.coord.CatalogSnapshot; import org.opensearch.index.remote.RemoteSegmentTransferTracker; import org.opensearch.index.remote.RemoteStoreStatsTrackerFactory; @@ -61,8 +65,11 @@ import static org.opensearch.index.store.RemoteSegmentStoreDirectory.METADATA_FILES_TO_FETCH; import static org.opensearch.test.RemoteStoreTestUtils.createMetadataFileBytes; import static org.opensearch.test.RemoteStoreTestUtils.getDummyMetadata; +import static org.mockito.AdditionalAnswers.delegatesTo; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -935,6 +942,125 @@ public void testCleanupNotTriggeredWhenThresholdDisabled() throws IOException { ); } + /** + * Verifies that {@code index.remote_store.flush_on_uncommitted_segments.enabled=false} stops the listener from + * publishing the accounting on a successful segments sync, not just the engine from consulting it: enabling + * the setting afterwards trips no flush at a 1b threshold until the next sync republishes, which then does. + */ + public void testDisabledFlushOnUncommittedSegmentsSkipsPublish() throws Exception { + indexShard = newStartedShard( + true, + Settings.builder() + .put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, true) + .put(IndexMetadata.SETTING_REMOTE_SEGMENT_STORE_REPOSITORY, "temp-fs") + .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY, "temp-fs") + .put(SETTING_REPLICATION_TYPE, ReplicationType.SEGMENT) + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING.getKey(), false) + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .build(), + new EngineBackedIndexerFactory(new InternalEngineFactory()) + ); + indexDocs(1, randomIntBetween(1, 10)); + indexShard.refresh("test"); + + clusterService = ClusterServiceUtils.createClusterService( + Settings.EMPTY, + new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), + threadPool + ); + remoteStoreStatsTrackerFactory = new RemoteStoreStatsTrackerFactory(clusterService, Settings.EMPTY); + remoteStoreStatsTrackerFactory.afterIndexShardCreated(indexShard); + RemoteSegmentTransferTracker tracker = remoteStoreStatsTrackerFactory.getRemoteSegmentTransferTracker(indexShard.shardId()); + remoteStoreRefreshListener = new RemoteStoreRefreshListener( + indexShard, + SegmentReplicationCheckpointPublisher.EMPTY, + tracker, + DefaultRemoteStoreSettings.INSTANCE + ); + + remoteStoreRefreshListener.afterRefresh(true); + + updateFlushOnUncommittedSegmentsEnabled(true); + assertFalse("No accounting is published while the setting is disabled", indexShard.shouldPeriodicallyFlush()); + + indexDocs(20, 1); + indexShard.refresh("test"); + remoteStoreRefreshListener.afterRefresh(true); + assertBusy(() -> assertTrue("Re-enabled sync publishes the accounting", indexShard.shouldPeriodicallyFlush())); + } + + /** + * Verifies that a successful segments sync on a shard whose indexer is not an {@link EngineBackedIndexer} + * (simulated by a delegating mock) skips the uncommitted-segment-bytes publication without breaking the sync: + * the replication checkpoint is still published. + */ + public void testNonEngineBackedIndexerSkipsUncommittedSegmentBytesPublish() throws Exception { + setup(true, 3); + Indexer delegatingIndexer = mock(Indexer.class, delegatesTo(indexShard.getIndexer())); + IndexShard spyShard = spy(indexShard); + doReturn(delegatingIndexer).when(spyShard).getIndexer(); + + SegmentReplicationCheckpointPublisher publisher = spy(SegmentReplicationCheckpointPublisher.EMPTY); + RemoteSegmentTransferTracker tracker = remoteStoreStatsTrackerFactory.getRemoteSegmentTransferTracker(indexShard.shardId()); + RemoteStoreRefreshListener listener = new RemoteStoreRefreshListener( + spyShard, + publisher, + tracker, + DefaultRemoteStoreSettings.INSTANCE + ); + try { + indexDocs(10, 1); + indexShard.refresh("test"); + listener.afterRefresh(true); + verify(publisher, atLeastOnce()).publish(any(), any()); + } finally { + listener.drainRefreshes(); + } + } + + /** + * Verifies that a successful segments sync on a shard whose {@link EngineBackedIndexer} wraps an engine other + * than {@code InternalEngine} (simulated by a delegating mock) skips the uncommitted-segment-bytes publication + * without breaking the sync: the replication checkpoint is still published. + */ + public void testNonInternalEngineSkipsUncommittedSegmentBytesPublish() throws Exception { + setup(true, 3); + EngineBackedIndexer delegatingIndexer = mock(EngineBackedIndexer.class, delegatesTo(indexShard.getIndexer())); + doReturn(mock(Engine.class)).when(delegatingIndexer).getEngine(); + IndexShard spyShard = spy(indexShard); + doReturn(delegatingIndexer).when(spyShard).getIndexer(); + + SegmentReplicationCheckpointPublisher publisher = spy(SegmentReplicationCheckpointPublisher.EMPTY); + RemoteSegmentTransferTracker tracker = remoteStoreStatsTrackerFactory.getRemoteSegmentTransferTracker(indexShard.shardId()); + RemoteStoreRefreshListener listener = new RemoteStoreRefreshListener( + spyShard, + publisher, + tracker, + DefaultRemoteStoreSettings.INSTANCE + ); + try { + indexDocs(10, 1); + indexShard.refresh("test"); + listener.afterRefresh(true); + verify(publisher, atLeastOnce()).publish(any(), any()); + } finally { + listener.drainRefreshes(); + } + } + + private void updateFlushOnUncommittedSegmentsEnabled(boolean enabled) { + indexShard.indexSettings() + .updateIndexMetadata( + IndexMetadata.builder(indexShard.indexSettings().getIndexMetadata()) + .settings( + Settings.builder() + .put(indexShard.indexSettings().getSettings()) + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING.getKey(), enabled) + ) + .build() + ); + } + private RemoteSegmentStoreDirectory setupDirectoryWithThreshold(int threshold) throws IOException { indexShard = newStartedShard( true, diff --git a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java index 5f747bc074d76..29011808a4699 100644 --- a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java +++ b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java @@ -146,10 +146,13 @@ private void testPeerRecovery(int numberOfIterations, boolean invokeFlush) throw ); } + /** + * Verifies the end-to-end periodic flush trigger on a remote-store shard: with the uncommitted-segment-bytes + * threshold lowered to 1b, a refresh-driven segments upload publishes the accounting and the next write + * operation's periodic flush poll fires the async flush. + */ public void testPeriodicFlushOnUncommittedSegmentBytes() throws Exception { String dataNode = internalCluster().startNodes(1).get(0); - // lower the uncommitted segment bytes threshold so that any refreshed but uncommitted segment breaches it; - // the condition itself is enabled by default createIndex( INDEX_NAME, Settings.builder() @@ -160,8 +163,6 @@ public void testPeriodicFlushOnUncommittedSegmentBytes() throws Exception { ensureGreen(INDEX_NAME); IndexShard indexShard = getIndexShard(dataNode, INDEX_NAME); assertEquals(0, indexShard.flushStats().getPeriodic()); - // each iteration drives one refresh (and hence one remote segments sync publishing the uncommitted bytes) - // followed by a write operation which polls the periodic flush condition and triggers the async flush assertBusy(() -> { indexSingleDoc(INDEX_NAME); refresh(INDEX_NAME); @@ -171,12 +172,11 @@ public void testPeriodicFlushOnUncommittedSegmentBytes() throws Exception { } /** - * End-to-end reproduction of the scenario in issue #22802: an update-heavy workload with regular refreshes on a - * remote-store shard. Every successful segments upload advances minSeqNoToKeep which suppresses the translog - * based flush condition, so without the uncommitted-segment-bytes condition no periodic flush ever happens, the - * safe commit never advances, and the soft-deleted documents produced by the updates stay protected from merge - * reclamation indefinitely. This test asserts the full recovery chain: periodic flushes fire, the safe commit - * advances, and a merge that does NOT flush (flush=false) is able to reclaim the tombstones. + * End-to-end reproduction of an update-heavy workload with regular refreshes on a remote-store shard, where + * every successful segments upload suppresses the translog based flush condition and soft-deleted documents + * pile up unreclaimed. Asserts the full recovery chain through the uncommitted-segment-bytes condition: + * periodic flushes fire, the safe commit advances past the workload, and a non-flushing expunge-deletes merge + * is able to reclaim the tombstones. */ public void testPeriodicFlushReclaimsSoftDeletesUnderUpdateHeavyWorkload() throws Exception { final int docCount = 20; @@ -238,10 +238,9 @@ public void testPeriodicFlushReclaimsSoftDeletesUnderUpdateHeavyWorkload() throw assertThat(committedCheckpoint, greaterThanOrEqualTo(maxSeqNoAfterWorkload)); }, 30, TimeUnit.SECONDS); - // a merge WITHOUT a flush can now reclaim tombstones below the advanced safe commit. Without the fix the - // soft-deletes floor never moves off the initial commit, nothing is reclaimable, and the count can never - // drop below the full tombstone pile -- so ANY reclamation proves the chain end to end. Each retry keeps - // writing so the retention lease and the async flush pipeline keep advancing. + // without the fix the soft-deletes floor never moves off the initial commit and the deleted-docs count can + // never drop below the full tombstone pile -- so ANY reclamation proves the chain. Each retry keeps writing + // so the retention lease and the async flush pipeline keep advancing. final long tombstonesCreated = (long) (rounds - 1) * docCount; final AtomicInteger nudge = new AtomicInteger(); assertBusy(() -> { From 27054decdb46a27390af389bc9a2208127bf955a Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Mon, 31 Aug 2026 11:05:51 +0000 Subject: [PATCH 06/10] Retrigger gradle check Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> From c9b0905a82c5e4a29e9ccdbe97c6b175487b917e Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Mon, 31 Aug 2026 17:44:30 +0530 Subject: [PATCH 07/10] Retrigger gradle check Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> From 3c236088891be7e743c8faa2ce0094270a221227 Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Mon, 31 Aug 2026 20:26:09 +0530 Subject: [PATCH 08/10] Make uncommitted-bytes flush hook engine-generic Address review feedback on #22865 (discussion r3894456901): - Hoist updateUncommittedSegmentBytes to the Engine base class as a default no-op, overridden by InternalEngine. This removes the instanceof InternalEngine downcast in RemoteStoreRefreshListener and gives NRTReplicationEngine/read-only engines correct-by-construction no-op behavior; only the publisher is remote-specific. - Drop the redundant current.bytes > 0 check in shouldFlushOnUncommittedSegmentBytes, subsumed by bytes >= threshold given the setting's 1-byte minimum. - Document index.periodic_flush_interval as an opt-in wall-clock flush backstop for low-throughput trickle workloads. No behavior change on remote-store primaries. Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> --- .../java/org/opensearch/index/IndexSettings.java | 4 +++- .../java/org/opensearch/index/engine/Engine.java | 11 +++++++++++ .../org/opensearch/index/engine/InternalEngine.java | 3 ++- .../index/shard/RemoteStoreRefreshListener.java | 5 ++--- .../shard/RemoteStoreRefreshListenerTests.java | 13 ++++++++----- 5 files changed, 26 insertions(+), 10 deletions(-) diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java index 72f49f1c8324d..56f1f581ff491 100644 --- a/server/src/main/java/org/opensearch/index/IndexSettings.java +++ b/server/src/main/java/org/opensearch/index/IndexSettings.java @@ -886,7 +886,9 @@ public static IndexMergePolicy fromString(String text) { * refreshed and uploaded to the remote store but not yet referenced by the last commit point. On remote-store * shards the translog based flush threshold ({@code index.translog.flush_threshold_size}) is ineffective because * uploaded translog generations are trimmed continuously; this condition restores an equivalent size-based flush - * signal driven by uncommitted segment bytes. + * signal driven by uncommitted segment bytes. Low-throughput trickle workloads that never cross the byte + * threshold and never go idle can additionally enable {@code index.periodic_flush_interval} (disabled by + * default) for a wall-clock flush backstop. */ public static final Setting INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING = Setting.boolSetting( "index.remote_store.flush_on_uncommitted_segments.enabled", diff --git a/server/src/main/java/org/opensearch/index/engine/Engine.java b/server/src/main/java/org/opensearch/index/engine/Engine.java index 9146f4f55f79d..b047a50656191 100644 --- a/server/src/main/java/org/opensearch/index/engine/Engine.java +++ b/server/src/main/java/org/opensearch/index/engine/Engine.java @@ -1232,6 +1232,17 @@ public boolean refreshNeeded() { */ public abstract boolean shouldPeriodicallyFlush(); + /** + * Publishes the total size of segment bytes not yet referenced by the last commit point, computed from the + * post-refresh local segment file sizes. Engine-generic: the notion of "bytes since the last commit" is not + * remote-specific -- only the publisher (the remote segment upload path) is. The default is a no-op so that + * engines which never publish (e.g. {@link NRTReplicationEngine}, read-only engines) are correct by + * construction; {@link InternalEngine} overrides it to drive its uncommitted-segment-bytes flush condition. + * + * @param localSegmentsSizeMap post-refresh local segment file names mapped to their sizes in bytes + */ + public void updateUncommittedSegmentBytes(Map localSegmentsSizeMap) {} + /** * Flushes the state of the engine including the transaction log, clearing memory. * diff --git a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java index db4bc20410150..247b11dfe152f 100644 --- a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java +++ b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java @@ -1538,6 +1538,7 @@ public boolean shouldPeriodicallyFlush() { * * @param localSegmentsSizeMap post-refresh local segment file names mapped to their sizes in bytes */ + @Override public void updateUncommittedSegmentBytes(Map localSegmentsSizeMap) { // the file set and the generation stamp are both taken from this single snapshot, so the published value is // always internally consistent. If a flush lands between this snapshot and the publish below, the stamp no @@ -1584,8 +1585,8 @@ private boolean shouldFlushOnUncommittedSegmentBytes() { } // snapshot the volatile once so the null check and the generation comparison observe the same commit point final SegmentInfos committedInfos = this.lastCommittedSegmentInfos; + // the threshold setting has a hard 1-byte minimum, so current.bytes >= threshold already implies bytes > 0 return committedInfos != null - && current.bytes > 0 && current.bytes >= indexSettings.getFlushOnUncommittedSegmentsThresholdSize().getBytes() && current.committedInfosGeneration == committedInfos.getGeneration(); } diff --git a/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java b/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java index 30e2158ae4d0d..567a79674436d 100644 --- a/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java +++ b/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java @@ -440,9 +440,8 @@ private void onSuccessfulSegmentsSync( // shards, hence publish the size of segment bytes not yet referenced by the last commit point so that the // engine can flush once they breach the configured threshold. if (indexShard.indexSettings().isFlushOnUncommittedSegmentsEnabled() - && indexShard.getIndexer() instanceof EngineBackedIndexer engineBacked - && engineBacked.getEngine() instanceof InternalEngine internalEngine) { - internalEngine.updateUncommittedSegmentBytes(localFileSizeMap); + && indexShard.getIndexer() instanceof EngineBackedIndexer engineBacked) { + engineBacked.getEngine().updateUncommittedSegmentBytes(localFileSizeMap); } // Publishing the new checkpoint which is used for remote store + segrep indexes checkpointPublisher.publish(indexShard, checkpoint); diff --git a/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java b/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java index e71cd51203816..49c278c7457d0 100644 --- a/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java +++ b/server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java @@ -1019,14 +1019,16 @@ public void testNonEngineBackedIndexerSkipsUncommittedSegmentBytesPublish() thro } /** - * Verifies that a successful segments sync on a shard whose {@link EngineBackedIndexer} wraps an engine other - * than {@code InternalEngine} (simulated by a delegating mock) skips the uncommitted-segment-bytes publication - * without breaking the sync: the replication checkpoint is still published. + * Verifies that on a shard whose {@link EngineBackedIndexer} wraps an engine other than {@code InternalEngine} + * (simulated by a generic {@link Engine} mock), a successful segments sync dispatches the uncommitted-segment-bytes + * publication uniformly through the {@link Engine} base type -- where it is a no-op -- without breaking the sync: + * the base method is invoked and the replication checkpoint is still published. */ - public void testNonInternalEngineSkipsUncommittedSegmentBytesPublish() throws Exception { + public void testNonInternalEngineNoOpUncommittedSegmentBytesPublish() throws Exception { setup(true, 3); + Engine mockEngine = mock(Engine.class); EngineBackedIndexer delegatingIndexer = mock(EngineBackedIndexer.class, delegatesTo(indexShard.getIndexer())); - doReturn(mock(Engine.class)).when(delegatingIndexer).getEngine(); + doReturn(mockEngine).when(delegatingIndexer).getEngine(); IndexShard spyShard = spy(indexShard); doReturn(delegatingIndexer).when(spyShard).getIndexer(); @@ -1042,6 +1044,7 @@ public void testNonInternalEngineSkipsUncommittedSegmentBytesPublish() throws Ex indexDocs(10, 1); indexShard.refresh("test"); listener.afterRefresh(true); + verify(mockEngine, atLeastOnce()).updateUncommittedSegmentBytes(any()); verify(publisher, atLeastOnce()).publish(any(), any()); } finally { listener.drainRefreshes(); From c1f2b80f2f4cc3aa592675a3376525d52cb6816c Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Mon, 31 Aug 2026 22:02:47 +0530 Subject: [PATCH 09/10] Retrigger gradle check Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> From 3d707793729c63ff281c052257f5e005cc35c8bf Mon Sep 17 00:00:00 2001 From: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com> Date: Mon, 31 Aug 2026 22:30:21 +0530 Subject: [PATCH 10/10] Retrigger gradle check Signed-off-by: Shourya Dutta Biswas <114977491+shourya035@users.noreply.github.com>