Skip to main content
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

PathCalls into the SPI
Handler exhausts its retry budgetFramework invokes save(failedEvent, policy) automatically.
Manual ctx.sendToDlq(reason) from a handlerSame — save(failedEvent, policy). Re-throws on failure so the user code can react.
Operator inspects the DLQCustom 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

ImplementationWhen activeStorage
MongoDlqStoreDefault in both imperative and reactive modesOne 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.

​
See Also

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