Add Dead Letter Queue support to the consumer pipeline

[nabisobhi]

Pull request overview

This PR adds dead letter queue (DLQ) support to the Dafda Kafka consumer pipeline so failed messages (after optional retries) can be forwarded to a Kafka dead-letter topic and the consumer can commit the offset and continue.

Why

Today a single "poison" message can take down or stall a consumer. When a message handler throws, the offset is never committed, so Dafda has only two outcomes:

• The app crashes — the exception escapes to ConsumerHostedService, and the default failure strategy calls StopApplication(). One bad message stops the whole service.
• The consumer loops forever — with RestartConsumer, the consumer restarts, re-reads from the same uncommitted offset, hits the same message, throws again… an infinite reprocessing loop that blocks every message behind it.

Neither is acceptable for a message that will never succeed (bad payload, unhandled edge case, downstream contract change). This PR adds a dead letter queue so poison messages are parked on a separate topic and the consumer moves on.

Changes

• Introduces IDeadLetterQueue with a Kafka implementation (KafkaDeadLetterQueue) and a no-op default (NullDeadLetterQueue).
• Extends the consumption flow to retry handler dispatch a configurable number of times, then publish the original message to the DLQ and commit so processing continues.
• Adds fluent configuration — WithDeadLetterQueue(...) and WithMaxRetries(...) — plus unit tests for topic resolution and DLQ behavior.

Usage

services.AddConsumer(options =>
{
    options.WithGroupId("order-processor");
    options.WithBootstrapServers("localhost:9092");
    options.RegisterMessageHandler<OrderPlaced, OrderPlacedHandler>("orders", "order-placed");

    // Enable DLQ with up to 10 retries before dead-lettering
    options.WithDeadLetterQueue("orders.dead-letter").WithMaxRetries(10);
});

The topic name is optional. When omitted, it's derived per message as "{topic}.{groupId}.dead-letter" (e.g. orders.order-processor.dead-letter). Including the consumer group scopes the DLQ to the group that actually failed — important because a message that's poison for one consumer may be perfectly valid for another consuming the same topic.

Behavior

GetNext → Dispatch ──success──────────────▶ Commit ─▶ next
              │
            throws
              │
      attempts < MaxRetries ? ──yes──▶ retry
              │ no
              ▼
   publish raw message + headers ─▶ dead-letter topic
              ▼
           Commit ─▶ next     (poison message parked, loop broken)

The dead-lettered record preserves the original key and raw payload, with diagnostics added as Kafka headers (source topic, exception type/message, timestamp).

Design notes

• Opt-in and fully backward compatible. Without WithDeadLetterQueue(...), the consumer uses NullDeadLetterQueue and exceptions propagate exactly as before. No existing option is affected.
• Complements WithConsumerErrorHandler. The DLQ absorbs per-message handler failures; genuinely catastrophic errors (connection loss, commit failures, and DLQ publish failures) still flow to the configured error handler.
• Cancellation-safe. Cancellation during dispatch is never retried or dead-lettered — it propagates for graceful shutdown.
• No resource leaks. KafkaDeadLetterQueue owns a producer and is disposed through the ConsumerHostedService → Consumer lifecycle on host shutdown.
• Config isolation. The DLQ producer is built from a filtered copy of the consumer settings (consumer-only keys removed), so it can't corrupt consumer configuration.

Tests

Added coverage for retry-then-dead-letter, retry-then-succeed (no DLQ), propagation when DLQ is disabled, cancellation, disposal, and topic-name resolution (explicit, derived with group id, and group-less fallback).

[DFDS-Snyk]

✅ Snyk checks have passed. No issues have been found so far.

Status	Scan Engine	Critical	High	Medium	Low	Total (0)
✅	Open Source Security	0	0	0	0	0 issues
✅	Licenses	0	0	0	0	0 issues
✅	Code Security	0	0	0	0	0 issues

💻 Catch issues earlier using the plugins for VS Code, JetBrains IDEs, Visual Studio, and Eclipse.

[Copilot]

Pull request overview

This PR adds dead letter queue (DLQ) support to the Dafda Kafka consumer pipeline so failed messages (after optional retries) can be forwarded to a Kafka dead-letter topic and the consumer can commit the offset and continue.

Changes:

• Introduces IDeadLetterQueue with implementations for Kafka (KafkaDeadLetterQueue) and a no-op default (NullDeadLetterQueue).
• Extends consumption flow to retry handler dispatch and publish failures to DLQ before committing.
• Adds fluent configuration (WithDeadLetterQueue, WithMaxRetries) and unit tests for topic resolution and DLQ behavior.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/Dafda/Consuming/NullDeadLetterQueue.cs	Adds a no-op DLQ implementation used when DLQ is not configured.
src/Dafda/Consuming/MessageResult.cs	Adds RawMessage to support forwarding the original Kafka payload to a DLQ.
src/Dafda/Consuming/KafkaDeadLetterQueue.cs	Implements publishing failed messages to a Kafka dead-letter topic with diagnostic headers.
src/Dafda/Consuming/KafkaConsumerScope.cs	Captures RawMessage and wires message metadata needed for DLQ publishing.
src/Dafda/Consuming/IDeadLetterQueue.cs	Defines the internal DLQ abstraction for failed message forwarding.
src/Dafda/Consuming/Consumer.cs	Adds retry + DLQ dispatch logic around handler execution.
src/Dafda/Configuration/DeadLetterQueueOptions.cs	Adds fluent DLQ configuration options including max retries validation.
src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs	Wires DLQ factory + max retries into the hosted consumer registration.
src/Dafda/Configuration/ConsumerOptions.cs	Exposes WithDeadLetterQueue(...) on the public consumer configuration surface.
src/Dafda/Configuration/ConsumerConfigurationBuilder.cs	Builds DLQ factory and propagates max retries into built configuration.
src/Dafda/Configuration/ConsumerConfiguration.cs	Stores DLQ factory and max retries in the built consumer configuration.
src/Dafda.Tests/Consuming/TestKafkaDeadLetterQueue.cs	Adds tests for DLQ topic name resolution behavior.
src/Dafda.Tests/Consuming/TestConsumer.cs	Adds tests for retry/DLQ behavior and cancellation behavior during dispatch.
src/Dafda.Tests/Builders/ConsumerBuilder.cs	Updates test builder to support injecting DLQ and max retries.

Suppressed comments (1)

src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs:86

• Same as above: the dead letter queue is created via the factory but never disposed (KafkaDeadLetterQueue implements IDisposable and owns an IProducer). Registering a shutdown callback here would avoid leaking the producer for the lifetime of the host.

                    configuration.EnableAutoCommit,
                    configuration.DeadLetterQueueFactory(provider),
                    configuration.MaxRetries
                ),

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 15 changed files in this pull request and generated 6 comments.

Suppressed comments (1)

src/Dafda.Tests/Consuming/TestConsumer.cs:11

• using TestDoubles; is inside the Dafda.Tests.Consuming namespace scope, so it will look for Dafda.Tests.Consuming.TestDoubles, which doesn’t exist (test doubles are under Dafda.Tests.TestDoubles). Use the fully-qualified namespace.

using TestDoubles;

[joseban-iii]

LGTM Nabi! ;)