Add support for pipeline DLQ to opensearch sink

[kkondaka]

Description

Add support for pipeline DLQ to opensearch sink. When "dlq_pipeline" is present in the configuration, opensearch sink uses that for DLQ and ignores the sink level dlq config, if present.

Issues Resolved

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

Check List

• [X ] New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• [X ] 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.

[graytaylor0]

For pipeline DLQ I don't think we can release the event handles right? Just because we send to pipeline DLQ doesn't mean data made it end to end. DLQ pipeline has to be responsible to acknowledgments

[kkondaka]

Of course. I am not releasing events here. This API is called only eventHandles are used. I added "event" related logic for completion sake. May be I should assert that release with event is never called?

[dlvenable]

There is a big item here that we should consider. You are putting a DlqObject into the failure pipeline. But, this is not the original Event. So this means that we will have different data structures in failure pipelines for each sink and processor.

I tend to think that the best approach would be to include the original Event in the failure pipeline. Additionally, we should include failure information as metadata in that Event.

This also opens the question of whether we want to save DlqObject into the final sinks in the failure pipeline.

[kkondaka]

I am not putting DLQobject into failure pipeline. It is the original event.

[dlvenable]

@kkondaka and I discussed this.

First, this PR already is sending the original Event.

Second, we discussed how to include the failure information. We could use metadata, but that would require that every failure pipeline copy metadata into a field. Instead, Data Prepper will add the failure information into a field in the event itself. By default we will name it _failure_data. However, we could also allow this to be overridden in the original pipeline by way of a failure_key key.

For example:

my-regular-pipeline:
  failure_key: pipeline_failure_data

When the DLQ pipeline receives this, it will have pipeline_failure_data instead.

[dlvenable]

I'd like to make this easier for plugin authors. This will quickly create tech debt if we continue this pattern.

Maybe a method on Event?

event.setFailureMetadata(failureMetadata);

Or a factory somehow?

[dlvenable]

Something with a builder-type pattern could also be nice:

event.updateFailureMetadata()
  .with("pluginId", dlqObject.getPluginId())
  .with("pluginName")
  ...
  .with("status", ...)
  .with("index", ...)
  .with("indexId", ((FailedDlqData) dlqObject.getFailedData()).getIndexId());

[dlvenable]

Typo: pluginaName

[dlvenable]

Something with a builder-type pattern could also be nice:

event.updateFailureMetadata()
  .with("pluginId", dlqObject.getPluginId())
  .with("pluginName")
  ...
  .with("status", ...)
  .with("index", ...)
  .with("indexId", ((FailedDlqData) dlqObject.getFailedData()).getIndexId());

[dlvenable]

assertThat(failureMetadata, instanceOf(DefaultEventFailureMetadata.class));

[dlvenable]

This could be an inner class of JacksonEvent. This would encapsulate it and simplify the code:

private class DefaultEventFailureMetadata {
        private static final String FAILURE_METADATA = "_failure_metadata";

        public DefaultEventFailureMetadata with(String key, Object value) {
            put(FAILURE_METADATA+"/"+key, value);
            return this;
        }
    }

    public EventFailureMetadata updateFailureMetadata() {
        return new DefaultEventFailureMetadata();
    }

[dlvenable]

This should not be public. Make it private, or package protected if needed for testing.

[KarstenSchnitter]

Thanks for providing this change. We are really looking forward to this feature.

We discussed internally using the DLQ-Pipeline to try a rectify possible field conflicts during bulk indexing. This is possible when the index failure response is available in the DLQ event. The idea is to rename or drop fields that cause conflicts and then try indexing them again. This probably needs a processor plugin.

Do I understand the PR correctly, that the bulk index response for the event is available in th DLQ event?

[kkondaka]

@KarstenSchnitter the response code and response message will be available. Is there anything else from bulk response are you looking for?

[KarstenSchnitter]

@kkondaka we are looking for the error part of the bulk response. For a single item this would look like this:

{
  "errors": true,
  "items": [
    {
      "index": {
        "_index": "logs-2025.09.02",
        "_id": "abc123",
        "status": 400,
        "error": {
          "type": "mapper_parsing_exception",
          "reason": "failed to parse field [a.b.c] of type [keyword] in document with id 'abc123'"
        }
      }
    }
  ]
}

We need the error object, i.e. type and reason.

[dlvenable]

Nice, Thanks @kkondaka !

[dlvenable]

@kkondaka we are looking for the error part of the bulk response. For a single item this would look like this:

{
  "errors": true,
  "items": [
    {
      "index": {
        "_index": "logs-2025.09.02",
        "_id": "abc123",
        "status": 400,
        "error": {
          "type": "mapper_parsing_exception",
          "reason": "failed to parse field [a.b.c] of type [keyword] in document with id 'abc123'"
        }
      }
    }
  ]
}

We need the error object, i.e. type and reason.

We should include this as well. Is the message the same thing? Is the type lost somehow?

[graytaylor0]

Not making this configurable at the moment?

[kkondaka]

What change @graytaylor0 ?

[kkondaka]

@dlvenable I will investigate to see what information is getting lost in case of errors. I made sure whatever we are currently sending to DLQ is still carried in the events forwarded to pipeline DLQ. But I am not sure if we are currently missing some.