fix(opensearch-sink): S3 DLQ takes precedence over failure pipeline

[graytaylor0]

Previously, when both a per-sink S3 DLQ and a global failure pipeline (dlq_pipeline) were configured, the failure pipeline always took precedence. This meant the S3 DLQ configured on the OpenSearch sink was never used.

This caused problems when the failure pipeline writes to the same OpenSearch cluster that is experiencing issues — the failure pipeline would also fail, blocking acknowledgements indefinitely (with acks enabled) or causing negative acks upstream.

New priority order for handling failed documents:

1. S3 DLQ file writer (dlqFileWriter) — if configured
2. S3 DLQ writer (dlqWriter) — if configured on the sink
3. Failure pipeline (dlq_pipeline) — fallback if no sink-level DLQ
4. Log warning and release(false) — nothing configured

If S3 DLQ write fails, fall back to the failure pipeline. If no failure pipeline exists either, release with false.

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 🔍

(Review updated until commit 4a99c9d)

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 fallback When dlqWriter.write throws IOException and the code falls back to sendToFailurePipelineOrRelease, if failurePipeline is non-null, sendToFailurePipeline is called but event handles are never released (neither true nor false). The previous code released with false on dlqWriter failure. This can cause acknowledgements to hang indefinitely when acks are enabled and the failure pipeline is used as a fallback. Additionally, sendToFailurePipeline silently drops any DlqObject whose event is null (no release called), which can also leak handles. } 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); Possible Issue In sendToFailurePipelineOrRelease, when failurePipeline is null the code logs an error and releases with false, but there is no corresponding release in the success path after failurePipeline.sendEvents(records). Previously, sending to the failure pipeline implicitly transferred ownership of event handles via the records. Confirm that sendEvents is responsible for releasing handles for all passed events — otherwise, events included in the pipeline send will have their handles leaked, and events excluded (null event) are definitely leaked. 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); } private void sendToFailurePipelineOrRelease(final HeadlessPipeline failurePipeline, final List<DlqObject> dlqObjects, final Throwable failure) { if (failurePipeline != null) { sendToFailurePipeline(failurePipeline, dlqObjects); } else { dlqObjects.forEach(dlqObject -> { LOG.error(NOISY, "Failed to write a document to the DLQ", failure); dlqObject.releaseEventHandle(false); }); } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 4a99c9d
Explore these optional code suggestions:

Category	Suggestion	Impact
General	Pass the actual IOException to fallback handler When dlqWriter.write throws IOException, event handles may not have been released yet, but if the fallback sendToFailurePipeline succeeds, event handles are forwarded without ever being released by this sink. Also, if failurePipeline is null, the fallback path logs the original failure (which may be null) rather than the IOException that triggered the fallback. Consider logging e in the fallback-release path and ensuring handle lifecycle is consistent. data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java [534-537] } catch (final IOException e) { LOG.error(NOISY, "Failed to write to S3 DLQ, falling back to failure pipeline", e); - sendToFailurePipelineOrRelease(failurePipeline, dlqObjects, failure); + sendToFailurePipelineOrRelease(failurePipeline, dlqObjects, e); } Suggestion importance[1-10]: 6 __ Why: Valid point - the fallback path uses the original failure (which may be null) instead of the IOException that caused the fallback, making error logging less useful when failurePipeline is also null.	Low
Possible issue	Release handles for events skipped in forwarding DlqObjects whose event is null are silently dropped without releasing their event handles, which can leak acknowledgements and starve upstream back-pressure. Release the handle (with success=false) for any skipped dlqObject, and consider releasing handles for forwarded events as well since ownership is transferring to the failure pipeline. 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: Correctly identifies that DlqObjects with null events are silently dropped without releasing event handles, which could cause acknowledgement leaks. This matches the pre-existing behavior from the removed code but is still a valid concern.	Low

---

Previous suggestions

Suggestions up to commit 932222c

Category	Suggestion	Impact
Possible issue	Release event handles on DLQ fallback path When falling back from a failed S3 DLQ write to the failure pipeline, the event handles are never released. sendToFailurePipeline sends records downstream but does not call releaseEventHandle, which can lead to acknowledgment leaks/hangs. Ensure handles are released (e.g., with success=true) after successful fallback dispatch, or release with false if no failure pipeline exists (already handled). Confirm the fallback path releases handles appropriately. data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java [530-537] } 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); + dlqObjects.forEach(dlqObject -> dlqObject.releaseEventHandle(true)); } Suggestion importance[1-10]: 6 __ Why: The suggestion identifies a potentially valid concern about event handle acknowledgment leaks in the fallback path, since sendToFailurePipeline does not release handles. However, releasing with true unconditionally after dispatching may not be the correct semantics (acknowledgment should typically be tied to downstream processing), so the fix is partially speculative.	Low

[dlvenable]

Thanks @graytaylor0 . I think the order you outlined is good. We should have tests to assert this.

[github-actions]

Persistent review updated to latest commit 4a99c9d

[dlvenable]

Thanks @graytaylor0 !