Configuration - FlowWarden

flowwarden-amqp is auto-configured via AmqpDlqAutoConfiguration. It activates when both spring-amqp is on the classpath and a RabbitTemplate bean is available — which is the default whenever you include spring-boot-starter-amqp.

Properties

All properties are bound to AmqpDlqProperties under the prefix flowwarden.amqp.

Property	Type	Default	Description
`flowwarden.amqp.exchange`	`String`	`"flowwarden.dlq"`	Default topic exchange to which failed events are published. Per-stream override via `@AmqpDlqOptions(exchange = "...")`.
`flowwarden.amqp.routing-key-prefix`	`String`	`"dlq."`	Prefix for the routing key — full key = `prefix + streamName` (for example, `"dlq.orders-stream"`). Per-stream full override via `@AmqpDlqOptions(routingKey = "...")`.
`flowwarden.amqp.confirm-mode`	`boolean`	`true`	Block `save(...)` until the broker ACKs the publish. Requires `spring.rabbitmq.publisher-confirm-type=correlated` (or `simple`); see Confirm-mode trade-offs.

application.yml

flowwarden:
  amqp:
    exchange: "alerts.dlq"
    routing-key-prefix: "dlq."
    confirm-mode: true

The AMQP connection itself (host, port, credentials, virtual host, SSL) is configured through the standard Spring Boot spring.rabbitmq.* properties. flowwarden-amqp consumes whichever RabbitTemplate Spring Boot exposes — pointing it at a single broker, a cluster, or a managed AMQP service (CloudAMQP, AWS MQ for RabbitMQ) is transparent.

What gets wired

The auto-configuration registers three beans:

Bean	Class	Purpose
`amqpDlqMessageBuilder`	`AmqpDlqMessageBuilder`	Jackson serializer for `FailedEvent` (with manual BSON handling for `documentKey`, `resumeToken`, `fullDocument`).
`amqpDlqStore`	`AmqpDlqStore`	The active `DlqStore` implementation. Publishes via `RabbitTemplate`.
`amqpDlqOptionsRegistrar`	`AmqpDlqOptionsRegistrar`	`BeanPostProcessor` that scans `@ChangeStream` beans for `@AmqpDlqOptions` and wires per-stream overrides on the store at `ApplicationReadyEvent`.

Conditions applied to the auto-configuration class:

@AutoConfiguration(after = RabbitAutoConfiguration.class) — runs after Spring Boot’s RabbitMQ auto-config so the RabbitTemplate bean is already available.
@ConditionalOnClass(RabbitTemplate.class) — skip entirely if spring-amqp is not on the classpath.
@ConditionalOnBean(RabbitTemplate.class) — skip if no RabbitTemplate exists in the context (for example, when the AMQP starter is present but no connection is configured).

Each @Bean is also @ConditionalOnMissingBean, so a user-declared DlqStore, AmqpDlqMessageBuilder, or AmqpDlqOptionsRegistrar wins without further configuration.

Message format on the wire

Body — `application/json`, UTF-8

FailedEvent is serialized to JSON with manual handling of MongoDB BSON types (no Jackson BSON module — kept off the dependency footprint). The body shape:

{
  "id": "c2e8a5b6-…",
  "streamName": "orders",
  "operationType": "INSERT",
  "documentKey": { "_id": { "$oid": "65f…" } },
  "fullDocument": { "_id": { "$oid": "65f…" }, "status": "PAID", … },
  "resumeToken": { "_data": "82…" },
  "error": {
    "type": "java.lang.IllegalStateException",
    "message": "downstream call failed",
    "stackTrace": "java.lang.IllegalStateException: …"
  },
  "attempts": 3,
  "status": "DLQ",
  "firstAttemptAt": "2026-06-09T14:32:11.123Z",
  "lastAttemptAt":  "2026-06-09T14:32:13.456Z",
  "createdAt":      "2026-06-09T14:32:13.500Z",
  "expiresAt":      "2026-07-09T14:32:13.500Z",
  "metadata": { "tenantId": "acme" }
}

documentKey, fullDocument, resumeToken use MongoDB extended JSON (via BsonDocument.toJson() / Document.toJson()). Null fields are omitted.
Timestamps are ISO-8601 strings (Jackson JavaTimeModule with WRITE_DATES_AS_TIMESTAMPS=false).
error.stackTrace, expiresAt, metadata are emitted only when non-null/non-empty.

Headers

Every publish carries the following AMQP headers in addition to the standard messageId, contentType=application/json, and contentEncoding=UTF-8:

Header	Source	Example
`flowwarden-event-id`	`FailedEvent.id()` (UUID)	`c2e8a5b6-…`
`flowwarden-stream-name`	`FailedEvent.streamName()`	`orders-stream`
`flowwarden-operation-type`	`FailedEvent.operationType()`	`INSERT`
`flowwarden-attempts`	`FailedEvent.attempts()`	`3`
`flowwarden-status`	`FailedEvent.status()`	`DLQ`
`flowwarden-first-attempt-at`	epoch ms (omitted if null)	`1717880400000`
`flowwarden-schema-version`	constant `AmqpDlqMessageBuilder.SCHEMA_VERSION`	`1`
`expiration` (AMQP-native)	`DlqPolicy.retentionDays()` × ms (omitted when ≤ 0)	`2592000000` (30 d)

