opensearch-project / opensearch-project/data-prepper
Pipeline DLQ
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 374
- Forks
- 354
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
Is your feature request related to a problem? Please describe.
Provide a way to send all failed events to a global/pipeline-level DLQ. Failed events any where in the pipeline (sources, processors, and sinks) are sent directly to this DLQ. This will eventually replace sink level DLQs we have today.
Describe the solution you'd like
Preferred solution (based on @dlvenable's initial thoughts and a discussion meeting)
- Option to define failure pipeline in the YAML file like
my-failure-pipeline:
type: failure
sink:
- s3:
bucket: "..."
codec:
ndjson:
And each sub-pipeline in the yaml may have an entry pointing to this as follows
sample-pipeline:
failure-pipeline: my-failure-pipeline
source:
...
processor:
...
sink:
...
In addition, there may be an option to have a default pipeline which is used if no failure pipeline is mentioned in a sub-pipeline
default-failure-pipeline:
type: failure
sink:
- s3:
bucket: "..."
codec:
ndjson:
And finally an implicit failure pipeline which is created without any entries in the YAML file. The implicit failure pipeline will send all failed events to stdout
This requires changes to code in many places and so it is better to introduce a new API (For example, executeWithFailures() API in processors which will return both output records and failed records). Data Prepper core code can then take the failed records and send them to appropriate failure pipeline (configured failure pipeline or default failure pipeline or implicit failure pipeline). Similarly new API at source and sink level maybe added. Once the API is added, code may be modified slowly so that all sources/sinks/processors use this new API.
Having a separate pipeline for failure, allows the same pipeline to be used by multiple pipelines. And also makes it possible to write sub-pipelines under it and do conditional routing etc.
Describe alternatives you've considered (Optional)
Instead of new API in processors/sources/sinks, we could have a global singleton for DLQEvents managed by DLQEventsManager which each source uses in its constructor and failed events are handed over to this DLQEventsManager which will route the events to failure pipeline. I think this approach is also ok.
Additional context
Add any other context or screenshots about the feature request here.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by tracing the YAML pipeline definitions and the source, processor, and sink APIs described in the issue. Compare the proposed executeWithFailures() approach with the alternative DLQEventsManager design. Done means failed events from each pipeline stage can reach a configured, default, or implicit failure pipeline, with shared failure pipelines supported.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100