Putting the ingest otel processor behind the logs stream feature flag (#129667) (#129783)

Author: masseyke

modules/ingest-otel/src/main/java/org/elasticsearch/ingest/otel/NormalizeForStreamPlugin.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.otel;
 
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.plugins.IngestPlugin;
 import org.elasticsearch.plugins.Plugin;
@@ -19,6 +20,10 @@ public class NormalizeForStreamPlugin extends Plugin implements IngestPlugin {
 
     @Override
     public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
-        return Map.of(NormalizeForStreamProcessor.TYPE, new NormalizeForStreamProcessor.Factory());
+        if (DataStream.LOGS_STREAM_FEATURE_FLAG) {
+            return Map.of(NormalizeForStreamProcessor.TYPE, new NormalizeForStreamProcessor.Factory());
+        } else {
+            return Map.of();
+        }
     }
 }

modules/ingest-otel/src/yamlRestTest/java/org/elasticsearch/ingest/otel/IngestOtelClientYamlTestSuiteIT.java

@@ -13,6 +13,7 @@
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.FeatureFlag;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
 import org.junit.ClassRule;
@@ -24,7 +25,10 @@ public IngestOtelClientYamlTestSuiteIT(@Name("yaml") ClientYamlTestCandidate tes
     }
 
     @ClassRule
-    public static ElasticsearchCluster cluster = ElasticsearchCluster.local().module("ingest-otel").build();
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .module("ingest-otel")
+        .feature(FeatureFlag.LOGS_STREAM)
+        .build();
 
     @Override
     protected String getTestRestCluster() {

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/FeatureFlag.java

@@ -18,7 +18,8 @@
 public enum FeatureFlag {
     TIME_SERIES_MODE("es.index_mode_feature_flag_registered=true", Version.fromString("8.0.0"), null),
     SUB_OBJECTS_AUTO_ENABLED("es.sub_objects_auto_feature_flag_enabled=true", Version.fromString("8.16.0"), null),
-    INFERENCE_CUSTOM_SERVICE_ENABLED("es.inference_custom_service_feature_flag_enabled=true", Version.fromString("8.19.0"), null);
+    INFERENCE_CUSTOM_SERVICE_ENABLED("es.inference_custom_service_feature_flag_enabled=true", Version.fromString("8.19.0"), null),
+    LOGS_STREAM("es.logs_stream_feature_flag_enabled=true", Version.fromString("8.19.0"), null);
 
     public final String systemProperty;
     public final Version from;