> ## Documentation Index > Fetch the complete documentation index at: https://docs.flowwarden.io/llms.txt > Use this file to discover all available pages before exploring further. # DlqStore SPI > Failed-event persistence backing @DeadLetterQueue — contract, FailedEvent record, and how to plug in your own backend `DlqStore` is the SPI behind the [`@DeadLetterQueue`](/reference/dead-letter-queue) 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 ```java theme={null} 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 findById(String id); /** Retrieves all failed events for the given stream. */ List 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 ```java theme={null} 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 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`: ```java theme={null} 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_` name). A MongoDB TTL index on `expiresAt` is created at startup so `retentionDays` takes effect without manual setup. | The Redis backend ([`flowwarden-redis`](/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: ```java theme={null} 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, …): ```java theme={null} @Configuration public class KafkaDlqConfig { @Bean DlqStore dlqStore(KafkaTemplate 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`: ```xml theme={null} io.flowwarden flowwarden-stream-core-testkit test ``` 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 The user-facing annotation that drives this SPI. DLQ routing, manual sends, document schema, best practices. The sister SPI that backs `@Checkpoint`. The sister SPI that backs `DeploymentMode.SINGLE_LEADER`.