feat(plugin): add ingestion plugin framework and example implementation

src/main/java/com/yelp/nrtsearch/server/config/IngestionPluginConfigs.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2025 Yelp Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yelp.nrtsearch.server.config;
+
+import com.fasterxml.jackson.annotation.JsonAnySetter;
+import java.util.HashMap;
+import java.util.Map;
+
+public class IngestionPluginConfigs {
+
+  private final Map<String, Map<String, Object>> pluginConfigs = new HashMap<>();
+
+  @JsonAnySetter
+  public void addPluginConfig(String pluginName, Map<String, Object> config) {
+    pluginConfigs.put(pluginName, config);
+  }
+
+  public Map<String, Map<String, Object>> getPluginConfigs() {
+    return pluginConfigs;
+  }
+}

src/main/java/com/yelp/nrtsearch/server/config/NrtsearchConfig.java

@@ -446,4 +446,14 @@ private static List<String> getPluginSearchPath(Object o) {
     }
     return paths;
   }
+
+  public Map<String, Map<String, Object>> getIngestionPluginConfigs() {
+    try {
+      return configReader.get(
+          "pluginConfigs.ingestion",
+          obj -> JsonUtils.convertValue(obj, IngestionPluginConfigs.class).getPluginConfigs());
+    } catch (ConfigKeyNotFoundException e) {
+      return Collections.emptyMap();
+    }
+  }
 }

src/main/java/com/yelp/nrtsearch/server/grpc/NrtsearchServer.java

@@ -79,6 +79,7 @@
 import com.yelp.nrtsearch.server.handler.UpdateFieldsHandler;
 import com.yelp.nrtsearch.server.handler.WriteNRTPointHandler;
 import com.yelp.nrtsearch.server.highlights.HighlighterService;
+import com.yelp.nrtsearch.server.ingestion.IngestionPluginUtils;
 import com.yelp.nrtsearch.server.logging.HitsLoggerCreator;
 import com.yelp.nrtsearch.server.modules.NrtsearchModule;
 import com.yelp.nrtsearch.server.monitoring.BootstrapMetrics;
@@ -95,6 +96,7 @@
 import com.yelp.nrtsearch.server.monitoring.SearchResponseCollector;
 import com.yelp.nrtsearch.server.monitoring.ThreadPoolCollector;
 import com.yelp.nrtsearch.server.monitoring.ThreadPoolCollector.RejectionCounterWrapper;
+import com.yelp.nrtsearch.server.plugins.IngestionPlugin;
 import com.yelp.nrtsearch.server.plugins.Plugin;
 import com.yelp.nrtsearch.server.plugins.PluginsService;
 import com.yelp.nrtsearch.server.remote.RemoteBackend;
