address comments

Signed-off-by: Yupeng Fu <yupeng@uber.com>

Author: yupeng9

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java

@@ -45,7 +45,7 @@
 /**
  * Integration test for Kafka ingestion
  */
-@ThreadLeakLingering(linger = 5000) // wait for container pull thread to die
+@ThreadLeakLingering(linger = 15000) // wait for container pull thread to die
 public class IngestFromKafkaIT extends OpenSearchIntegTestCase {
     static final String topicName = "test";
 

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

@@ -666,8 +666,7 @@ public void validate(final String value, final Map<Setting<?>, Object> settings)
                 // TODO: validate this with the ingestion source params
             }
         },
-        Property.IndexScope,
-        Property.Dynamic
+        Property.IndexScope
     );
 
     /**
@@ -704,11 +703,6 @@ public void validate(final String value, final Map<Setting<?>, Object> settings)
         }, Property.IndexScope)
     );
 
-    /**
-     * Used to specify the params for the ingestion index.
-     */
-    public static final String SETTING_INGESTION_SOURCE_PARAMS = "index.ingestion_source.params";
-
     /**
      * an internal index format description, allowing us to find out if this index is upgraded or needs upgrading
      */

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

@@ -40,10 +40,6 @@ public Map<String, Object> params() {
         return params;
     }
 
-    public void params(Map<String, Object> params) {
-        this.params = params;
-    }
-
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

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

@@ -80,7 +80,7 @@
 import java.util.function.BiFunction;
 import java.util.function.UnaryOperator;
 
-import static org.opensearch.index.engine.NoOpEngine.EMPTY_TRANSLOG_SNAPSHOT;
+import static org.opensearch.index.translog.Translog.EMPTY_TRANSLOG_SNAPSHOT;
 
 /**
  * IngestionEngine is an engine that ingests data from a stream source.

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

@@ -58,6 +58,8 @@
 import java.util.Map;
 import java.util.function.Function;
 
+import static org.opensearch.index.translog.Translog.EMPTY_TRANSLOG_SNAPSHOT;
+
 /**
  * NoOpEngine is an engine implementation that does nothing but the bare minimum
  * required in order to have an engine. All attempts to do something (search,
@@ -72,21 +74,6 @@ public final class NoOpEngine extends ReadOnlyEngine {
     private final SegmentsStats segmentsStats;
     private final DocsStats docsStats;
 
-    static final Translog.Snapshot EMPTY_TRANSLOG_SNAPSHOT = new Translog.Snapshot() {
-        @Override
-        public void close() {}
-
-        @Override
-        public int totalOperations() {
-            return 0;
-        }
-
-        @Override
-        public Translog.Operation next() {
-            return null;
-        }
-    };
-
     public NoOpEngine(EngineConfig config) {
         super(config, null, null, true, Function.identity(), true);
         this.segmentsStats = new SegmentsStats();

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

@@ -66,7 +66,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
-import static org.opensearch.index.engine.NoOpEngine.EMPTY_TRANSLOG_SNAPSHOT;
+import static org.opensearch.index.translog.Translog.EMPTY_TRANSLOG_SNAPSHOT;
 
 /**
  * A basic read-only engine that allows switching a shard to be true read-only temporarily or permanently.

server/src/main/java/org/opensearch/index/translog/Translog.java

@@ -959,6 +959,21 @@ public int hashCode() {
         }
     }
 
+    public static final Translog.Snapshot EMPTY_TRANSLOG_SNAPSHOT = new Translog.Snapshot() {
+        @Override
+        public void close() {}
+
+        @Override
+        public int totalOperations() {
+            return 0;
+        }
+
+        @Override
+        public Translog.Operation next() {
+            return null;
+        }
+    };
+
     /**
      * A snapshot of the transaction log, allows to iterate over all the transaction log operations.
      *