Quickstart

This quickstart adds the FlowWarden AMQP DLQ backend to an existing Spring Boot application that already uses flowwarden-stream-core. After these three steps, every event that exhausts its retry budget is published to a topic exchange instead of the default MongoDB _fw_dlq collection.

1. Add the dependency

<dependency>
    <groupId>io.flowwarden</groupId>
    <artifactId>flowwarden-amqp</artifactId>
    <version>1.0.0-rc.1</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

implementation 'io.flowwarden:flowwarden-amqp:1.0.0-rc.1'
implementation 'org.springframework.boot:spring-boot-starter-amqp'

flowwarden-amqp requires spring-boot-starter-amqp on the classpath. Spring Boot creates the RabbitTemplate bean that flowwarden-amqp consumes, and resolves the underlying connection factory.

2. Configure the broker

Use the standard Spring Boot properties for the AMQP connection:

application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated   # required for confirm-mode (the default)

Optionally, override the FlowWarden-specific routing defaults:

application.yml

flowwarden:
  amqp:
    exchange: "alerts.dlq"        # default: "flowwarden.dlq"
    routing-key-prefix: "dlq."    # default: "dlq."  (full key = prefix + streamName)

See Configuration for the full property reference.

flowwarden.amqp.confirm-mode defaults to true — AmqpDlqStore.save(...) blocks until the broker acknowledges the publish. This requires spring.rabbitmq.publisher-confirm-type to be correlated (or simple). If it isn’t, the autoconfig logs a WARN at startup and silently falls back to fire-and-forget — set the property explicitly to honor confirm-mode.

3. Use `@DeadLetterQueue` as usual

Your @ChangeStream / @DeadLetterQueue annotations don’t change. The AMQP-backed bean is picked up automatically when the application context starts:

OrderStream.java

@ChangeStream(collection = "orders")
@DeadLetterQueue(retentionDays = 30)
public class OrderStream {

    @OnInsert
    void onInsert(Order order, ChangeStreamContext<Order> ctx) {
        // your business logic — any exception that exhausts the retry budget
        // is published to the AMQP exchange with routing key "dlq.orders"
    }
}

Per-stream overrides

Use @AmqpDlqOptions to override the default exchange or routing key on a specific stream — useful when one criticality tier should fan out to a different queue:

PaymentStream.java

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

With mandatory = true, the broker returns the message to the publisher if no queue is bound to the routing key (instead of silently dropping it) — useful to detect misconfigured consumers. See Configuration → Per-stream overrides.

Verify

Start the application and confirm the wiring. On boot, look for the auto-config log line:

flowwarden-amqp: DlqStore active on exchange='flowwarden.dlq', routingKeyPrefix='dlq.', confirmMode=true

Then force a handler to throw and check that a message lands on the exchange. With the RabbitMQ management plugin:

# bind a temporary queue to the DLQ exchange and consume one message
rabbitmqadmin declare queue name=dlq.tap durable=false
rabbitmqadmin declare binding source=flowwarden.dlq destination=dlq.tap routing_key='dlq.#'
rabbitmqadmin get queue=dlq.tap requeue=false

The payload is a JSON FailedEvent document. Expected AMQP headers:

Header	Example
`flowwarden-event-id`	`c2e8a5b6-…`
`flowwarden-stream-name`	`orders`
`flowwarden-operation-type`	`INSERT`
`flowwarden-attempts`	`3`
`flowwarden-status`	`DLQ`
`flowwarden-schema-version`	`1`

AmqpDlqStore is publish-only. DlqStore.findById(...) and DlqStore.findByStreamName(...) return empty values and log a one-shot WARN pointing operators to the AMQP consumer side — failed events are inspected on the broker, not via the FlowWarden API.

Configuration

Properties, wire format, per-stream overrides, confirm-mode trade-offs.

@DeadLetterQueue reference

The annotation that drives the DlqStore SPI this backend implements.

Retry & DLQ guide

How retries and DLQ delivery fit together in the core.

DlqStore SPI

The contract behind every DLQ backend.

Getting Started

Reference

1. Add the dependency

2. Configure the broker

3. Use `@DeadLetterQueue` as usual

Per-stream overrides

Verify

See Also

Configuration

@DeadLetterQueue reference

Retry & DLQ guide

DlqStore SPI

​1. Add the dependency

​2. Configure the broker

​3. Use @DeadLetterQueue as usual

​Per-stream overrides

​Verify

​See Also

Configuration

@DeadLetterQueue reference

Retry & DLQ guide

DlqStore SPI

1. Add the dependency

2. Configure the broker

3. Use `@DeadLetterQueue` as usual

Per-stream overrides

Verify

See Also