Additional PR changes and cleanup

Author: lukewhiting

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

@@ -171,12 +171,12 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
 
     @Override
     public boolean equals(Object o) {
-    return o == this || (o instanceof BulkResponse that
-        && tookInMillis == that.tookInMillis
-        && ingestTookInMillis == that.ingestTookInMillis
-        && Arrays.equals(responses, that.responses)
-        && Objects.equals(incrementalState, that.incrementalState));
-}
+        return o == this
+            || (o instanceof BulkResponse that
+                && tookInMillis == that.tookInMillis
+                && ingestTookInMillis == that.ingestTookInMillis
+                && Arrays.equals(responses, that.responses)
+                && Objects.equals(incrementalState, that.incrementalState));
     }
 
     @Override

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

@@ -413,16 +413,15 @@ private void applyPipelinesAndDoInternalExecute(
         for (StreamType streamType : enabledStreamTypes) {
             for (int i = 0; i < bulkRequest.requests.size(); i++) {
                 DocWriteRequest<?> req = bulkRequestModifier.bulkRequest.requests.get(i);
-                String StreamTypePrefix = streamType.getStreamName() + ".";
 
-                if (req instanceof IndexRequest ir && ir.index().startsWith(prefix) && ir.isPipelineResolved() == false) {
+                if (req instanceof IndexRequest ir && streamType.matchesStreamPrefix(req.index()) && ir.isPipelineResolved() == false) {
                     IllegalArgumentException e = new IllegalArgumentException(
                         "Direct writes to child streams are prohibited. Index directly into the ["
                             + streamType.getStreamName()
                             + "] stream instead"
                     );
                     Boolean failureStoreEnabled = resolveFailureStore(req.index(), projectMetadata, threadPool.absoluteTimeInMillis());
-                    if (Boolean.TRUE.equals(failureStore)) {
+                    if (Boolean.TRUE.equals(failureStoreEnabled)) {
                         bulkRequestModifier.markItemForFailureStore(i, req.index(), e);
                     } else {
                         bulkRequestModifier.markItemAsFailed(i, e, IndexDocFailureStoreStatus.NOT_ENABLED);

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

@@ -33,4 +33,8 @@ public boolean streamTypeIsEnabled(ProjectMetadata projectMetadata) {
         };
     }
 
+    public boolean matchesStreamPrefix(String indexName) {
+        return indexName.startsWith(streamName + ".");
+    }
+
 }

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

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