dlq

[untitaker]

Configuration Schema

pipeline:
  dlq:
    topic: "default-dlq"           # Default DLQ topic
    bootstrap_servers: ["127.0.0.1:9092"]  # Optional, defaults to source
  segments:
    - steps_config:
        my_source:
          bootstrap_servers: ["127.0.0.1:9092"]
        my_parser:
          dlq:
            enabled: true          # Required opt-in
        my_map_step:
          dlq:
            enabled: true
            topic: "custom-dlq"    # Optional: override default
        my_reducer:
          # No dlq = crash on error (current behavior)

DLQ Message Format

{
  "original_topic": "source-topic",
  "original_partition": 5,
  "original_offset": 12345,
  "original_key": "user-123",
  "step_name": "my_parser",
  "consumer_group": "pipeline-my_source",
  "error": "Schema validation failed",
  "timestamp": 1704067200.123
}

Implementation

The implementation will change drastically, still. Right now the exceptions from userland are mostly caught in python and directly forwarded to a dlq producer, then reraised to skip the message. The producer should move to Rust to avoid it becoming a bottleneck.

[untitaker]

i think this test was wrong previuosly, it tested the same thing as the test before it.

[fpacifici]

Some high level questions.

[fpacifici]

What about just producing the payload we got at the specific step directly on the DLQ as well ? Metadata could be either a message wrapper or headers.

[fpacifici]

High level design questions.

• Would we have a DLQ for aggregate steps and batching steps ?
• In case we did not support aggregate steps (that are lossy), this may be ok, though, what about invalid messages for the following steps ?
• For batch scenarios, we should at least identify specific invalid messages and DLQ them as they still retain the original identity. If we did not support this, the DLQ would not be useful for SBC either as we batch first and process everything in multiprocess in batches.

I think the semantics of aggergate steps could be this:

• An invalid message is DLQed when we try to add it to an aggregate and fail. This does not compromise the existing aggregate
• DLQ in batched message should still work as if the messages were not batched. Batching is generally an optimization, it should not change the semantics of the application, other aggregations do.
• We should supoprt DLQ even after a batch/aggregate step. At this point the original identity may be lost, so I think we may want to DLQ the aggregate payload entirely. This would be viable in the new architecture as we could reprocess data from a specific step of the pipeline as the steps have identity and they are statically connected.