[NOCOMMIT] Hacky ingestion source wiring ideas

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

@@ -52,6 +52,7 @@
 import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.xcontent.XContentHelper;
+import org.opensearch.common.xcontent.json.JsonXContent;
 import org.opensearch.core.Assertions;
 import org.opensearch.core.common.Strings;
 import org.opensearch.core.common.io.stream.BufferedChecksumStreamOutput;
@@ -661,6 +662,7 @@ public static APIBlock readFrom(StreamInput input) throws IOException {
     public static final String REMOTE_STORE_CUSTOM_KEY = "remote_store";
     public static final String TRANSLOG_METADATA_KEY = "translog_metadata";
     public static final String CONTEXT_KEY = "context";
+    public static final String KEY_INGESTION_SOURCE = "ingestion_source";
 
     public static final String INDEX_STATE_FILE_PREFIX = "state-";
 
@@ -1352,6 +1354,38 @@ public static Builder builder(IndexMetadata indexMetadata) {
         return new Builder(indexMetadata);
     }
 
+    public static class IngestionSource {
+        public String type;
+        public Map<String, Object> config;
+
+        public static IngestionSource fromXContent(XContentParser parser) throws IOException {
+            if (parser.currentToken() == XContentParser.Token.START_OBJECT) {  // on a start object move to next token
+                parser.nextToken();
+            }
+            String currentField = null;
+            String type;
+            JsonXContent config = null;
+            while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
+                if (parser.nextToken() == XContentParser.Token.FIELD_NAME) {
+                    currentField = parser.currentName();
+                } else if ("type".equals(currentField)) {
+                    type = parser.text();
+                } else if ("config".equals(currentField)) {
+                    XContentBuilder.builder(JsonXContent.jsonXContent)
+                    config = parser.map();
+                }
+            }
+            config = Map.of(
+                "topic", "test",
+                "bootstrapServer", "localhost:9092",
+                "Props", Map.of(
+                    "auto.commit", false
+                )
+            );
+        }
+    }
+
+
     /**
      * Builder of index metadata.
      *
@@ -1474,6 +1508,10 @@ public Builder settings(Settings settings) {
             return this;
         }
 
+        public Builder ingestionSource(IngestionSource source) {
+            //
+        }
+
         public MappingMetadata mapping() {
             return mappings.get(MapperService.SINGLE_MAPPING_NAME);
         }
@@ -1956,6 +1994,8 @@ public static IndexMetadata fromXContent(XContentParser parser) throws IOExcepti
                         parser.skipChildren();
                     } else if (CONTEXT_KEY.equals(currentFieldName)) {
                         builder.context(Context.fromXContent(parser));
+                    } else if (KEY_INGESTION_SOURCE.equals(currentFieldName)) {
+                        builder.ingestionSourceConfig(IngestionSource.fromXcontent(parser));
                     } else {
                         // assume it's custom index metadata
                         builder.putCustom(currentFieldName, parser.mapStrings());

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

@@ -741,7 +741,8 @@ public IndexService newIndexService(
                 remoteStoreSettings,
                 fileCache,
                 compositeIndexSettings,
-                replicator
+                replicator,
+                ingestionSourceFactory
             );
             success = true;
             return indexService;

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

@@ -662,6 +662,8 @@ protected void closeInternal() {
                 path
             );
             eventListener.onStoreCreated(shardId);
+
+            IngestionSource ingestionSource = ingestionSourceFactory.create(shardId);
             indexShard = new IndexShard(
                 routing,
                 this.indexSettings,
@@ -691,7 +693,8 @@ protected void closeInternal() {
                 recoverySettings,
                 remoteStoreSettings,
                 seedRemote,
-                discoveryNodes
+                discoveryNodes,
+                ingestionSource
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -343,6 +343,7 @@ public class IndicesService extends AbstractLifecycleComponent
     private final Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders;
     private final Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories;
     private final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories;
+    private final Map<String, IngestionSourceFactoryProvider> ingestionSourceFactoryProviders;
     final AbstractRefCounted indicesRefCount; // pkg-private for testing
     private final CountDownLatch closeLatch = new CountDownLatch(1);
     private volatile boolean idFieldDataEnabled;
@@ -949,6 +950,9 @@ private synchronized IndexService createIndexService(
             indexCreationContext
         );
 
+        IndexMetadata.IngestionSourceConfig ingestionSource = indexMetadata.getIngestionSource();
+        IngestionSourceFactory ingestionSourceFactory = ingestionSourceFactoryProviderRegistry.get(ingestionSource.type).create(ingestionSource.config);
+
         final IndexModule indexModule = new IndexModule(
             idxSettings,
             analysisRegistry,
@@ -959,7 +963,8 @@ private synchronized IndexService createIndexService(
             indexNameExpressionResolver,
             recoveryStateFactories,
             fileCache,
-            compositeIndexSettings
+            compositeIndexSettings,
+            ingestionSourceFactory
         );
         for (IndexingOperationListener operationListener : indexingOperationListeners) {
             indexModule.addIndexOperationListener(operationListener);

server/src/main/java/org/opensearch/node/Node.java

@@ -919,6 +919,10 @@ protected Node(
             CacheModule cacheModule = new CacheModule(pluginsService.filterPlugins(CachePlugin.class), settings);
             CacheService cacheService = cacheModule.getCacheService();
             final SegmentReplicator segmentReplicator = new SegmentReplicator(threadPool);
+            Map<String, IngestionSourceFactoryProvider> ingestionSourceFactoryProviders = new HashMap<>();
+            pluginsService.filterPlugins(IngestionSourcePlugin.class)
+                .stream()
+                .forEach(p -> ingestionSourceFactoryProviders.putAll(p.getIngestionSourceFactoryProviders()));
             final IndicesService indicesService = new IndicesService(
                 settings,
                 pluginsService,
@@ -949,7 +953,8 @@ protected Node(
                 remoteStoreSettings,
                 fileCache,
                 compositeIndexSettings,
-                segmentReplicator::startReplication
+                segmentReplicator::startReplication,
+                ingestionSourceFactoryProviders
             );
 
             final IngestService ingestService = new IngestService(