DlqStore SPI

DlqStore is the SPI behind the @DeadLetterQueue annotation. It persists events that exhausted their retry budget so an operator (or an automated replay job) can investigate, fix, and re-process them. The MongoDB-backed implementation is auto-configured; any custom backend (Kafka topic, RabbitMQ queue, JDBC table, S3 bucket, …) plugs in by registering a DlqStore Spring bean.

When the SPI is exercised

Path	Calls into the SPI
Handler exhausts its retry budget	Framework invokes `save(failedEvent, policy)` automatically.
Manual `ctx.sendToDlq(reason)` from a handler	Same — `save(failedEvent, policy)`. Re-throws on failure so the user code can react.
Operator inspects the DLQ	Custom callers (admin tooling, the FlowWarden Console) use `findById` and `findByStreamName`.

Interface

public interface DlqStore {

    /**
     * Persists a failed event with its cross-cutting DLQ policy.
     * The caller has already applied the include-flags to the FailedEvent
     * payload, so impls typically only consult policy.retentionDays() to
     * set up backend-native expiry (TTL index, message TTL, etc.).
     */
    void save(FailedEvent event, DlqPolicy policy);

    /** Retrieves a failed event by its unique identifier (UUID). */
    Optional<FailedEvent> findById(String id);

    /** Retrieves all failed events for the given stream. */
    List<FailedEvent> findByStreamName(String streamName);

    /** Always-no-op implementation. */
    static DlqStore noOp();
}

Semantic contract

save must store the entry under event.id() and, if the backend supports it, set up native expiry from policy.retentionDays() (e.g. MongoDB TTL index, Kafka message TTL). When retentionDays == 0, entries are kept indefinitely.
The caller applies policy.includeOriginalDocument() and policy.includeStackTrace() to the FailedEvent payload before calling save. Implementations don’t re-check those flags — they just persist what they’re given.
findById returns Optional.empty() for missing or already-expired entries.
findByStreamName returns an empty list (never null) when no entry matches. Ordering is implementation-defined.

`FailedEvent` record

public record FailedEvent(
        String              id,                // UUID
        String              streamName,
        String              operationType,     // MongoDB op: INSERT, UPDATE, DELETE, ...
        BsonValue           documentKey,       // _id of the source document
        Document            fullDocument,      // nullable — null if includeOriginalDocument=false
        BsonDocument        resumeToken,
        ErrorInfo           error,
        int                 attempts,
        String              status,            // STATUS_PENDING for new entries
        Instant             firstAttemptAt,
        Instant             lastAttemptAt,
        Instant             createdAt,
        Instant             expiresAt,         // nullable — null when retentionDays=0
        Map<String, Object> metadata
) {

    public static final String STATUS_PENDING = "PENDING";

    public record ErrorInfo(
            String type,         // exception class name
            String message,
            String stackTrace    // nullable — null if includeStackTrace=false
    ) {}
}

expiresAt is precomputed by the framework from createdAt + policy.retentionDays() (see DlqPolicy.computeExpiresAt). Backends with native TTL use this directly; backends without it can ignore the field and rely on a custom sweep.

`DlqPolicy` record

The backend-neutral view of @DeadLetterQueue:

public record DlqPolicy(
        int     retentionDays,            // 0 = permanent
        boolean includeOriginalDocument,
        boolean includeStackTrace
) {
    public static DlqPolicy fromAnnotation(DeadLetterQueue ann) { /* ... */ }
    public Instant computeExpiresAt(Instant createdAt) { /* ... */ }
}

The annotation also accepts backend-specific options (e.g. @MongoDlqOptions(collection = "...")). Those live outside DlqPolicy — the implementation reads its own companion annotation directly from the @ChangeStream definition.

Provided implementations

Implementation	When active	Storage
`MongoDlqStore`	Default in both imperative and reactive modes	One collection per `@ChangeStream` (configurable via `@MongoDlqOptions(collection = "...")`, defaulting to a per-stream `_fw_dlq_<streamName>` name). A MongoDB TTL index on `expiresAt` is created at startup so `retentionDays` takes effect without manual setup.

The Redis backend (flowwarden-redis) does not ship a DlqStore implementation — only LockService and CheckpointStore. DLQ persistence in a Redis-backed deployment falls back to the default MongoDlqStore (so MongoDB is still required), or to a user-provided custom bean.

No-op default

DlqStore.noOp() returns a shared singleton that silently discards everything:

DlqStore store = DlqStore.noOp();
store.save(event, policy);             // discarded
store.findById("any");                 // Optional.empty()
store.findByStreamName("any");         // empty list

Suitable for tests that don’t care about DLQ persistence. Not a production choice — failed events that should land in a DLQ vanish silently.

Providing a custom implementation

The auto-configured bean is guarded by @ConditionalOnMissingBean, so any @Bean DlqStore you declare wins. A custom implementation routes events by reading event.streamName() and resolving it to its backend resource (a Kafka topic, a Rabbit queue, a JDBC table, …):

@Configuration
public class KafkaDlqConfig {

    @Bean
    DlqStore dlqStore(KafkaTemplate<String, FailedEvent> kafka) {
        return new KafkaDlqStore(kafka);
    }
}

For a backend that mirrors MongoDlqStore’s pattern (per-stream resources resolved from a companion annotation), the framework exposes a registerStream(streamName, collection)-style hook at startup — see MongoDlqStore in the core module as the reference implementation, which uses this hook to create the TTL index.

Testing the contract

If you write a custom DlqStore, run it against the contract tests published in flowwarden-stream-core-testkit:

<dependency>
    <groupId>io.flowwarden</groupId>
    <artifactId>flowwarden-stream-core-testkit</artifactId>
    <scope>test</scope>
</dependency>

Extend DlqStoreContractTest and implement the abstract factory method. The contract covers save / findById / findByStreamName semantics and policy-driven expiry — the same suite MongoDlqStore is validated against.

@DeadLetterQueue annotation

The user-facing annotation that drives this SPI.

Retry & DLQ Guide

DLQ routing, manual sends, document schema, best practices.

CheckpointStore SPI

The sister SPI that backs @Checkpoint.

LockService SPI

The sister SPI that backs DeploymentMode.SINGLE_LEADER.

Annotations

Runtime

SPI

When the SPI is exercised

Interface

Semantic contract

`FailedEvent` record

`DlqPolicy` record

Provided implementations

No-op default

Providing a custom implementation

Testing the contract

See Also

@DeadLetterQueue annotation

Retry & DLQ Guide

CheckpointStore SPI

LockService SPI

​When the SPI is exercised

​Interface

​Semantic contract

​FailedEvent record

​DlqPolicy record

​Provided implementations

​No-op default

​Providing a custom implementation

​Testing the contract

​See Also

@DeadLetterQueue annotation

Retry & DLQ Guide

CheckpointStore SPI

LockService SPI

When the SPI is exercised

Interface

Semantic contract

`FailedEvent` record

`DlqPolicy` record

Provided implementations

No-op default

Providing a custom implementation

Testing the contract

See Also