@@ -390,6 +392,7 @@ static class LuceneServerImpl extends LuceneServerGrpc.LuceneServerImplBase {
       this.globalState = GlobalState.createState(configuration, remoteBackend);
 
       // Initialize handlers
+      initIngestionPlugin(globalState, plugins);
       addDocumentHandler = new AddDocumentHandler(globalState);
       backupWarmingQueriesHandler = new BackupWarmingQueriesHandler(globalState);
       commitHandler = new CommitHandler(globalState);
@@ -428,6 +431,15 @@ static class LuceneServerImpl extends LuceneServerGrpc.LuceneServerImplBase {
       updateFieldsHandler = new UpdateFieldsHandler(globalState);
     }
 
+    private void initIngestionPlugin(GlobalState globalState, List<Plugin> plugins)
+        throws IOException {
+      for (Plugin plugin : plugins) {
+        if (plugin instanceof IngestionPlugin ingestionPlugin) {
+          IngestionPluginUtils.initializeAndStart(ingestionPlugin, globalState);
+        }
+      }
+    }
+
     @VisibleForTesting
     static void initQueryCache(NrtsearchConfig configuration) {
       QueryCacheConfig cacheConfig = configuration.getQueryCacheConfig();

src/main/java/com/yelp/nrtsearch/server/ingestion/AbstractIngestor.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2025 Yelp Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yelp.nrtsearch.server.ingestion;
+
+import com.yelp.nrtsearch.server.config.NrtsearchConfig;
+import com.yelp.nrtsearch.server.grpc.AddDocumentRequest;
+import com.yelp.nrtsearch.server.handler.AddDocumentHandler;
+import com.yelp.nrtsearch.server.index.IndexState;
+import com.yelp.nrtsearch.server.state.GlobalState;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Abstract base class for ingestion implementations. Provides common ingestion utilities like
+ * addDocuments and commit. Plugin-specific ingestion logic should extend this class and implement
+ * start/stop.
+ */
+public abstract class AbstractIngestor implements Ingestor {
+  protected final NrtsearchConfig config;
+  protected GlobalState globalState;
+
+  public AbstractIngestor(NrtsearchConfig config) {
+    this.config = config;
+  }
+
+  /**
+   * Called by the framework to initialize the ingestor with global state. Must be called before
+   * addDocuments or commit.
+   */
+  @Override
+  public void initialize(GlobalState globalState) {
+    this.globalState = globalState;
+  }
+
+  /**
+   * Add documents to the index.
+   *
+   * @param addDocRequests list of documents to add
+   * @param indexName target index
+   * @return sequence number of the indexing operation
+   * @throws Exception if indexing fails
+   */
+  @Override
+  public long addDocuments(List<AddDocumentRequest> addDocRequests, String indexName)
+      throws Exception {
+    verifyInitialized(indexName);
+    return new AddDocumentHandler.DocumentIndexer(globalState, addDocRequests, indexName)
+        .runIndexingJob();
+  }
+
+  /**
+   * Commit changes to the index.
+   *
+   * @param indexName target index
+   * @throws IOException if commit fails
+   */
+  @Override
+  public void commit(String indexName) throws IOException {
+    verifyInitialized(indexName);
+    IndexState indexState = globalState.getIndexOrThrow(indexName);
+    indexState.commit();
+  }
+
+  private void verifyInitialized(String indexName) throws IOException {
+    if (globalState == null) {
+      throw new IllegalStateException("Ingestor not initialized with GlobalState");
+    }
+    IndexState indexState = globalState.getIndexOrThrow(indexName);
+    if (indexState == null) {
+      throw new IllegalStateException("Index not found: " + indexName);
+    }
+  }
+
+  /**
+   * Start ingestion logic. Must be implemented by plugin-specific subclass.
+   *
+   * @throws IOException if startup fails
+   */
+  @Override
+  public abstract void start() throws IOException;
+
+  /**
+   * Stop ingestion logic. Must be implemented by plugin-specific subclass.
+   *
+   * @throws IOException if shutdown fails
+   */
+  @Override
+  public abstract void stop() throws IOException;
+}

src/main/java/com/yelp/nrtsearch/server/ingestion/IngestionPluginUtils.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2025 Yelp Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yelp.nrtsearch.server.ingestion;
+
+import com.yelp.nrtsearch.server.plugins.IngestionPlugin;
+import com.yelp.nrtsearch.server.state.GlobalState;
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+
+public class IngestionPluginUtils {
+  public static void initializeAndStart(IngestionPlugin plugin, GlobalState globalState)
+      throws IOException {
+    Ingestor ingestor = plugin.getIngestor();
+    if (ingestor instanceof AbstractIngestor abstractIngestor) {
+      abstractIngestor.initialize(globalState);
+    }
+
+    ExecutorService executor = plugin.getIngestionExecutor();
+    executor.submit(
+        () -> {
+          try {
+            ingestor.start();
+          } catch (IOException e) {
+            e.printStackTrace();
+          }
+        });
+  }
+}

src/main/java/com/yelp/nrtsearch/server/ingestion/Ingestor.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2024 Yelp Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yelp.nrtsearch.server.ingestion;
+
+import com.yelp.nrtsearch.server.grpc.AddDocumentRequest;
+import com.yelp.nrtsearch.server.state.GlobalState;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Interface for ingestion logic used by plugins.
+ *
+ * <p>This interface defines the lifecycle and operations for ingesting documents into an index.
+ * Plugin implementations can use this interface to encapsulate source-specific ingestion logic
+ * (e.g., reading from Kafka, S3, etc.) while leveraging shared indexing utilities.
+ *
+ * <p>Implementations are expected to:
+ *
+ * <ul>
+ *   <li>Initialize with {@link GlobalState} before starting
+ *   <li>Start ingestion in a background thread (if needed)
+ *   <li>Use {@link #addDocuments(List, String)} and {@link #commit(String)} to index data
+ *   <li>Clean up resources in {@link #stop()}
+ * </ul>
+ */
+public interface Ingestor {
+
+  /**
+   * Initialize the ingestor with the global server state.
+   *
+   * <p>This method is called once by the framework before ingestion starts. Implementations should
+   * store the global state for later use (e.g., to access index state).
+   *
+   * @param globalState the global server state
+   */
+  void initialize(GlobalState globalState);
+
+  /**
+   * Start the ingestion process.
+   *
+   * <p>This method should contain the logic to begin reading from the ingestion source (e.g.,
+   * Kafka, file system, etc.). It may block or spawn background threads depending on the
+   * implementation.
+   *
+   * @throws IOException if ingestion startup fails
+   */
+  void start() throws IOException;
+
+  /**
+   * Stop the ingestion process and clean up resources.
+   *
+   * <p>This method is called during plugin shutdown. Implementations should stop any background
+   * threads, close connections, and release resources.
+   *
+   * @throws IOException if ingestion shutdown fails
+   */
+  void stop() throws IOException;
+
+  /**
+   * Add a batch of documents to the specified index.
+   *
+   * <p>This method is typically called from within the ingestion loop to index new data. It returns
+   * the Lucene sequence number of the indexing operation.
+   *
+   * @param addDocRequests list of document requests to add
+   * @param indexName name of the target index
+   * @return sequence number of the indexing operation
+   * @throws Exception if indexing fails
+   */
+  long addDocuments(List<AddDocumentRequest> addDocRequests, String indexName) throws Exception;
+
+  /**
+   * Commit any pending changes to the specified index.
+   *
+   * <p>This method should be called periodically or after a batch of documents is added to ensure
+   * durability and visibility of the changes.
+   *
+   * @param indexName name of the target index
+   * @throws IOException if commit fails
+   */
+  void commit(String indexName) throws IOException;
+}

src/main/java/com/yelp/nrtsearch/server/plugins/IngestionPlugin.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2025 Yelp Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yelp.nrtsearch.server.plugins;
+
+import com.yelp.nrtsearch.server.ingestion.Ingestor;
+import java.util.concurrent.ExecutorService;
+
+public interface IngestionPlugin {
+  /** Interface for ingestion logic used by plugins */
+  Ingestor getIngestor();
+
+  /**
+   * Provide an executor service for running ingestion. Plugin is responsible for managing its
+   * lifecycle.
+   */
+  ExecutorService getIngestionExecutor();
+}