Skip to content

Commit 881fc6c

Browse files
samxbrsamxbr
committed
Handle individual doc parsing failure gracefully in bulk request with pipeline (elastic#138624)
(cherry picked from commit 933354b)
1 parent 50f58de commit 881fc6c

File tree

4 files changed

+190
-1
lines changed

4 files changed

+190
-1
lines changed

‎docs/changelog/138624.yaml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
pr: 138624
2+
summary: Handle individual doc parsing failure in bulk request with pipeline
3+
area: Ingest Node
4+
type: bug
5+
issues:
6+
- 138445

‎server/src/internalClusterTest/java/org/elasticsearch/ingest/IngestClientIT.java‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -378,6 +378,92 @@ public void testPipelineProcessorOnFailure() throws Exception {
378378
assertThat(inserted.get("readme"), equalTo("pipeline with id [3] is a bad pipeline"));
379379
}
380380

381+
public void testBulkRequestWithInvalidJsonAndPipeline() throws Exception {
382+
// Test that when a document with invalid JSON is in a bulk request with a pipeline,
383+
// the invalid document fails gracefully without causing the entire bulk request to fail.
384+
// This tests the fix for https://github.com/elastic/elasticsearch/issues/138445
385+
386+
createIndex("test_index");
387+
388+
putJsonPipeline(
389+
"test-pipeline",
390+
(builder, params) -> builder.field("description", "test pipeline")
391+
.startArray("processors")
392+
.startObject()
393+
.startObject("test")
394+
.endObject()
395+
.endObject()
396+
.endArray()
397+
);
398+
399+
// Create a bulk request with valid and invalid documents
400+
BulkRequest bulkRequest = new BulkRequest();
401+
402+
// Valid document
403+
IndexRequest validRequest = new IndexRequest("test_index").id("valid_doc");
404+
validRequest.source("{\"valid\":\"test\"}", XContentType.JSON);
405+
validRequest.setPipeline("test-pipeline");
406+
bulkRequest.add(validRequest);
407+
408+
// Invalid document with missing closing brace
409+
IndexRequest invalidRequest = new IndexRequest("test_index").id("invalid_doc");
410+
invalidRequest.source("{\"invalid\":\"json\"", XContentType.JSON);
411+
invalidRequest.setPipeline("test-pipeline");
412+
bulkRequest.add(invalidRequest);
413+
414+
// Invalid document with duplicate fields
415+
IndexRequest invalidRequest2 = new IndexRequest("test_index").id("invalid_doc2");
416+
invalidRequest2.source("{\"invalid\":\"json\", \"invalid\":\"json\"}", XContentType.JSON);
417+
invalidRequest2.setPipeline("test-pipeline");
418+
bulkRequest.add(invalidRequest2);
419+
420+
// Another valid document
421+
IndexRequest validRequest2 = new IndexRequest("test_index").id("valid_doc2");
422+
validRequest2.source("{\"valid\":\"test2\"}", XContentType.JSON);
423+
validRequest2.setPipeline("test-pipeline");
424+
bulkRequest.add(validRequest2);
425+
426+
BulkResponse response = client().bulk(bulkRequest).actionGet();
427+
428+
// The bulk request should succeed
429+
assertThat(response.hasFailures(), is(true));
430+
assertThat(response.getItems().length, equalTo(4));
431+
432+
// First document should succeed
433+
BulkItemResponse item0 = response.getItems()[0];
434+
assertThat(item0.isFailed(), is(false));
435+
assertThat(item0.getResponse().getId(), equalTo("valid_doc"));
436+
assertThat(item0.getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
437+
438+
// Second document should fail
439+
BulkItemResponse item1 = response.getItems()[1];
440+
assertThat(item1.isFailed(), is(true));
441+
assertThat(item1.getFailure().getStatus(), equalTo(org.elasticsearch.rest.RestStatus.BAD_REQUEST));
442+
assertThat(item1.getFailure().getCause(), instanceOf(IllegalArgumentException.class));
443+
444+
// Third document should fail
445+
BulkItemResponse item2 = response.getItems()[2];
446+
assertThat(item2.isFailed(), is(true));
447+
assertThat(item2.getFailure().getStatus(), equalTo(org.elasticsearch.rest.RestStatus.BAD_REQUEST));
448+
assertThat(item2.getFailure().getCause(), instanceOf(IllegalArgumentException.class));
449+
450+
// Fourth document should succeed
451+
BulkItemResponse item3 = response.getItems()[3];
452+
assertThat(item3.isFailed(), is(false));
453+
assertThat(item3.getResponse().getId(), equalTo("valid_doc2"));
454+
assertThat(item3.getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
455+
456+
// Verify that the valid documents were indexed
457+
assertThat(client().prepareGet("test_index", "valid_doc").get().isExists(), is(true));
458+
assertThat(client().prepareGet("test_index", "valid_doc2").get().isExists(), is(true));
459+
// Verify that the invalid documents were not indexed
460+
assertThat(client().prepareGet("test_index", "invalid_doc").get().isExists(), is(false));
461+
assertThat(client().prepareGet("test_index", "invalid_doc2").get().isExists(), is(false));
462+
463+
// cleanup
464+
deletePipeline("test-pipeline");
465+
}
466+
381467
public static class ExtendedIngestTestPlugin extends IngestTestPlugin {
382468

383469
@Override

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

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -878,7 +878,20 @@ protected void doRun() {
878878
}
879879
final int slot = i;
880880
final Releasable ref = refs.acquire();
881-
final IngestDocument ingestDocument = newIngestDocument(indexRequest);
881+
final IngestDocument ingestDocument;
882+
try {
883+
ingestDocument = newIngestDocument(indexRequest);
884+
} catch (Exception e) {
885+
// Document parsing failed (e.g. invalid JSON). Handle this gracefully
886+
// by marking this document as failed and continuing with other documents.
887+
final long ingestTimeInNanos = System.nanoTime() - startTimeInNanos;
888+
totalMetrics.postIngest(ingestTimeInNanos);
889+
totalMetrics.ingestFailed();
890+
ref.close();
891+
i++;
892+
onFailure.apply(slot, e, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
893+
continue;
894+
}
882895
final org.elasticsearch.script.Metadata originalDocumentMetadata = ingestDocument.getMetadata().clone();
883896
// the document listener gives us three-way logic: a document can fail processing (1), or it can
884897
// be successfully processed. a successfully processed document can be kept (2) or dropped (3).

‎server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@
7373
import org.elasticsearch.test.MockLog;
7474
import org.elasticsearch.threadpool.ThreadPool;
7575
import org.elasticsearch.xcontent.XContentBuilder;
76+
import org.elasticsearch.xcontent.XContentParseException;
7677
import org.elasticsearch.xcontent.XContentType;
7778
import org.elasticsearch.xcontent.cbor.CborXContent;
7879
import org.junit.Before;
@@ -1723,6 +1724,89 @@ public void testBulkRequestExecutionWithFailures() throws Exception {
17231724
verify(completionHandler, times(1)).accept(Thread.currentThread(), null);
17241725
}
17251726

1727+
public void testBulkRequestExecutionWithInvalidJsonDocument() throws Exception {
1728+
// Test that when a document with invalid JSON (e.g., duplicate keys) is in a bulk request with a pipeline,
1729+
// the invalid document fails gracefully without causing the entire bulk request to fail.
1730+
BulkRequest bulkRequest = new BulkRequest();
1731+
String pipelineId = "_id";
1732+
1733+
// Valid document that should succeed
1734+
IndexRequest validRequest = new IndexRequest("_index").id("valid").setPipeline(pipelineId).setFinalPipeline("_none");
1735+
validRequest.source(Requests.INDEX_CONTENT_TYPE, "field1", "value1");
1736+
validRequest.setListExecutedPipelines(true);
1737+
bulkRequest.add(validRequest);
1738+
1739+
// Invalid document with missing closing brace
1740+
String invalidJson = "{\"invalid\":\"json\"";
1741+
IndexRequest invalidRequest = new IndexRequest("_index").id("invalid").setPipeline(pipelineId).setFinalPipeline("_none");
1742+
invalidRequest.source(new BytesArray(invalidJson), XContentType.JSON);
1743+
bulkRequest.add(invalidRequest);
1744+
1745+
// Another valid document that should succeed
1746+
IndexRequest validRequest2 = new IndexRequest("_index").id("valid2").setPipeline(pipelineId).setFinalPipeline("_none");
1747+
validRequest2.source(Requests.INDEX_CONTENT_TYPE, "field2", "value2");
1748+
validRequest2.setListExecutedPipelines(true);
1749+
bulkRequest.add(validRequest2);
1750+
1751+
// Invalid document with duplicated keys
1752+
String invalidJson2 = "{\"@timestamp\":\"2024-06-01T00:00:00Z\",\"@timestamp\":\"2024-06-01T00:00:00Z\"}";
1753+
IndexRequest invalidRequest2 = new IndexRequest("_index").id("invalid").setPipeline(pipelineId).setFinalPipeline("_none");
1754+
invalidRequest2.source(new BytesArray(invalidJson2), XContentType.JSON);
1755+
bulkRequest.add(invalidRequest2);
1756+
1757+
final Processor processor = mock(Processor.class);
1758+
when(processor.getType()).thenReturn("mock");
1759+
when(processor.getTag()).thenReturn("mockTag");
1760+
doAnswer(args -> {
1761+
BiConsumer<IngestDocument, Exception> handler = args.getArgument(1);
1762+
handler.accept(RandomDocumentPicks.randomIngestDocument(random()), null);
1763+
return null;
1764+
}).when(processor).execute(any(), any());
1765+
1766+
IngestService ingestService = createWithProcessors(Map.of("mock", (factories, tag, description, config, projectId) -> processor));
1767+
PutPipelineRequest putRequest = putJsonPipelineRequest("_id", "{\"processors\": [{\"mock\" : {}}]}");
1768+
var projectId = randomProjectIdOrDefault();
1769+
ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT)
1770+
.putProjectMetadata(ProjectMetadata.builder(projectId).build())
1771+
.build();
1772+
ClusterState previousClusterState = clusterState;
1773+
clusterState = executePut(projectId, putRequest, clusterState);
1774+
ingestService.applyClusterState(new ClusterChangedEvent("", clusterState, previousClusterState));
1775+
1776+
TriConsumer<Integer, Exception, IndexDocFailureStoreStatus> requestItemErrorHandler = mock();
1777+
final BiConsumer<Thread, Exception> onCompletion = mock();
1778+
1779+
ingestService.executeBulkRequest(
1780+
projectId,
1781+
4,
1782+
bulkRequest.requests(),
1783+
indexReq -> {},
1784+
(s) -> false,
1785+
(slot, targetIndex, e) -> fail("Should not redirect to failure store"),
1786+
requestItemErrorHandler,
1787+
onCompletion,
1788+
EsExecutors.DIRECT_EXECUTOR_SERVICE
1789+
);
1790+
// The invalid documents should fail with a parsing error
1791+
verify(requestItemErrorHandler).apply(
1792+
eq(1), // slot 1 is the invalid document
1793+
argThat(e -> e instanceof XContentParseException),
1794+
eq(IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN)
1795+
);
1796+
verify(requestItemErrorHandler).apply(
1797+
eq(3), // slot 3 is the other invalid document
1798+
argThat(e -> e instanceof XContentParseException),
1799+
eq(IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN)
1800+
);
1801+
1802+
// The bulk listener should still be called with success
1803+
verify(onCompletion).accept(any(), eq(null));
1804+
assertStats(ingestService.stats().totalStats(), 4, 2, 0);
1805+
// Verify that the valid documents were processed (they should have their pipelines executed)
1806+
assertThat(validRequest.getExecutedPipelines(), equalTo(List.of(pipelineId)));
1807+
assertThat(validRequest2.getExecutedPipelines(), equalTo(List.of(pipelineId)));
1808+
}
1809+
17261810
public void testExecuteFailureRedirection() throws Exception {
17271811
final CompoundProcessor processor = mockCompoundProcessor();
17281812
IngestService ingestService = createWithProcessors(

0 commit comments

Comments
 (0)