Patterns

<dependency>
  <groupId>org.acemq</groupId>
  <artifactId>acemq-amqp-patterns</artifactId>
  <version>0.7.15</version>
</dependency>

Three things live here, and they are the three that every message-driven service rebuilds badly: an idempotency store, a transactional outbox, and a schema registry that survives a restart.

What is here, and what is not

The last three were advertised here for months before they existed. That was removed rather than excused, and they are now in the jar — built after three applications had needed them, which is why the shapes are what they are.

Three more live in the core rather than here, because they are wired into the consumer runtime: the retry ladder, request/reply, and pipelines — below.

Pipelines

A flow with named stages, where each stage is its own queue and its own consumer group:

try (Pipeline<Order> fulfilment = mq.pipeline("fulfilment", Order.class)
        .step("validate", Order.class, m -> validate(m.payload()))
        .step("reserve", Reservation.class, m -> stock.reserve(m.payload()))
                .describedAs("hold stock for 15 minutes so payment cannot oversell")
                .withRetry(RetryPolicy.exponential(4, Duration.ofSeconds(1), Duration.ofMinutes(1)))
                .concurrency(4)
        .step("dispatch", Void.class, m -> shipping.dispatch(m.payload()))
        .build()) {

    fulfilment.send(order);
}

Each step gets a queue named <pipeline>.<step>, so a stage that is slow shows up as a queue that is deep — you can see which stage, scale that one alone, and retry it on its own schedule. A single consumer doing all four in sequence gives you none of that.

The type flows through the builder: reserve receives what validate returned, checked at compile time. A step returning null stops the run early, which is a decision rather than a failure and is counted separately.

Naming and describing a step

.step("reserve", Reservation.class, handler)
        .describedAs("hold stock for 15 minutes so payment cannot oversell")

The name is wire format. It is the routing key, the queue suffix, and the entry in the routing slip that tracks where a message has got to — so it is restricted to letters, digits, dashes and dots, and renaming a step strands every message currently in flight against the old name.

The description is for people. Free text, no constraints, no effect on routing. It appears in the declaration log line and wherever a step is reported:

pipeline fulfilment declared with 3 steps: validate | reserve (hold stock for 15
minutes so payment cannot oversell) | dispatch

Keeping them separate is the point. A step name good enough to route on is rarely a sentence, and a sentence is never safe to route on — so improving how a stage reads should not be able to change where its messages go.

Two routing slips, and when each is right

Where a message is going travels with the message. There are two ways to say it and the library speaks both, reading either and writing whichever you ask for.

The declared route — x-acemq-route, and the default. A comma-separated list of step names with a position, resolved against the pipeline that declared them:

x-acemq-route: validate,reserve,dispatch
x-acemq-route-position: 1

The better shape when there is a declaration to resolve against. It costs one short header, it says where a message is without anybody decoding anything, and the route is checked at build time against steps whose types the compiler already threaded together. Nothing changes here; this is what pipeline.send(order) writes and what a pipeline reads when no itinerary is present.

The carried itinerary — acemq-routing-slip. The whole route as JSON, each stop naming its own exchange and routing key:

{"steps":[{"exchange":"orders-events","routingKey":"order.charge","name":"charge"}],
 "done":[{"exchange":"orders-events","routingKey":"order.validate","name":"validate",
          "completedAt":"2026-09-09T14:03:11Z"}]}

For when the route is decided per message — an order over a threshold visits an approver and one under it does not — and for when whoever consumes it has no declaration at all. This is the shape Go, Python and Ruby write, field for field, so a slip written by any of them is followed here and a slip written here is followed by any of them.

pipeline.send(Itinerary.empty()
        .then("orders-events", "order.validate", "validate")
        .then("orders-events", "order.charge", "charge"), order);

A step reading a carried itinerary publishes to the next stop on the slip, not to the next step in the declaration — that is the whole point, and it is what lets a Java consumer sit in the middle of a route that no Java service declared. The done list grows as the message travels, each stop stamped when it finished, so a run that stops halfway says how far it got.

Itinerary is immutable: then and advance return new instances. A slip that cannot be parsed is fatal rather than retryable — it will not parse next time either, and a message going round the broker while nothing can tell where it is meant to go is the worst of both.

Idempotency

At-least-once delivery means duplicates. An idempotency store is what stops a duplicate being handled twice: the second delivery of a message whose first delivery was confirmed is acknowledged without running the handler. That is the half of "exactly once" the broker cannot provide, and it is not the whole of it — the claim is a lease rather than a fact, so a handler that dies mid-work has its claim expire and the message is handled again, which is the behaviour that stops a crash losing the work. What this buys is effectively-once processing, and what it costs is that a handler still has to be safe to run twice.

