SNAPSHOT - Refactor to move substream check earlier in indexing process

Author: lukewhiting

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

@@ -43,7 +43,6 @@
 import org.elasticsearch.cluster.routing.IndexRouting;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.collect.Iterators;
-import org.elasticsearch.common.streams.StreamType;
 import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
@@ -61,8 +60,6 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -74,7 +71,6 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.LongSupplier;
-import java.util.stream.Collectors;
 
 import static org.elasticsearch.action.bulk.TransportBulkAction.LAZY_ROLLOVER_ORIGIN;
 import static org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.EXCLUDED_DATA_STREAMS_KEY;
@@ -283,29 +279,29 @@ private long buildTookInMillis(long startTimeNanos) {
     }
 
     private Map<ShardId, List<BulkItemRequest>> groupBulkRequestsByShards(ClusterState clusterState) {
-        ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(clusterState);
-
-        Set<StreamType> enabledStreamTypes = Arrays.stream(StreamType.values())
-            .filter(t -> streamsPermissionsUtils.streamTypeIsEnabled(t, projectMetadata))
-            .collect(Collectors.toCollection(() -> EnumSet.noneOf(StreamType.class)));
-
-        for (StreamType streamType : enabledStreamTypes) {
-            for (int i = 0; i < bulkRequest.requests.size(); i++) {
-                DocWriteRequest<?> req = bulkRequest.requests.get(i);
-                String prefix = streamType.getStreamName() + ".";
-                if (req != null && req.index().startsWith(prefix)) {
-                    IllegalArgumentException exception = new IllegalArgumentException(
-                        "Writes to child stream ["
-                            + req.index()
-                            + "] are not allowed, use the parent stream instead: ["
-                            + streamType.getStreamName()
-                            + "]"
-                    );
-                    IndexDocFailureStoreStatus failureStoreStatus = processFailure(new BulkItemRequest(i, req), projectMetadata, exception);
-                    addFailureAndDiscardRequest(req, i, req.index(), exception, failureStoreStatus);
-                }
-            }
-        }
+        // ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(clusterState);
+        //
+        // Set<StreamType> enabledStreamTypes = Arrays.stream(StreamType.values())
+        // .filter(t -> streamsPermissionsUtils.streamTypeIsEnabled(t, projectMetadata))
+        // .collect(Collectors.toCollection(() -> EnumSet.noneOf(StreamType.class)));
+        //
+        // for (StreamType streamType : enabledStreamTypes) {
+        // for (int i = 0; i < bulkRequest.requests.size(); i++) {
+        // DocWriteRequest<?> req = bulkRequest.requests.get(i);
+        // String prefix = streamType.getStreamName() + ".";
+        // if (req != null && req.index().startsWith(prefix)) {
+        // IllegalArgumentException exception = new IllegalArgumentException(
+        // "Writes to child stream ["
+        // + req.index()
+        // + "] are not allowed, use the parent stream instead: ["
+        // + streamType.getStreamName()
+        // + "]"
+        // );
+        // IndexDocFailureStoreStatus failureStoreStatus = processFailure(new BulkItemRequest(i, req), projectMetadata, exception);
+        // addFailureAndDiscardRequest(req, i, req.index(), exception, failureStoreStatus);
+        // }
+        // }
+        // }
 
         return groupRequestsByShards(
             clusterState,

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

@@ -31,6 +31,8 @@
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.streams.StreamType;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.Releasable;
@@ -44,12 +46,16 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.LongSupplier;
+import java.util.stream.Collectors;
 
 /**
  * This is an abstract base class for bulk actions. It traverses all indices that the request gets routed to, executes all applicable
@@ -396,6 +402,31 @@ private void applyPipelinesAndDoInternalExecute(
         ActionListener<BulkResponse> listener
     ) throws IOException {
         final long relativeStartTimeNanos = relativeTimeNanos();
+
+        // Validate child stream writes before processing pipelines
+        ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(clusterService.state());
+        Set<StreamType> enabledStreamTypes = Arrays.stream(StreamType.values())
+            .filter(t -> StreamsPermissionsUtils.getInstance().streamTypeIsEnabled(t, projectMetadata))
+            .collect(Collectors.toCollection(() -> EnumSet.noneOf(StreamType.class)));
+
+        BulkRequestModifier bulkRequestModifier = new BulkRequestModifier(bulkRequest);
+
+        for (StreamType streamType : enabledStreamTypes) {
+            for (int i = 0; i < bulkRequest.requests.size(); i++) {
+                DocWriteRequest<?> req = bulkRequest.requests.get(i);
+                String prefix = streamType.getStreamName() + ".";
+                if (req != null && req.index() != null && req.index().startsWith(prefix)) {
+                    IllegalArgumentException e = new IllegalArgumentException("Can't write to child stream");
+                    Boolean failureStore = resolveFailureStore(req.index(), projectMetadata, threadPool.absoluteTimeInMillis());
+                    if (failureStore != null && failureStore) {
+                        bulkRequestModifier.markItemForFailureStore(i, req.index(), e);
+                    } else {
+                        bulkRequestModifier.markItemAsFailed(i, e, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
+                    }
+                }
+            }
+        }
+
         if (applyPipelines(task, bulkRequest, executor, listener) == false) {
             doInternalExecute(task, bulkRequest, executor, listener, relativeStartTimeNanos);
         }