Register listener in ctor

Signed-off-by: Rishab Nahata <rishabnahata07@gmail.com>

Author: imRishN

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

@@ -29,8 +29,9 @@
 /**
  * Handles ingest pipeline resolution and execution for pull-based ingestion.
  *
- * <p>Resolves configured pipelines from index settings (lazy, cached) and executes them
+ * <p>Resolves configured pipelines from index settings at initialization and executes them
  * synchronously by bridging IngestService's async callback API with CompletableFuture.
+ * Also registers a dynamic settings listener to pick up runtime changes to {@code final_pipeline}.
  * Only {@code final_pipeline} is supported.
  */
 public class IngestPipelineExecutor {
@@ -42,25 +43,26 @@ public class IngestPipelineExecutor {
 
     private final IngestService ingestService;
     private final String index;
-
-    // Cached pipeline names — resolved lazily on the first document
     private volatile String resolvedFinalPipeline;
-    private volatile boolean pipelinesResolved = false;
 
     /**
      * Creates an IngestPipelineExecutor for the given index.
+     * Resolves the final pipeline from index settings and registers a dynamic settings listener.
      *
      * @param ingestService the ingest service for pipeline execution
      * @param index the index name
+     * @param indexSettings the index settings to resolve a pipeline from and register listener on
      */
-    public IngestPipelineExecutor(IngestService ingestService, String index) {
+    public IngestPipelineExecutor(IngestService ingestService, String index, IndexSettings indexSettings) {
         this.ingestService = Objects.requireNonNull(ingestService);
         this.index = Objects.requireNonNull(index);
+        indexSettings.getScopedSettings().addSettingsUpdateConsumer(IndexSettings.FINAL_PIPELINE, this::updateFinalPipeline);
+        updateFinalPipeline(IndexSettings.FINAL_PIPELINE.get(indexSettings.getSettings()));
     }
 
     /**
      * Visible for testing. Creates an executor with a pre-resolved pipeline name,
-     * bypassing lazy resolution from index settings.
+     * bypassing resolution from index settings.
      *
      * @param ingestService the ingest service
      * @param index the index name
@@ -70,23 +72,6 @@ public IngestPipelineExecutor(IngestService ingestService, String index) {
         this.ingestService = Objects.requireNonNull(ingestService);
         this.index = Objects.requireNonNull(index);
         this.resolvedFinalPipeline = finalPipeline;
-        this.pipelinesResolved = true;
-    }
-
-    /**
-     * Resolves pipeline names from index settings. Called lazily on first document and cached.
-     * Also registers a dynamic settings listener for final_pipeline updates.
-     */
-    void resolvePipelineNames(IndexSettings indexSettings) {
-        if (pipelinesResolved) {
-            return;
-        }
-        updateFinalPipeline(IndexSettings.FINAL_PIPELINE.get(indexSettings.getSettings()));
-
-        // Register dynamic settings listener for final_pipeline updates
-        indexSettings.getScopedSettings().addSettingsUpdateConsumer(IndexSettings.FINAL_PIPELINE, this::updateFinalPipeline);
-
-        pipelinesResolved = true;
     }
 
     /**

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

@@ -126,8 +126,7 @@ static class MessageProcessor {
             String indexName = engine.config().getIndexSettings().getIndex().getName();
             this.engine = engine;
             this.index = indexName;
-            this.pipelineExecutor = new IngestPipelineExecutor(ingestService, indexName);
-            this.pipelineExecutor.resolvePipelineNames(engine.config().getIndexSettings());
+            this.pipelineExecutor = new IngestPipelineExecutor(ingestService, indexName, engine.config().getIndexSettings());
         }
 
         /**

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

@@ -39,17 +39,17 @@ public void setUp() throws Exception {
     // --- Construction tests ---
 
     public void testConstructorRequiresNonNullIngestService() {
-        expectThrows(NullPointerException.class, () -> new IngestPipelineExecutor(null, "test_index"));
+        expectThrows(NullPointerException.class, () -> new IngestPipelineExecutor(null, "test_index", (String) null));
     }
 
     public void testConstructorRequiresNonNullIndex() {
-        expectThrows(NullPointerException.class, () -> new IngestPipelineExecutor(ingestService, null));
+        expectThrows(NullPointerException.class, () -> new IngestPipelineExecutor(ingestService, null, (String) null));
     }
 
     // --- Pipeline resolution tests ---
 
     public void testHasPipelines_NoPipelineConfigured() {
-        IngestPipelineExecutor executor = new IngestPipelineExecutor(ingestService, "test_index", null);
+        IngestPipelineExecutor executor = new IngestPipelineExecutor(ingestService, "test_index", (String) null);
         assertFalse(executor.hasPipelines());
     }
 
@@ -59,7 +59,7 @@ public void testHasPipelines_PipelineConfigured() {
     }
 
     public void testUpdateFinalPipeline_SetsPipeline() {
-        IngestPipelineExecutor executor = new IngestPipelineExecutor(ingestService, "test_index", null);
+        IngestPipelineExecutor executor = new IngestPipelineExecutor(ingestService, "test_index", (String) null);
         assertFalse(executor.hasPipelines());
 
         executor.updateFinalPipeline("new-pipeline");
@@ -77,7 +77,7 @@ public void testUpdateFinalPipeline_NoopClearsPipeline() {
     // --- Execution: no pipeline configured ---
 
     public void testExecutePipelines_NoPipeline_ReturnsSourceUnchanged() throws Exception {
-        IngestPipelineExecutor executor = new IngestPipelineExecutor(ingestService, "test_index", null);
+        IngestPipelineExecutor executor = new IngestPipelineExecutor(ingestService, "test_index", (String) null);
 
         Map<String, Object> source = new HashMap<>();
         source.put("name", "alice");

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

@@ -64,7 +64,7 @@ public void setUp() throws Exception {
         processor = new MessageProcessorRunnable.MessageProcessor(
             ingestionEngine,
             "index",
-            new IngestPipelineExecutor(mock(IngestService.class), "index", null)
+            new IngestPipelineExecutor(mock(IngestService.class), "index", (String) null)
         );
     }