IdempotencyStore seen = new JdbcIdempotencyStore(dataSource);

mq.consume("payments.new", Payment.class,
        ConsumerOptions.prefetch(10).idempotent(seen),
        message -> payments.charge(message.payload()));

Keyed on the message id, which AceMQ sets on publish and which survives the retry ladder unchanged.

Claim, then work, then confirm

store.tryClaim(id);   // before the handler: CLAIMED, ALREADY_CONFIRMED or IN_PROGRESS
store.confirm(id);    // after it returns
store.release(id);    // after it throws

What the consumer does with each answer:

A claim in progress is not a duplicate. Until the 0.7 line it was treated as one and acknowledged, which lost the message whenever the handler holding the claim had died and its release had failed too. Now the delivery waits a second and is republished to its own queue with the envelope unchanged: no attempt is spent, the retry policy never sees it, and it is never dead-lettered for it. Republished rather than requeued because a quorum queue counts every requeue against its delivery limit (twenty by default on RabbitMQ 4), which would drop the message long before a five-minute lease ran out. It comes round until the claim is confirmed (and it is acknowledged as a duplicate) or the lease expires (and it is taken over and handled). Telemetry records outcome="in_progress" for each trip.

claim still answers true/false. tryClaim is a default method built from claim and isConfirmed, so a store written against the older contract works unchanged. The in-memory store's claims are leases too now (claimTimeout, five minutes by default, never longer than the retention).

The claim is taken before the work, not after. Recording afterwards leaves a window — the process dies between the charge and the record — in which the work happened and nothing knows, so the retry does it again. Closing that window is the entire purpose, so the record has to come first.

The cost is the opposite failure: a process that dies after claiming and before working leaves a claim blocking a retry that should happen. Hence the lease — a claim not confirmed within claimTimeout expires and the message can be tried again. Set it comfortably above your slowest handler.

Any JDBC driver, including ones that report no SQLState. A duplicate claim is decided by looking for the row, not by reading an error code, so it behaves the same on PostgreSQL, H2, SQL Server and SQLite. Through the 0.7 line it read SQLState for the standard's integrity-constraint class, which sqlite-jdbc does not set, so on SQLite a duplicate threw instead of returning false and an ordinary duplicate was dead-lettered rather than skipped. A failure that is not a duplicate is still raised: a lock timeout must not be mistaken for "somebody already has this", or the caller acknowledges a message nothing handled.

Two replicas racing on the same id: one insert wins, the other takes a constraint violation and is told false. That is the whole of the distributed correctness, and it is correct because databases are good at exactly this.

purgeExpired() on a schedule. Retention should exceed your longest retry ladder by a comfortable margin.

The transactional outbox

The problem it solves is not duplicates. It is silence:

// Do not do this.
@Transactional
public void placeOrder(Order order) {
    orders.save(order);
    mq.publisher("orders", "order.placed").send(order);
}

The transaction covers the database and not the broker. When the save commits and the publish fails, the order exists and nobody was told — and nothing anywhere logs that as a problem, because both halves did what they were asked.

The fix is to write the message to the same database in the same transaction, so one commit decides both:

JdbcOutboxStore outbox = new JdbcOutboxStore(
        () -> currentTransactionConnection(),
        relayDataSource);

connection.setAutoCommit(false);
try {
    orders.save(connection, order);
    outbox.add(OutboxRecord.of("orders", "order.placed", envelope, json));
    connection.commit();
} catch (Exception e) {
    connection.rollback();
}

Both writes must be on the same connection. That is why the write side takes a ConnectionSupplier and not a DataSource: a record written on a connection from a pool is a record in a different transaction, which reproduces the exact bug this pattern exists to prevent. The relay gets its own DataSource because it runs outside your transaction and must not borrow it.

try (OutboxRelay relay = new OutboxRelay(mq, outbox)) {
    relay.start();
}

The relay polls, claims a batch under a lease, publishes, and marks each row published. A relay that dies mid-batch loses the lease and another takes it.

A row is marked published only once the broker has confirmed it and it reached a queue — even on a connection made withoutPublisherConfirms(), because the relay publishes on a confirmed channel of its own. A record nothing is bound to receive comes back unroutable, stays in the table and is tried again; after enough failed attempts the store stops handing it out and it waits there, visible, for someone to look at. It is never marked done and lost.

What it does and does not promise

It gives you at-least-once publishing, atomic with the database write. The message is never lost and is never sent for work that rolled back.

It does not give exactly-once publishing: the relay can publish and die before recording that it did, and the next relay publishes again. The duplicate is handled on the consuming side by the idempotency store. The pair is what people mean by "exactly once"; neither half is sufficient alone.

The relay publishes bytes unchanged

