Add additional tests for delete by query

Author: lukewhiting

modules/streams/build.gradle

@@ -20,7 +20,7 @@ esplugin {
 
 restResources {
   restApi {
-    include '_common', 'streams', "bulk", "index", "ingest", "indices"
+    include '_common', 'streams', "bulk", "index", "ingest", "indices", "delete_by_query", "search"
   }
 }
 
@@ -39,4 +39,5 @@ artifacts {
 dependencies {
   testImplementation project(path: ':test:test-clusters')
   clusterModules project(':modules:ingest-common')
+  clusterModules project(':modules:reindex')
 }

modules/streams/src/yamlRestTest/java/org/elasticsearch/streams/StreamsYamlTestSuiteIT.java

@@ -32,6 +32,7 @@ public static Iterable<Object[]> parameters() throws Exception {
     public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
         .module("streams")
         .module("ingest-common")
+        .module("reindex")
         .feature(FeatureFlag.LOGS_STREAM)
         .build();
 

modules/streams/src/yamlRestTest/resources/rest-api-spec/test/streams/logs/20_substream_restrictions.yml

@@ -70,3 +70,87 @@
   - match: { items.0.index.status: 400 }
   - match: { items.0.index.error.type: "illegal_argument_exception" }
   - match: { items.0.index.error.reason: "Pipelines can't re-route documents to child streams, but pipeline [reroute-to-logs-foo] tried to reroute this document from index [bad-index] to index [logs.foo]. Reroute history: bad-index" }
+
+---
+"Check Bulk Index With Script Processor To Substream Is Rejected":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      ingest.put_pipeline:
+        id: "script-to-logs-foo"
+        body:
+          processors:
+            - script:
+                source: "ctx._index = 'logs.foo'"
+  - do:
+      indices.create:
+        index: "bad-index-script"
+        body:
+          settings:
+            index.default_pipeline: "script-to-logs-foo"
+  - do:
+      bulk:
+        body: |
+          { "index": { "_index": "bad-index-script" } }
+          { "foo": "bar" }
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: "illegal_argument_exception" }
+  - match: { items.0.index.error.reason: "Pipelines can't re-route documents to child streams, but pipeline [script-to-logs-foo] tried to reroute this document from index [bad-index-script] to index [logs.foo]. Reroute history: bad-index-script" }
+
+---
+"Check Delete By Query Directly On Substream After Reroute Succeeds":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      ingest.put_pipeline:
+        id: "reroute-to-logs-foo-success"
+        body:
+          processors:
+            - reroute:
+                destination: "logs.foo"
+  - do:
+      indices.create:
+        index: "logs"
+        body:
+          settings:
+            index.default_pipeline: "reroute-to-logs-foo-success"
+  - do:
+      bulk:
+        refresh: true
+        body: |
+          { "index": { "_index": "logs" } }
+          { "foo": "bar", "baz": "qux" }
+  - match: { errors: false }
+  - match: { items.0.index.status: 201 }
+
+  - do:
+      delete_by_query:
+        index: logs.foo
+        refresh: true
+        body:
+          query:
+            match:
+              foo: "bar"
+  - match: { deleted: 1 }
+  - match: { total: 1 }
+
+  - do:
+      search:
+        index: logs.foo
+        body:
+          query:
+            match_all: {}
+  - match: { hits.total.value: 0 }

server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java

@@ -405,7 +405,7 @@ private void applyPipelinesAndDoInternalExecute(
         // Validate child stream writes before processing pipelines
         ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(clusterService.state());
         Set<StreamType> enabledStreamTypes = Arrays.stream(StreamType.values())
-            .filter(t -> StreamType.streamTypeIsEnabled(t, projectMetadata))
+            .filter(t -> t.streamTypeIsEnabled(projectMetadata))
             .collect(Collectors.toCollection(() -> EnumSet.noneOf(StreamType.class)));
 
         BulkRequestModifier bulkRequestModifier = new BulkRequestModifier(bulkRequest);

server/src/main/java/org/elasticsearch/common/streams/StreamType.java

@@ -26,9 +26,9 @@ public String getStreamName() {
         return streamName;
     }
 
-    public static boolean streamTypeIsEnabled(StreamType streamType, ProjectMetadata projectMetadata) {
+    public boolean streamTypeIsEnabled(ProjectMetadata projectMetadata) {
         StreamsMetadata metadata = projectMetadata.custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
-        return switch (streamType) {
+        return switch (this) {
             case LOGS -> metadata.isLogsEnabled();
         };
     }

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -1200,7 +1200,7 @@ private void executePipelines(
                     }
 
                     for (StreamType streamType : StreamType.values()) {
-                        if (StreamType.streamTypeIsEnabled(streamType, project)) {
+                        if (streamType.streamTypeIsEnabled(project)) {
                             if (newIndex.startsWith(streamType.getStreamName() + ".")
                                 && ingestDocument.getIndexHistory().stream().noneMatch(s -> s.equals(streamType.getStreamName()))) {
                                 exceptionHandler.accept(