The flowwarden-schema-version header is bumped on every breaking change to FailedEvent so consumers can route by version if needed.

Per-stream overrides — `@AmqpDlqOptions`

Decorate a @ChangeStream + @DeadLetterQueue bean with @AmqpDlqOptions to override the global defaults for that one stream:

@ChangeStream(name = "payments-stream", collection = "payments")
@DeadLetterQueue(retentionDays = 90)
@AmqpDlqOptions(
    exchange = "alerts.critical",
    routingKey = "alert.payment.failed",
    mandatory = true)
public class PaymentsHandler { ... }

Attribute	Default	Behavior
`exchange`	`""` (empty)	Falls back to `flowwarden.amqp.exchange`.
`routingKey`	`""` (empty)	Falls back to `flowwarden.amqp.routing-key-prefix + streamName`.
`mandatory`	`false`	When `true`, the broker returns the message to the publisher if no queue is bound to the routing key. Returned messages are logged at `WARN`; wire a `RabbitTemplate.ReturnsCallback` for custom handling.

Precedence is annotation > properties > built-in defaults. Per-stream overrides are registered by the AmqpDlqOptionsRegistrar BPP at ApplicationReadyEvent time, so they’re effective from the very first failed publish.

Confirm-mode trade-offs

confirm-mode: true (default)
confirm-mode: false

AmqpDlqStore.save(...) blocks until the broker confirms the publish (publisher confirms over an invoke(...) channel), with a 10-second timeout. On NACK, timeout, or broker crash before persist, an AmqpException propagates — the core then emits onEventDlqFailed(streamName, cause).Requires spring.rabbitmq.publisher-confirm-type set to correlated or simple. If the property is missing, set to none, or set to any other value, the autoconfig logs a WARN at startup and silently falls back to fire-and-forget rather than failing the application. Set the Spring property explicitly to honor confirm-mode=true.Choose when: data loss on broker failure is unacceptable, and you can afford a per-publish round-trip (≤ 10 s ceiling).

AmqpDlqStore.save(...) returns immediately after handing the message to RabbitTemplate.send(...). No broker ACK is awaited.The core’s onEventSentToDlq(...) signal can be a false positive: a broker crash between the network write and the disk persist won’t be visible to the publisher.Choose when: throughput dominates, and either the loss is acceptable (volatile alerts, non-critical streams) or an upstream layer (the source database, a write-ahead log, the operator) can replay the missing events.

Mixing backends

AmqpDlqStore and the core’s MongoDlqStore are not auto-configured side by side — the @ConditionalOnMissingBean(DlqStore.class) on the AMQP bean means the AMQP store wins as soon as flowwarden-amqp is on the classpath. To get both (Mongo for inspectable in-app replay, AMQP for downstream fanout), declare a small composite DlqStore bean that delegates to both:

HybridDlqConfig.java

@Configuration
public class HybridDlqConfig {

    @Bean
    DlqStore dlqStore(MongoTemplate mongo,
                      RabbitTemplate rabbit,
                      AmqpDlqProperties amqpProperties,
                      AmqpDlqMessageBuilder messageBuilder) {

        MongoDlqStore mongoStore = new MongoDlqStore(mongo);
        AmqpDlqStore  amqpStore  = new AmqpDlqStore(rabbit, amqpProperties, messageBuilder, true);

        return new DlqStore() {
            @Override public void save(FailedEvent event, DlqPolicy policy) {
                mongoStore.save(event, policy);   // persist for in-app inspection
                amqpStore.save(event, policy);    // fan out to downstream consumers
            }
            @Override public Optional<FailedEvent> findById(String id) {
                return mongoStore.findById(id);
            }
            @Override public List<FailedEvent> findByStreamName(String streamName) {
                return mongoStore.findByStreamName(streamName);
            }
        };
    }
}

User-declared beans win over both auto-configs (@ConditionalOnMissingBean(DlqStore.class)), so no other configuration is needed.

Disabling

There is no dedicated enabled flag. To stop publishing failed events to AMQP, take one of the following actions:

Drop the dependency — remove flowwarden-amqp from the build. The core’s MongoDlqStore re-activates automatically the next time the application starts.
Override the bean — declare a @Bean MongoDlqStore (or any other DlqStore) in your configuration; the @ConditionalOnMissingBean on amqpDlqStore defers to it.
Skip the auto-config — spring.autoconfigure.exclude=io.flowwarden.amqp.autoconfigure.AmqpDlqAutoConfiguration keeps the library on the classpath but skips the wiring entirely.

DlqStore SPI

The contract AmqpDlqStore implements.

@DeadLetterQueue

The user-facing annotation behind every DLQ publish.

Retry & DLQ guide

Lifecycle of a failed event from retry budget exhaustion to DLQ.

Quickstart

Add flowwarden-amqp to your project in 5 minutes.

​Properties

​What gets wired

​Message format on the wire

​Body — application/json, UTF-8

​Headers

​Per-stream overrides — @AmqpDlqOptions

​Confirm-mode trade-offs

​Mixing backends

​Disabling

​See Also

DlqStore SPI

@DeadLetterQueue

Retry & DLQ guide

Quickstart

Properties

What gets wired

Message format on the wire

Body — `application/json`, UTF-8

Headers

Per-stream overrides — `@AmqpDlqOptions`

Confirm-mode trade-offs

Mixing backends

Disabling

See Also