The record holds an already-serialised payload, and the relay republishes those bytes exactly. This is not an implementation detail — a relay that deserialised and re-encoded would double-encode the payload, which shipped once and is why there is a test for it.

The schema registry

Avro and Protobuf carry no description of themselves, so a reader must already hold the schema the writer used. What travels in the message is a small integer, and something must remember which integer means which schema.

JdbcSchemaRegistry registry = new JdbcSchemaRegistry(dataSource);
registry.createSchemaIfAbsent();

mq.publisher("events", "order.placed", OrderPlaced.class)
        .as(AvroCodec.registered(registry));

Identifiers must be stable forever. A registry that hands out fresh ones on restart makes every message written before the restart unreadable — and the bytes still parse, just as the wrong schema, which is the worst way to fail.

InMemorySchemaRegistry (in the Avro module) forgets on restart and belongs in tests only. Full detail is in serialization.

The claim check

Codec checked = ClaimCheckCodec.wrapping(new JsonCodec(), store);

A scanned medical report is tens of megabytes. Putting it on a queue is possible and is a mistake: it fills the broker's memory, it is copied to every bound queue, it makes a dead-letter queue impossible to inspect, and it turns a broker into a filesystem with worse tools. What travels instead is a claim check — the payload goes to a store, the message carries the key.

Only above a threshold. Below 64 KiB by default the payload travels inline, because offloading a small message turns one round trip into two — making the common case slower to fix the rare one. The framing says which it is, so a consumer reads both without being told, and the threshold can change without a flag day. Messages written before the codec was introduced are still readable.

ClaimCheckCodec.keyOf(body) reads which object a message needs without fetching it: the question in front of a dead-letter queue.

Retention is what goes wrong. The store and the queue have different lifetimes and nothing relates them. A message replayed a month later carries a key, and if the store expired it the replay produces a message nobody can read — worse than a lost message, because it looks like a message. Store retention must outlast every queue TTL, every dead-letter queue, and any manual replay.

Scheduling

scheduler.in(Duration.ofHours(4), "billing", "invoice.due", invoice);
scheduler.at(renewalDate, "policies", "policy.renew", policy);

Not a per-message time to live. The obvious implementation — set expiration, drop it in a queue nobody consumes, let it dead-letter — is wrong for anything but a single fixed delay, because a classic queue expires messages only at its head. A four-hour message followed by a one-minute message delivers the second one in four hours, and nothing reports it.

Instead: a ladder of queues each with a uniform TTL, and a message hops through them until due. Every message in a rung has the same delay, so the head is always the one due soonest. A four-hour delay is four one-hour hops.

The cost, stated plainly: a long delay is several round trips rather than one, and accuracy is about the smallest rung. A scheduler that must fire at 09:00:00.000 is a scheduler, not a message broker.

A message that is due but reaches no queue — the target exchange has nothing bound for its routing key yet — is not dead-lettered. It stays in the scheduler and is tried again every ten seconds until it is delivered, logging a warning each time, as the .NET and Ruby schedulers do. Earlier 0.7 releases sent it to acemq.schedule.due.dlq on the first attempt.

Saga

Saga<Order> booking = Saga.<Order>named("place-order")
        .step("take-payment", order -> payments.charge(order))
                .compensateWith(order -> payments.refund(order))
        .step("reserve-stock", order -> inventory.reserve(order))
                .compensateWith(order -> inventory.release(order))
        .step("book-courier", order -> couriers.book(order))
        .build();

SagaResult result = booking.run(order);

If book-courier throws, the stock is released and the payment refunded, in that order — reverse of the order the world was changed in.

Not a distributed transaction, and the difference matters. After take-payment the money really has moved and anybody looking sees it. The refund is a new fact, not an erasure, and for a few seconds the world contained a charge that should not have happened. A saga is honest about that where a two-phase commit pretends otherwise — so steps must be undoable by doing something else. Sending an email cannot be compensated; the apology is a second email. Put it last, after everything that can still fail.

Not durable. State is on the stack, so a crash midway leaves the saga half-applied with nothing to resume it. Where that matters the steps have to be messages and the state has to be in a database, which is a much larger thing.

The number to alert on

result.unresolved()   // steps whose compensation itself failed

When a compensation throws, it is logged and the remaining ones still run — stopping would leave more undone than continuing. What comes back is the list of effects that happened, were meant to be undone, and were not. No retry resolves those; a person does.

createSchemaIfAbsent() is for development

All three JDBC classes have it, and none of them should be running it in production. The table belongs in whatever migration tool already owns your schema; a library that creates tables at start-up is a library deciding when your database changes.

Related

• Reliability — retries and dead letters, the patterns that live in the core
• Serialization — the schema registry in context
• Tutorial 3 — all of this built up step by step