> ## 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.
# Configuration
> Properties, wire format, per-stream overrides, confirm-mode trade-offs, hybrid setups
`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 = "...")`](#per-stream-overrides-amqpdlqoptions). |
| `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 = "...")`](#per-stream-overrides-amqpdlqoptions). |
| `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](#confirm-mode-trade-offs). |
```yaml application.yml theme={null}
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:
```json theme={null}
{
"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:
```java theme={null}
@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
`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:
```java HybridDlqConfig.java theme={null}
@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 findById(String id) {
return mongoStore.findById(id);
}
@Override public List 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.
## See Also
The contract `AmqpDlqStore` implements.
The user-facing annotation behind every DLQ publish.
Lifecycle of a failed event from retry budget exhaustion to DLQ.
Add `flowwarden-amqp` to your project in 5 minutes.