precommit

Author: jbaiera

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

@@ -43,8 +43,7 @@ public IndexRequest convert() throws IOException {
         // up that the parent data stream needs to be created.
         // One option is to make use of the eventual flag to perform an operation on the failure store. Ughh who would have thought the
         // dependencies would be swapped like that...
-        return new IndexRequest()
-            .index(targetIndexName)
+        return new IndexRequest().index(targetIndexName)
             .source(createSource())
             .opType(DocWriteRequest.OpType.CREATE)
             .setWriteToFailureStore(true);

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

@@ -969,21 +969,14 @@ public boolean isForceExecution() {
         );
     }
 
-    static boolean shouldStoreFailure(
-        String indexName,
-        Metadata metadata,
-        long epochMillis
-    ) {
-        return DataStream.isFailureStoreEnabled() && resolveFailureStoreFromMetadata(indexName, metadata, epochMillis)
-            .or(() -> resolveFailureStoreFromTemplate(indexName, metadata))
-            .orElse(false);
+    static boolean shouldStoreFailure(String indexName, Metadata metadata, long epochMillis) {
+        return DataStream.isFailureStoreEnabled()
+            && resolveFailureStoreFromMetadata(indexName, metadata, epochMillis).or(
+                () -> resolveFailureStoreFromTemplate(indexName, metadata)
+            ).orElse(false);
     }
 
-    private static Optional<Boolean> resolveFailureStoreFromMetadata(
-        String indexName,
-        Metadata metadata,
-        long epochMillis
-    ) {
+    private static Optional<Boolean> resolveFailureStoreFromMetadata(String indexName, Metadata metadata, long epochMillis) {
         if (indexName == null) {
             return Optional.empty();
         }
@@ -1009,10 +1002,7 @@ private static Optional<Boolean> resolveFailureStoreFromMetadata(
         return Optional.of(targetDataStream != null && targetDataStream.isFailureStore());
     }
 
-    private static Optional<Boolean> resolveFailureStoreFromTemplate(
-        String indexName,
-        Metadata metadata
-    ) {
+    private static Optional<Boolean> resolveFailureStoreFromTemplate(String indexName, Metadata metadata) {
         if (indexName == null) {
             return Optional.empty();
         }

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

@@ -670,12 +670,15 @@ void validatePipeline(Map<DiscoveryNode, IngestInfo> ingestInfos, String pipelin
     }
 
     private record IngestPipelinesExecutionResult(boolean success, boolean kept, Exception exception, String failedIndex) {}
+
     private static IngestPipelinesExecutionResult successResult() {
         return new IngestPipelinesExecutionResult(true, true, null, null);
     }
+
     private static IngestPipelinesExecutionResult discardResult() {
         return new IngestPipelinesExecutionResult(true, false, null, null);
     }
+
     private static IngestPipelinesExecutionResult failAndStoreFor(String index, Exception e) {
         return new IngestPipelinesExecutionResult(false, true, e, index);
     }
@@ -726,43 +729,40 @@ protected void doRun() {
                         final IngestDocument ingestDocument = newIngestDocument(indexRequest, documentParsingObserver);
                         // the document listener gives us three-way logic: a document can fail processing (1), or it can
                         // be successfully processed. a successfully processed document can be kept (2) or dropped (3).
-                        final ActionListener<IngestPipelinesExecutionResult> documentListener = ActionListener.runAfter(new ActionListener<>() {
-                            @Override
-                            public void onResponse(IngestPipelinesExecutionResult result) {
-                                assert result != null;
-                                if (result.success) {
-                                    if (result.kept == false) {
-                                        onDropped.accept(slot);
+                        final ActionListener<IngestPipelinesExecutionResult> documentListener = ActionListener.runAfter(
+                            new ActionListener<>() {
+                                @Override
+                                public void onResponse(IngestPipelinesExecutionResult result) {
+                                    assert result != null;
+                                    if (result.success) {
+                                        if (result.kept == false) {
+                                            onDropped.accept(slot);
+                                        }
+                                    } else {
+                                        // We were given a failure result in the onResponse method, so we must store the failure
+                                        // Recover the original document state, track a failed ingest, and pass it along
+                                        updateIndexRequestMetadata(indexRequest, ingestDocument.getOriginalMetadata());
+                                        totalMetrics.ingestFailed();
+                                        onStoreFailure.apply(slot, result.failedIndex, result.exception);
                                     }
-                                } else {
-                                    // We were given a failure result in the onResponse method, so we must store the failure
-                                    // Recover the original document state, track a failed ingest, and pass it along
-                                    updateIndexRequestMetadata(indexRequest, ingestDocument.getOriginalMetadata());
-                                    totalMetrics.ingestFailed();
-                                    onStoreFailure.apply(slot, result.failedIndex, result.exception);
                                 }
-                            }
 
-                            @Override
-                            public void onFailure(Exception e) {
-                                totalMetrics.ingestFailed();
-                                onFailure.accept(slot, e);
+                                @Override
+                                public void onFailure(Exception e) {
+                                    totalMetrics.ingestFailed();
+                                    onFailure.accept(slot, e);
+                                }
+                            },
+                            () -> {
+                                // regardless of success or failure, we always stop the ingest "stopwatch" and release the ref to indicate
+                                // that we're finished with this document
+                                final long ingestTimeInNanos = System.nanoTime() - startTimeInNanos;
+                                totalMetrics.postIngest(ingestTimeInNanos);
+                                ref.close();
                             }
-                        }, () -> {
-                            // regardless of success or failure, we always stop the ingest "stopwatch" and release the ref to indicate
-                            // that we're finished with this document
-                            final long ingestTimeInNanos = System.nanoTime() - startTimeInNanos;
-                            totalMetrics.postIngest(ingestTimeInNanos);
-                            ref.close();
-                        });
-
-                        executePipelines(
-                            pipelines,
-                            indexRequest,
-                            ingestDocument,
-                            shouldStoreFailure,
-                            documentListener
                         );
+
+                        executePipelines(pipelines, indexRequest, ingestDocument, shouldStoreFailure, documentListener);
                         indexRequest.setPipelinesHaveRun();
 
                         assert actionRequest.index() != null;

server/src/test/java/org/elasticsearch/action/bulk/FailureStoreDocumentTests.java

@@ -24,14 +24,9 @@
 public class FailureStoreDocumentTests extends ESTestCase {
 
     public void testFailureStoreDocumentConverstion() throws Exception {
-        IndexRequest source = new IndexRequest("original_index")
-            .routing("fake_routing")
+        IndexRequest source = new IndexRequest("original_index").routing("fake_routing")
             .id("1")
-            .source(JsonXContent.contentBuilder()
-                .startObject()
-                .field("key", "value")
-                .endObject()
-            );
+            .source(JsonXContent.contentBuilder().startObject().field("key", "value").endObject());
 
         // The exception will be wrapped for the test to make sure the converter correctly unwraps it
         Exception exception = new ElasticsearchException("Test exception please ignore");
@@ -64,8 +59,8 @@ public void testFailureStoreDocumentConverstion() throws Exception {
         assertThat(
             ObjectPath.eval("error.stack_trace", convertedRequest.sourceAsMap()),
             startsWith(
-                "org.elasticsearch.ElasticsearchException: Test exception please ignore\n" +
-                "\tat org.elasticsearch.action.bulk.FailureStoreDocumentTests.testFailureStoreDocumentConverstion"
+                "org.elasticsearch.ElasticsearchException: Test exception please ignore\n"
+                    + "\tat org.elasticsearch.action.bulk.FailureStoreDocumentTests.testFailureStoreDocumentConverstion"
             )
         );