opensearch-project / opensearch-project/data-prepper

Pipeline DLQ

Open
#3,857 5 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

enhancement Roadmap:Stability/Availability/Resiliency
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)

  1. 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

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.