Make warmup settings dynamic for pull-based ingestion

CHANGELOG.md

@@ -25,6 +25,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for enabling pluggable data formats, starting with phase-1 of decoupling shard from engine, and introducing basic abstractions ([#20675](https://github.com/opensearch-project/OpenSearch/pull/20675))
 
 - Add warmup phase to wait for lag to catch up in pull-based ingestion before serving ([#20526](https://github.com/opensearch-project/OpenSearch/pull/20526))
+- Make warmup settings dynamic for pull-based ingestion ([#20931](https://github.com/opensearch-project/OpenSearch/pull/20931))
 ### Changed
 - Make telemetry `Tags` immutable ([#20788](https://github.com/opensearch-project/OpenSearch/pull/20788))
 - Move Randomness from server to libs/common ([#20570](https://github.com/opensearch-project/OpenSearch/pull/20570))

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -1041,7 +1041,7 @@ public Iterator<Setting<?>> settings() {
         TimeValue.timeValueMillis(-1),
         TimeValue.timeValueMillis(-1),
         Property.IndexScope,
-        Property.Final
+        Property.Dynamic
     );
 
     /**
@@ -1054,7 +1054,7 @@ public Iterator<Setting<?>> settings() {
         100L,
         0L,
         Property.IndexScope,
-        Property.Final
+        Property.Dynamic
     );
 
     /**

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

@@ -1070,6 +1070,16 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
      */
     private volatile boolean isStarTreeIndexEnabled;
 
+    /**
+     * The warmup timeout for pull-based ingestion.
+     */
+    private volatile TimeValue warmupTimeout;
+
+    /**
+     * The warmup lag threshold for pull-based ingestion.
+     */
+    private volatile long warmupLagThreshold;
+
     /**
      * Returns the default search fields for this index.
      */
@@ -1239,6 +1249,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         setDocIdFuzzySetFalsePositiveProbability(scopedSettings.get(INDEX_DOC_ID_FUZZY_SET_FALSE_POSITIVE_PROBABILITY_SETTING));
         isCompositeIndex = scopedSettings.get(StarTreeIndexSettings.IS_COMPOSITE_INDEX_SETTING);
         isStarTreeIndexEnabled = scopedSettings.get(StarTreeIndexSettings.STAR_TREE_SEARCH_ENABLED_SETTING);
+        this.warmupTimeout = IndexMetadata.INGESTION_SOURCE_WARMUP_TIMEOUT_SETTING.get(nodeSettings);
+        this.warmupLagThreshold = IndexMetadata.INGESTION_SOURCE_WARMUP_LAG_THRESHOLD_SETTING.get(nodeSettings);
         scopedSettings.addSettingsUpdateConsumer(
             TieredMergePolicyProvider.INDEX_COMPOUND_FORMAT_SETTING,
             tieredMergePolicyProvider::setNoCFSRatio
@@ -1381,6 +1393,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             this::setRemoteStoreTranslogRepository
         );
         scopedSettings.addSettingsUpdateConsumer(StarTreeIndexSettings.STAR_TREE_SEARCH_ENABLED_SETTING, this::setStarTreeIndexEnabled);
+        scopedSettings.addSettingsUpdateConsumer(IndexMetadata.INGESTION_SOURCE_WARMUP_TIMEOUT_SETTING, this::setWarmupTimeout);
+        scopedSettings.addSettingsUpdateConsumer(IndexMetadata.INGESTION_SOURCE_WARMUP_LAG_THRESHOLD_SETTING, this::setWarmupLagThreshold);
     }
 
     private void setSearchIdleAfter(TimeValue searchIdleAfter) {
@@ -2002,6 +2016,22 @@ public boolean getStarTreeIndexEnabled() {
         return isStarTreeIndexEnabled;
     }
 
+    private void setWarmupTimeout(TimeValue warmupTimeout) {
+        this.warmupTimeout = warmupTimeout;
+    }
+
+    private void setWarmupLagThreshold(long warmupLagThreshold) {
+        this.warmupLagThreshold = warmupLagThreshold;
+    }
+
+    public TimeValue getWarmupTimeout() {
+        return warmupTimeout;
+    }
+
+    public long getWarmupLagThreshold() {
+        return warmupLagThreshold;
+    }
+
     /**
      * Returns the merge policy that should be used for this index.
      *

server/src/main/java/org/opensearch/index/engine/IngestionEngine.java

@@ -486,6 +486,15 @@ public void close() throws IOException {
         super.close();
     }
 
+    /**
+     * Updates warmup configuration dynamically.
+     */
+    public void updateWarmupConfig(IngestionSource.WarmupConfig newConfig) {
+        if (streamPoller != null) {
+            streamPoller.updateWarmupConfig(newConfig);
+        }
+    }
+
     public DocumentMapperForType getDocumentMapperForType() {
         return documentMapperForType;
     }

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -68,6 +68,7 @@
 import org.opensearch.action.support.replication.ReplicationResponse;
 import org.opensearch.cluster.metadata.DataStream;
 import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.cluster.metadata.IngestionSource;
 import org.opensearch.cluster.metadata.MappingMetadata;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodes;
@@ -3338,6 +3339,19 @@ public void onSettingsChanged() {
                 indexSettings.getSoftDeleteRetentionOperations()
             );
         }
+
+        // Update warmup config if this is an ingestion engine
+        Indexer indexer = getIndexerOrNull();
+        if (indexer instanceof EngineBackedIndexer) {
+            Engine engine = ((EngineBackedIndexer) indexer).getEngine();
+            if (engine instanceof IngestionEngine) {
+                IngestionSource.WarmupConfig newConfig = new IngestionSource.WarmupConfig(
+                    indexSettings.getWarmupTimeout(),
+                    indexSettings.getWarmupLagThreshold()
+                );
+                ((IngestionEngine) engine).updateWarmupConfig(newConfig);
+            }
+        }
     }
 
     private void turnOffTranslogRetention() {

server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java

@@ -63,10 +63,10 @@ public class DefaultStreamPoller implements StreamPoller {
     private volatile long lastPointerBasedLagUpdateTime = 0;
 
     // Warmup configuration and state
-    private final IngestionSource.WarmupConfig warmupConfig;
+    private volatile IngestionSource.WarmupConfig warmupConfig;
     private volatile boolean warmupComplete = false;
     private volatile long warmupStartTime = 0;
-    private final CountDownLatch warmupLatch = new CountDownLatch(1);
+    private volatile CountDownLatch warmupLatch = new CountDownLatch(1);
 
     @Nullable
     private IngestionShardConsumer consumer;
@@ -430,6 +430,31 @@ public boolean awaitWarmupComplete(long timeoutMs) throws InterruptedException {
         return completed;
     }
 
+    /**
+     * Updates the warmup configuration dynamically.
+     * Called when index settings are changed at runtime.
+     */
+    @Override
+    public void updateWarmupConfig(IngestionSource.WarmupConfig newConfig) {
+        IngestionSource.WarmupConfig oldConfig = this.warmupConfig;
+        this.warmupConfig = newConfig;
+
+        // If warmup was enabled and is now disabled, mark as complete
+        if (oldConfig.isEnabled() && !newConfig.isEnabled() && !warmupComplete) {
+            warmupComplete = true;
+            warmupLatch.countDown();
+            logger.info("Warmup disabled for index {} shard {} via dynamic settings update", indexName, shardId);
+        }
+
+        logger.info(
+            "Warmup config updated for index {} shard {}: timeout={}, lagThreshold={}",
+            indexName,
+            shardId,
+            newConfig.timeout(),
+            newConfig.lagThreshold()
+        );
+    }
+
     /**
      * Check if warmup conditions are met and mark warmup as complete if so.
      *

server/src/main/java/org/opensearch/indices/pollingingest/StreamPoller.java

@@ -83,6 +83,12 @@ public interface StreamPoller extends Closeable, ClusterStateListener {
      */
     void requestConsumerReinitialization(IngestionSource updatedIngestionSource);
 
+    /**
+     * Updates the warmup configuration dynamically.
+     * Called when index settings are changed at runtime.
+     */
+    void updateWarmupConfig(IngestionSource.WarmupConfig config);
+
     /**
      * @return true if the warmup phase is complete and the shard is ready to serve
      */

server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java

@@ -976,4 +976,142 @@ public void testWarmupAwaitReturnsImmediatelyWhenAlreadyComplete() throws Interr
 
         warmupPoller.close();
     }
+
+    // ==================== Dynamic Warmup Config Update Tests ====================
+
+    public void testUpdateWarmupConfigDisableWhileInProgress() throws InterruptedException {
+        // Create a poller with warmup enabled
+        IngestionSource.WarmupConfig enabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 100L);
+        DefaultStreamPoller warmupPoller = new DefaultStreamPoller(
+            new FakeIngestionSource.FakeIngestionShardPointer(0),
+            fakeConsumerFactory,
+            "",
+            0,
+            partitionedBlockingQueueContainer,
+            StreamPoller.ResetState.NONE,
+            "",
+            errorStrategy,
+            StreamPoller.State.NONE,
+            1000,
+            1000,
+            10000,
+            indexSettings,
+            new DefaultIngestionMessageMapper(),
+            enabledConfig
+        );
+
+        // Warmup should not be complete yet
+        assertFalse(warmupPoller.isWarmupComplete());
+
+        // Dynamically disable warmup (timeout=-1)
+        IngestionSource.WarmupConfig disabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMillis(-1), 100L);
+        warmupPoller.updateWarmupConfig(disabledConfig);
+
+        // Warmup should now be complete since we disabled it
+        assertTrue(warmupPoller.isWarmupComplete());
+
+        warmupPoller.close();
+    }
+
+    public void testUpdateWarmupConfigThresholdWhileInProgress() {
+        // Create a poller with warmup enabled and a high threshold
+        IngestionSource.WarmupConfig initialConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 100L);
+        DefaultStreamPoller warmupPoller = new DefaultStreamPoller(
+            new FakeIngestionSource.FakeIngestionShardPointer(0),
+            fakeConsumerFactory,
+            "",
+            0,
+            partitionedBlockingQueueContainer,
+            StreamPoller.ResetState.NONE,
+            "",
+            errorStrategy,
+            StreamPoller.State.NONE,
+            1000,
+            1000,
+            10000,
+            indexSettings,
+            new DefaultIngestionMessageMapper(),
+            initialConfig
+        );
+
+        // Warmup should not be complete yet
+        assertFalse(warmupPoller.isWarmupComplete());
+
+        // Update threshold to a different value while warmup is in progress
+        IngestionSource.WarmupConfig updatedConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 50L);
+        warmupPoller.updateWarmupConfig(updatedConfig);
+
+        // Warmup should still not be complete (we just changed threshold, not disabled it)
+        assertFalse(warmupPoller.isWarmupComplete());
+
+        warmupPoller.close();
+    }
+
+    public void testUpdateWarmupConfigDoesNotReEnableAfterCompletion() {
+        // Create a poller with warmup disabled (warmup immediately complete)
+        IngestionSource.WarmupConfig disabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMillis(-1), 100L);
+        DefaultStreamPoller warmupPoller = new DefaultStreamPoller(
+            new FakeIngestionSource.FakeIngestionShardPointer(0),
+            fakeConsumerFactory,
+            "",
+            0,
+            partitionedBlockingQueueContainer,
+            StreamPoller.ResetState.NONE,
+            "",
+            errorStrategy,
+            StreamPoller.State.NONE,
+            1000,
+            1000,
+            10000,
+            indexSettings,
+            new DefaultIngestionMessageMapper(),
+            disabledConfig
+        );
+
+        // Warmup should be complete since it was disabled
+        assertTrue(warmupPoller.isWarmupComplete());
+
+        // Dynamically enable warmup - should NOT re-trigger since shard is already serving
+        IngestionSource.WarmupConfig enabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 50L);
+        warmupPoller.updateWarmupConfig(enabledConfig);
+
+        // Warmup should still be complete (not re-triggered)
+        assertTrue(warmupPoller.isWarmupComplete());
+
+        warmupPoller.close();
+    }
+
+    public void testUpdateWarmupConfigTimeoutWhileInProgress() {
+        // Create a poller with warmup enabled with a short timeout
+        IngestionSource.WarmupConfig initialConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueSeconds(30), 100L);
+        DefaultStreamPoller warmupPoller = new DefaultStreamPoller(
+            new FakeIngestionSource.FakeIngestionShardPointer(0),
+            fakeConsumerFactory,
+            "",
+            0,
+            partitionedBlockingQueueContainer,
+            StreamPoller.ResetState.NONE,
+            "",
+            errorStrategy,
+            StreamPoller.State.NONE,
+            1000,
+            1000,
+            10000,
+            indexSettings,
+            new DefaultIngestionMessageMapper(),
+            initialConfig
+        );
+
+        // Warmup should not be complete yet
+        assertFalse(warmupPoller.isWarmupComplete());
+
+        // Update timeout to a longer value while warmup is in progress
+        IngestionSource.WarmupConfig updatedConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(10), 100L);
+        warmupPoller.updateWarmupConfig(updatedConfig);
+
+        // Warmup should still not be complete (we just changed timeout)
+        assertFalse(warmupPoller.isWarmupComplete());
+
+        warmupPoller.close();
+    }
 }