[graytaylor0]

Description

[Describe what this change achieves]

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Event handles not released on failure pipeline send sendToFailurePipeline builds records and calls failurePipeline.sendEvents(records) but never releases the event handles of the DlqObjects. In the previous code path, when failurePipeline was used, the acknowledgement was presumably transferred via the Event, but DlqObjects whose event is null are silently dropped here with no releaseEventHandle call, which can leak acknowledgement handles. Also, when sendToFailurePipelineOrRelease is invoked from the S3 DLQ fallback after a successful partial write attempt, the handles routed through the failure pipeline are not explicitly released/acknowledged, which may differ from the behavior of the direct DLQ-writer success path. private void sendToFailurePipeline(final HeadlessPipeline failurePipeline, final List<DlqObject> dlqObjects) { final List<Record<Event>> records = new ArrayList<>(); for (final DlqObject dlqObject : dlqObjects) { final Event event = dlqObject.getEvent(); if (event != null) { event.updateFailureMetadata() .withPluginId(dlqObject.getPluginId()) .withPluginName(dlqObject.getPluginName()) .withPipelineName(dlqObject.getPipelineName()) .withErrorMessage(((FailedDlqData) dlqObject.getFailedData()).getMessage()) .with("status", ((FailedDlqData) dlqObject.getFailedData()).getStatus()) .with("index", ((FailedDlqData) dlqObject.getFailedData()).getIndex()) .with("indexId", ((FailedDlqData) dlqObject.getFailedData()).getIndexId()); records.add(new Record<>(event)); } } failurePipeline.sendEvents(records); } Precedence change for failurePipeline vs dlqWriter Previously, when a failurePipeline was configured, it took precedence over dlqFileWriter/dlqWriter. In the new logic, dlqFileWriter and dlqWriter are tried first and failurePipeline is only used as a fallback on IOException or when no DLQ writer is configured. This is a user-visible behavior change for existing deployments that configure both a DLQ and a failure pipeline — events will now go to the DLQ instead of the failure pipeline. Confirm this is the intended semantics and documented. if (dlqFileWriter != null) { dlqObjects.forEach(dlqObject -> { final FailedDlqData failedDlqData = (FailedDlqData) dlqObject.getFailedData(); final String message = failure == null ? failedDlqData.getMessage() : failure.getMessage(); try { dlqFileWriter.write(String.format("{\"Document\": [%s], \"failure\": %s}\n", BulkOperationWriter.dlqObjectToString(dlqObject), message)); dlqObject.releaseEventHandle(true); } catch (final IOException e) { LOG.error(NOISY, "Failed to write a document to the DLQ", e); dlqObject.releaseEventHandle(false); } }); } else if (dlqWriter != null) { try { dlqWriter.write(dlqObjects, pipeline, PLUGIN_NAME); dlqObjects.forEach(dlqObject -> dlqObject.releaseEventHandle(true)); } catch (final IOException e) { LOG.error(NOISY, "Failed to write to S3 DLQ, falling back to failure pipeline", e); sendToFailurePipelineOrRelease(failurePipeline, dlqObjects, failure); } } else if (failurePipeline != null) { sendToFailurePipeline(failurePipeline, dlqObjects); } else { dlqObjects.forEach(dlqObject -> { final FailedDlqData failedDlqData = (FailedDlqData) dlqObject.getFailedData(); final String message = failure == null ? failedDlqData.getMessage() : failure.getMessage(); LOG.warn("Document failed to write to OpenSearch with error code {}. Configure a DLQ to save failed documents. Error: {}", failedDlqData.getStatus(), message); dlqObject.releaseEventHandle(false); }); }

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Preserve failure pipeline precedence over DLQ writers The reordering of the DLQ handling priorities changes behavior: previously failurePipeline had the highest priority, now it is only consulted after dlqFileWriter and dlqWriter. If a user configured both a failure pipeline and a DLQ file/S3 writer, events will no longer be routed to the failure pipeline. Confirm this precedence change is intended, or restore failurePipeline as the primary route when configured. data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java [517-540] private void logFailureForDlqObjects(final HeadlessPipeline failurePipeline, final List<DlqObject> dlqObjects, final Throwable failure) { + if (failurePipeline != null) { + sendToFailurePipeline(failurePipeline, dlqObjects); + return; + } if (dlqFileWriter != null) { Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern about the precedence change where failurePipeline is no longer the primary route when a DLQ writer is also configured. However, this may be intentional based on the PR design, so it's primarily a verification suggestion.	Low
Possible issue	Release event handles for skipped DLQ objects When event is null, the corresponding dlqObject's event handle is never released, causing acknowledgement/handle leaks. Also, the handles for events successfully forwarded to the failure pipeline are not released here, which may leak if the downstream failure pipeline doesn't own them. Explicitly release handles for skipped (null-event) DLQ objects. data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java [551-568] private void sendToFailurePipeline(final HeadlessPipeline failurePipeline, final List<DlqObject> dlqObjects) { final List<Record<Event>> records = new ArrayList<>(); for (final DlqObject dlqObject : dlqObjects) { final Event event = dlqObject.getEvent(); if (event != null) { - ... + event.updateFailureMetadata() + .withPluginId(dlqObject.getPluginId()) + .withPluginName(dlqObject.getPluginName()) + .withPipelineName(dlqObject.getPipelineName()) + .withErrorMessage(((FailedDlqData) dlqObject.getFailedData()).getMessage()) + .with("status", ((FailedDlqData) dlqObject.getFailedData()).getStatus()) + .with("index", ((FailedDlqData) dlqObject.getFailedData()).getIndex()) + .with("indexId", ((FailedDlqData) dlqObject.getFailedData()).getIndexId()); records.add(new Record<>(event)); + } else { + dlqObject.releaseEventHandle(false); } } failurePipeline.sendEvents(records); } Suggestion importance[1-10]: 6 __ Why: Valid concern about potential handle leaks when event is null in the new sendToFailurePipeline method. The original code had the same behavior though, so this is a pre-existing issue rather than a regression.	Low