I wanted to finish the checkout example from my article on HTTP idempotency. That version handled a familiar problem: a customer retries a request after a timeout, and the service needs to recognize the same checkout. There is another failure to account for once checkout also sends an event to Kafka.
Imagine the order has committed in PostgreSQL. The next line sends OrderPlaced, which tells fulfillment to reserve the items. If the JVM stops between those two operations, the order survives while fulfillment keeps waiting for an event. On restart, the application needs a durable record of what it still owes Kafka.
That’s the change I want to make here. I’ll store the event beside the order, in the same database transaction, and publish it afterward. This is the transactional outbox pattern. It gives the application a place to resume after a crash, and it gives us a second problem to handle: publishing an event again when Kafka received the first attempt.
Give checkout a durable record of the event
I split the flow into three parts:
Accept the order. A PostgreSQL transaction inserts the order and an outbox row containing
OrderPlaced.Publish the event. A relay reads committed outbox rows, sends them to Kafka, and records successful publication.
Record fulfillment. The consumer identifies each event by its UUID and inserts one fulfillment row for that identifier.
The order and event share a transaction. Recording publication and processing fulfillment happen independently after the send.
The relay is the piece that connects the database to Kafka. If it stops before sending, the outbox row waits for the next attempt. If it stops after Kafka acknowledges the send, it can send the same event again. We therefore need both the durable event and a consumer that tolerates repeated delivery.
I model fulfillment as a row in PostgreSQL so we can follow the entire flow locally. A stock update could join that consumer transaction later. Calling an external warehouse service would introduce another delivery boundary and its own idempotency requirements.
Why I start with a polling relay
Debezium is a natural choice for publishing an outbox. Its PostgreSQL connector can capture inserts through logical decoding of the database’s write-ahead log, or WAL. That is change data capture, usually shortened to CDC. The outbox still holds the business event; Debezium provides the mechanism for discovering committed changes. The PostgreSQL connector documentation describes that path.
There are several ways to host it. A Kafka Connect deployment can run the connector independently of checkout. The Quarkus Debezium Engine extension embeds capture inside a Quarkus application. Its PostgreSQL extension, io.debezium.quarkus:debezium-quarkus-postgres, delivers captured events to methods annotated with @Capturing. An embedded relay can then arrange their publication to Kafka.
For this article, I chose a scheduler because I want to follow the entire send through a few Java methods. We can see which rows it selects, where it waits for Kafka, and exactly where a crash leaves publication uncertain. That makes the duplicate-delivery problem easier to explain. The cost is a query every second, a published_at update for each event, and coordination to add if we run several publishers.
I’ll keep that relay small and make its boundaries explicit. Toward the end, we’ll return to the Debezium option and see which parts of this application carry over.
Get the application running locally
I used Quarkus 3.39.4 and Java 25. Dev Services starts PostgreSQL 18.4 and Apache Kafka 4.2.0 through Podman, and Flyway creates the tables. To follow along, have Java 25, Git, curl, and Podman available. On macOS or Windows, Podman also needs a running machine. The project includes a Maven Wrapper.
Start from the complete application in my Github repository:
git clone https://github.com/myfear/the-main-thread.git
cd the-main-thread/quarkus-checkout-outboxI’ll walk through the changes in those files. The checkout already includes the supporting entities, request records, crash switch, and inspection endpoints. I assume you’re comfortable with Jakarta REST, Panache, and a @Transactional service method.
For anyone recreating the scaffold, this was my Quarkus CLI command. Run it in a separate directory from the cloned project:
quarkus create app com.themainthread:quarkus-checkout-outbox \
--platform-bom=io.quarkus.platform:quarkus-bom:3.39.4 \
--java=25 \
--extension=rest-jackson,hibernate-validator,hibernate-orm-panache,jdbc-postgresql,flyway,messaging-kafka,scheduler \
--no-codeThe REST and validation extensions handle incoming requests. Panache and the PostgreSQL driver handle persistence, with Flyway owning the schema. Kafka Messaging supplies the producer and consumer, and Scheduler runs the relay. The POM also includes quarkus-junit, REST Assured, and Awaitility for the tests; their versions come from the platform BOM.
Add the outbox beside the order
I kept purchase_orders as the checkout record and added two tables. outbox_event stores the event awaiting publication. fulfillment records the consumer’s database effect.
The migration is in src/main/resources/db/migration/V1__checkout_outbox.sql:
CREATE TABLE purchase_orders (
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
sku VARCHAR(40) NOT NULL,
quantity INTEGER NOT NULL CHECK (quantity > 0),
status VARCHAR(20) NOT NULL,
created_at TIMESTAMP WITH TIME ZONE NOT NULL
);
CREATE TABLE outbox_event (
id UUID PRIMARY KEY,
aggregatetype VARCHAR(255) NOT NULL,
aggregateid VARCHAR(255) NOT NULL,
type VARCHAR(255) NOT NULL,
payload JSONB NOT NULL,
created_at TIMESTAMP WITH TIME ZONE NOT NULL,
published_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX outbox_event_unpublished_idx
ON outbox_event (created_at)
WHERE published_at IS NULL;
CREATE TABLE fulfillment (
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
event_id UUID NOT NULL UNIQUE,
order_id BIGINT NOT NULL REFERENCES purchase_orders (id),
sku VARCHAR(40) NOT NULL,
quantity INTEGER NOT NULL,
created_at TIMESTAMP WITH TIME ZONE NOT NULL
);An outbox row contains everything the publisher needs to reconstruct the event. Its UUID remains the same across every send attempt. The order identifier becomes the Kafka record key, while the JSON payload carries the order details.
I chose the column names expected by the Debezium outbox event router. aggregatetype identifies the kind of entity, here Order; aggregateid identifies the individual order; and type describes the event, here OrderPlaced. These names also give us a defined mapping if we later adopt CDC.
For the polling relay, I added published_at. A null value means the row is eligible for another send. The partial index covers those rows, so the selection can avoid scanning the growing history of published events.
On the consumer side, the unique constraint on fulfillment.event_id lets PostgreSQL arbitrate competing inserts for the same event. We’ll use that directly in the consumer.
The entity mappings follow the migration’s identity columns: they extend PanacheEntityBase and declare GenerationType.IDENTITY. I keep Hibernate’s schema strategy at validate so startup checks the mappings against the Flyway schema.
Start with the two independent writes
Before changing checkout, I kept a version that saves the order and then sends to Kafka. Having both paths in the same application makes the failure easy to reproduce.
Both paths publish this record, defined in src/main/java/com/themainthread/checkout/OrderPlaced.java:
package com.themainthread.checkout;
import java.time.Instant;
import java.util.UUID;
public record OrderPlaced(
UUID eventId,
long orderId,
String sku,
int quantity,
Instant placedAt) {
static OrderPlaced of(PurchaseOrder order, UUID eventId) {
return new OrderPlaced(eventId, order.id, order.sku, order.quantity, order.createdAt);
}
}The producer in OrderPlacedProducer.java waits for acknowledgement using MutinyEmitter.sendMessageAndAwait. It also attaches the order id as the Kafka key and the event UUID as an id header:
package com.themainthread.checkout;
import java.nio.charset.StandardCharsets;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Message;
import io.smallrye.reactive.messaging.MutinyEmitter;
import io.smallrye.reactive.messaging.kafka.api.OutgoingKafkaRecordMetadata;
import jakarta.enterprise.context.ApplicationScoped;
@ApplicationScoped
public class OrderPlacedProducer {
private final MutinyEmitter<OrderPlaced> emitter;
OrderPlacedProducer(@Channel("order-placed") MutinyEmitter<OrderPlaced> emitter) {
this.emitter = emitter;
}
public void send(OrderPlaced event) {
emitter.sendMessageAndAwait(Message.of(event).addMetadata(metadata(event)));
}
private static OutgoingKafkaRecordMetadata<String> metadata(OrderPlaced event) {
return OutgoingKafkaRecordMetadata.<String>builder()
.withKey(Long.toString(event.orderId()))
.withHeaders(new RecordHeaders()
.add("id", event.eventId().toString().getBytes(StandardCharsets.UTF_8)))
.build();
}
}The key keeps records for an order on the same partition while the partition count stays unchanged. The event UUID gives the consumer an identifier for duplicate detection. I carry it in the payload as well as the header, so our typed consumer can read it directly.
Now look at DualWriteCheckoutService.java:
package com.themainthread.checkout;
import java.util.UUID;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.transaction.Transactional;
@ApplicationScoped
public class DualWriteCheckoutService {
private final OrderPlacedProducer producer;
private final CrashSwitch crashes;
DualWriteCheckoutService(OrderPlacedProducer producer, CrashSwitch crashes) {
this.producer = producer;
this.crashes = crashes;
}
public PurchaseOrder place(CheckoutRequest request, CrashPoint crashAfter) {
crashes.arm(crashAfter);
PurchaseOrder order = persistOrder(request);
crashes.maybeCrash(CrashPoint.COMMIT);
producer.send(OrderPlaced.of(order, UUID.randomUUID()));
crashes.maybeCrash(CrashPoint.KAFKA);
return order;
}
@Transactional
PurchaseOrder persistOrder(CheckoutRequest request) {
return PurchaseOrder.accepted(request.sku(), request.quantity());
}
}persistOrder runs in a database transaction. Quarkus intercepts this call even though it comes from another method in the same bean, so the transaction commits before control returns to place. At that point, the database contains an order. The event UUID and payload are created afterward, immediately before the Kafka send.
That ordering leaves a gap. A process exit at maybeCrash(COMMIT) preserves the order and loses the opportunity to send its event. Restarting the application gives the producer nothing to resume from.
I’ve made that exit deliberate through CrashSwitch. With checkout.crash.halt=true, it calls Runtime.getRuntime().halt(1) and skips shutdown hooks. A request header selects the point: X-Crash-After: commit or X-Crash-After: kafka. These endpoints are for the local experiment. Run one request at a time with an empty pending backlog, because the switch is shared across the application.
Put the event inside the checkout transaction
The change belongs in CheckoutService.java. I create OrderPlaced while saving the order, then serialize it into the outbox before the transaction commits:
package com.themainthread.checkout;
import java.util.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.transaction.Transactional;
@ApplicationScoped
public class CheckoutService {
private final ObjectMapper json;
private final CrashSwitch crashes;
CheckoutService(ObjectMapper json, CrashSwitch crashes) {
this.json = json;
this.crashes = crashes;
}
public PurchaseOrder place(CheckoutRequest request, CrashPoint crashAfter) {
crashes.arm(crashAfter);
PurchaseOrder order = persistOrderAndOutbox(request);
crashes.maybeCrash(CrashPoint.COMMIT);
return order;
}
@Transactional
PurchaseOrder persistOrderAndOutbox(CheckoutRequest request) {
PurchaseOrder order = PurchaseOrder.accepted(request.sku(), request.quantity());
OrderPlaced event = OrderPlaced.of(order, UUID.randomUUID());
OutboxEvent.enqueue(event, json);
return order;
}
}PurchaseOrder.accepted persists the order and obtains its identity. OutboxEvent.enqueue stores the event UUID, routing fields, and JSON payload. Both inserts participate in persistOrderAndOutbox‘s transaction. If either insert fails, PostgreSQL rolls the transaction back. After a successful commit, the publisher has a durable event to read.
I leave place outside that transaction so the simulated crash happens after commit. Moving @Transactional onto place would move the commit past maybeCrash(COMMIT) and change what the experiment demonstrates.
A correction to my CDI event example
In Decoupled by Design, I described persisting an outbox row from an AFTER_SUCCESS observer. Stefano asked whether that row belonged in the original business transaction. Yes, it does, and this is the change I should have shown there.
CDI’s AFTER_SUCCESS phase runs after successful transaction completion. An outbox insert at that point needs another transaction, leaving a committed order exposed if the second insert fails. An IN_PROGRESS observer invoked within the business transaction, or a BEFORE_COMPLETION observer, can participate before commit. Here I put the two inserts together in the service method so their shared boundary is visible.
Send committed events to Kafka
The publisher has three steps: select pending events, wait for Kafka to acknowledge each send, and then record publication in a fresh transaction. I keep the database transactions short while the Kafka call runs between them.
Here’s OutboxPublisher.java:
package com.themainthread.checkout;
import java.time.Instant;
import java.util.List;
import java.util.UUID;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.jboss.logging.Logger;
import io.quarkus.narayana.jta.QuarkusTransaction;
import io.quarkus.scheduler.Scheduled;
import jakarta.enterprise.context.ApplicationScoped;
@ApplicationScoped
public class OutboxPublisher {
private static final Logger LOG = Logger.getLogger(OutboxPublisher.class);
private static final int BATCH_SIZE = 20;
private final ObjectMapper json;
private final OrderPlacedProducer producer;
private final CrashSwitch crashes;
OutboxPublisher(ObjectMapper json, OrderPlacedProducer producer, CrashSwitch crashes) {
this.json = json;
this.producer = producer;
this.crashes = crashes;
}
@Scheduled(every = "{checkout.outbox.poll-every}", identity = "outbox-publisher",
concurrentExecution = Scheduled.ConcurrentExecution.SKIP)
void poll() {
publishPending();
}
public int publishPending() {
List<UUID> ids = QuarkusTransaction.requiringNew().call(this::unpublishedIds);
int published = 0;
for (UUID id : ids) {
if (publish(id)) {
published++;
}
}
return published;
}
public boolean publishOne(UUID id) {
return publish(id);
}
private List<UUID> unpublishedIds() {
return OutboxEvent.unpublished(BATCH_SIZE).stream().map(event -> event.id).toList();
}
private boolean publish(UUID id) {
OutboxEvent event = QuarkusTransaction.requiringNew().call(() -> OutboxEvent.findById(id));
if (event == null || event.publishedAt != null) {
return false;
}
OrderPlaced payload = readPayload(event);
producer.send(payload);
crashes.maybeCrash(CrashPoint.KAFKA);
QuarkusTransaction.requiringNew().run(() -> markPublished(id));
return true;
}
private void markPublished(UUID id) {
OutboxEvent event = OutboxEvent.findById(id);
if (event == null || event.publishedAt != null) {
return;
}
event.publishedAt = Instant.now();
LOG.infof("Published outbox event %s for order %s", id, event.aggregateid);
}
private OrderPlaced readPayload(OutboxEvent event) {
try {
return json.readValue(event.payload, OrderPlaced.class);
} catch (JsonProcessingException e) {
throw new IllegalStateException("Could not read outbox payload " + event.id, e);
}
}
}The order of producer.send and markPublished determines how recovery behaves. Recording publication first would allow a crash to leave a row marked complete before Kafka received it. Recording it after acknowledgement preserves the opportunity to send again.
That also explains the duplicate. Suppose Kafka accepts the record and the JVM exits on the next line. published_at is still null, so the next publisher sends the stored payload with the same event UUID. This is at-least-once delivery: recovery can repeat a send.
I set concurrentExecution = SKIP because a slow Kafka acknowledgement can take longer than the one-second schedule. Quarkus then skips an overlapping invocation in this JVM. Several application instances would need a shared claiming or locking scheme, with ownership lasting through the send. For this example, I run one publisher.
Let PostgreSQL recognize a repeated event
The fulfillment consumer can now receive the same event more than once. I want each delivery to reach the same database decision: insert a row for a new event id, or retain the existing row for a repeated id.
PostgreSQL’s ON CONFLICT (event_id) DO NOTHING expresses that decision in one statement. FulfillmentConsumer.java executes it inside a transaction:
package com.themainthread.checkout;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.jboss.logging.Logger;
import io.smallrye.reactive.messaging.annotations.Blocking;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.persistence.EntityManager;
import jakarta.transaction.Transactional;
@ApplicationScoped
public class FulfillmentConsumer {
private static final Logger LOG = Logger.getLogger(FulfillmentConsumer.class);
private final EntityManager entityManager;
FulfillmentConsumer(EntityManager entityManager) {
this.entityManager = entityManager;
}
@Incoming("fulfillment")
@Blocking
@Transactional
public void onOrderPlaced(OrderPlaced event) {
int inserted = entityManager.createNativeQuery("""
INSERT INTO fulfillment (event_id, order_id, sku, quantity, created_at)
VALUES (:eventId, :orderId, :sku, :quantity, CURRENT_TIMESTAMP)
ON CONFLICT (event_id) DO NOTHING
""")
.setParameter("eventId", event.eventId())
.setParameter("orderId", event.orderId())
.setParameter("sku", event.sku())
.setParameter("quantity", event.quantity())
.executeUpdate();
if (inserted == 0) {
LOG.infof("Skipping duplicate event %s for order %s", event.eventId(), event.orderId());
} else {
LOG.infof("Inserted fulfillment for order %s from event %s", event.orderId(), event.eventId());
}
}
}The returned update count tells us whether this delivery inserted a row. Two concurrent deliveries can reach the statement together; the unique index resolves the conflict, and both calls can complete normally. A foreign-key error, such as an event referring to a missing order, still propagates as a persistence failure.
I prefer this approach to catching a failed Hibernate flush. A constraint violation during a flush already marks the transaction for rollback, so catching the exception inside that transaction leaves us with an aborted transaction to finish.
If I add a stock decrement later, it belongs in this same transaction under inserted == 1. Then the event marker and the stock change commit together. For the current example, the fulfillment row itself is the entire business effect.
Connect the HTTP and Kafka endpoints
CheckoutResource exposes both checkout paths. /orders calls the outbox service, and /orders/dual-write calls the original save-then-send service:
@POST
@Path("/orders")
@Consumes(MediaType.APPLICATION_JSON)
@Blocking
public Response place(@Valid CheckoutRequest request, @HeaderParam("X-Crash-After") String crashAfter) {
PurchaseOrder order = checkout.place(request, CrashPoint.fromHeader(crashAfter));
return Response.created(URI.create("/orders/" + order.id))
.entity(OrderView.from(order))
.build();
}
@POST
@Path("/orders/dual-write")
@Consumes(MediaType.APPLICATION_JSON)
@Blocking
public Response dualWrite(@Valid CheckoutRequest request, @HeaderParam("X-Crash-After") String crashAfter) {
PurchaseOrder order = dualWrite.place(request, CrashPoint.fromHeader(crashAfter));
return Response.created(URI.create("/orders/" + order.id))
.entity(OrderView.from(order))
.build();
}The outbox path can return 201 once the order and event row have committed. The scheduler and consumer proceed independently, so their progress may be ahead of or behind the HTTP response. I added GET /orders/{id}, GET /outbox, and GET /fulfillments to inspect each stage. The blocking annotations keep JDBC calls, and the dual-write path’s wait for Kafka, on threads that permit blocking.
The channel and database settings live in src/main/resources/application.properties:
quarkus.application.name=checkout-outbox
quarkus.datasource.db-kind=postgresql
quarkus.datasource.jdbc.min-size=2
quarkus.datasource.jdbc.max-size=12
%dev,test.quarkus.datasource.devservices.image-name=docker.io/library/postgres:18.4-alpine3.24
%dev,test.quarkus.datasource.devservices.db-name=checkout
%dev,test.quarkus.datasource.devservices.username=checkout
%dev,test.quarkus.datasource.devservices.password=checkout
quarkus.flyway.migrate-at-start=true
quarkus.hibernate-orm.schema-management.strategy=validate
quarkus.hibernate-orm.log.sql=false
mp.messaging.outgoing.order-placed.connector=smallrye-kafka
mp.messaging.outgoing.order-placed.topic=order-placed
mp.messaging.outgoing.order-placed.cloud-events=false
mp.messaging.outgoing.order-placed.key.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.outgoing.order-placed.value.serializer=io.quarkus.kafka.client.serialization.ObjectMapperSerializer
mp.messaging.incoming.fulfillment.connector=smallrye-kafka
mp.messaging.incoming.fulfillment.topic=order-placed
mp.messaging.incoming.fulfillment.cloud-events=false
mp.messaging.incoming.fulfillment.auto.offset.reset=earliest
mp.messaging.incoming.fulfillment.group.id=checkout-fulfillment
mp.messaging.incoming.fulfillment.value.deserializer=com.themainthread.checkout.OrderPlacedDeserializer
checkout.outbox.poll-every=1s
checkout.crash.halt=true
# Tests drive the publisher themselves so crash-window assertions stay deterministic.
%test.quarkus.scheduler.enabled=false
%test.checkout.crash.halt=false
# Keep tests isolated even when local development enables container reuse.
%test.quarkus.datasource.devservices.reuse=false
%dev,test.quarkus.kafka.devservices.shared=falseI give the outgoing channel the name order-placed and the incoming channel the name fulfillment; both point to the order-placed topic. Jackson serializes the payload, and cloud-events=false keeps that payload as the record value. The incoming channel uses this OrderPlacedDeserializer:
package com.themainthread.checkout;
import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
public class OrderPlacedDeserializer extends ObjectMapperDeserializer<OrderPlaced> {
public OrderPlacedDeserializer() {
super(OrderPlaced.class);
}
}The consumer group remembers its Kafka offsets. Its earliest setting applies when that group has no committed offset for a partition, which lets a fresh group read events already present in the topic.
Follow an order through a restart
Let’s bring up the application. I need the database to survive the JVM exits, so I enable Testcontainers reuse before the first launch:
podman machine start
export TESTCONTAINERS_REUSE_ENABLE=true
./mvnw quarkus:devOn Linux with a running Podman socket, skip podman machine start. Keep the reuse variable set and the database settings unchanged for every restart. In a new shell, export it again. Flyway and Hibernate validation preserve the existing tables and rows.
Record the PostgreSQL container ID from the startup log; we’ll remove that container after the examples. The test profile disables database reuse so automated tests start with fresh data. Kafka sharing is disabled as well, which keeps other local applications away from this demo’s broker.
First, a completed checkout
curl -i \
-H 'Content-Type: application/json' \
-d '{"sku":"keyboard-1","quantity":1}' \
http://localhost:8080/ordersThe response from my recorded session looked like this. IDs and timestamps depend on the database’s existing contents:
HTTP/1.1 201 Created
Content-Type: application/json;charset=UTF-8
Location: http://localhost:8080/orders/1
{"id":1,"sku":"keyboard-1","quantity":1,"status":"ACCEPTED","createdAt":"2026-09-21T10:45:56.458659Z"}After publication, GET /outbox shows a publishedAt value. GET /fulfillments/order/1 then returns the fulfillment row. Its event id matches the outbox UUID. Those are the three stages we want to preserve through a restart: order, event, fulfillment.
Stop after the order commits
Now send the same kind of request through the dual-write endpoint, with the commit crash armed:
curl -i \
-H 'Content-Type: application/json' \
-H 'X-Crash-After: commit' \
-d '{"sku":"lost-event-1","quantity":1}' \
http://localhost:8080/orders/dual-writecurl reported Empty reply from server. After restarting with ./mvnw quarkus:dev in the same shell, I could retrieve the committed order:
{"id":2,"sku":"lost-event-1","quantity":1,"status":"ACCEPTED","createdAt":"2026-09-21T10:52:52.040135Z"}For this order, GET /outbox has no matching aggregate id and GET /fulfillments/order/2 returns 404. The process stopped before constructing and sending the event. This is the failure the outbox insert addresses.
Try that crash point through /orders:
curl -i \
-H 'Content-Type: application/json' \
-H 'X-Crash-After: commit' \
-d '{"sku":"headset-1","quantity":1}' \
http://localhost:8080/ordersThe connection closes again. Restart the application, then inspect the order and its outbox row. Here are the order and selected outbox fields from my run:
{"id":3,"sku":"headset-1","quantity":1,"status":"ACCEPTED","createdAt":"2026-09-21T10:53:48.690361Z"}{
"id": "090e844d-4265-4706-8b9a-146f8cd8267a",
"aggregatetype": "Order",
"aggregateid": "3",
"type": "OrderPlaced",
"publishedAt": "2026-09-21T10:54:09.189284Z"
}The publisher found the committed event after startup and sent it. Fulfillment for order 3 arrived with the same event id. The durable event gives this restart a precise action to resume.
Stop after Kafka acknowledges the send
The other crash point leaves us uncertain about whether publication was recorded:
curl -i \
-H 'Content-Type: application/json' \
-H 'X-Crash-After: kafka' \
-d '{"sku":"webcam-1","quantity":1}' \
http://localhost:8080/ordersIn my run, curl received 201 before the publisher sent the record and halted. The scheduler can also run before the response reaches the client, so response timing will vary. The crash happens immediately after the broker acknowledgement and before the published_at transaction.
On restart, the outbox row is eligible for another send. After publication, the event should have one fulfillment row. In this local setup, Testcontainers cleans up the Kafka container after a JVM exit and starts a fresh broker on the next launch. That sequence demonstrates recovery from the outbox. The companion test crashAfterKafkaRepublishesWithoutDuplicateFulfillment exercises repeated delivery on a single live broker: it sends twice, waits for the consumer group to commit past the second record, and checks that one fulfillment remains.
Replace the relay with Debezium when the application calls for it
The outbox transaction and event identifier can stay as they are when adopting CDC. Debezium takes over discovering committed inserts. Its outbox event router extracts the payload and routing metadata; with the default routing, aggregatetype=Order produces topic outbox.event.Order. Our consumer currently listens to order-placed, so I would align either the router’s topic configuration or the consumer subscription. I would also configure the converter so the consumer receives the expected JSON object. The router reference covers those settings.
For a team already running Kafka Connect, I would deploy the PostgreSQL connector there and let the relay have its own lifecycle. For an application that hosts capture itself, I would look at the Quarkus Debezium Engine extension. It brings connector configuration and capture callbacks into Quarkus, and its documentation includes experimental Dev Services support for a CDC-capable database.
Either route needs PostgreSQL logical replication configured, including the publication, replication slot, and appropriate permissions. The slot retains WAL needed by the connector, so I would monitor capture lag and retained WAL as part of operating it. These are database responsibilities that come with the PostgreSQL CDC path.
With an embedded engine, I would also choose durable offset storage and make sure completion of event handling follows successful Kafka publication. The extension’s introductory configuration uses an in-memory offset store; a restartable relay needs its capture position to survive process exit. The Debezium Engine documentation explains offset storage and processing completion. Recovery can still repeat events, so I would keep the consumer’s event-id constraint.
At that point, the scheduler and published_at updates can go. Row retention becomes an explicit cleanup policy: the router filters delete events, while the application or a separate job deletes old rows. I’d verify publication, restart, and cleanup together before switching the relay. The embedded extension deserves that full example of its own; this article’s runnable path remains the polling implementation.
Keep the recovery checks close to the code
The companion tests cover rollback of the order and outbox, repeated delivery, concurrent inserts for one event id, and propagation of a foreign-key failure. They give me a way to revisit those boundaries as I change the application.
Run them with the wrapper. I used separate HTTP test ports because another Quarkus test was running locally:
./mvnw test -Dquarkus.http.test-port=8094 -Dquarkus.http.test-ssl-port=8095All eleven tests passed. In the test profile, the crash switch throws CrashWindowException and the publisher is called directly, so the tests can control the send sequence. The real process-exit checks are recorded in VALIDATION.md. The Kafka reference and scheduler reference cover the APIs used here. My claims pipeline article continues into retries and dead-letter handling across a longer flow.
When you’re done, stop the application with Ctrl+C and remove the PostgreSQL container whose ID you recorded using podman rm -f <postgres-container-id>. That discards this demo database. Dev Services/Testcontainers cleans up the non-reused Kafka container separately.
For my checkout, the resulting contract is straightforward. Accepting an order also commits its event. Publication can resume after a process exit, and fulfillment recognizes the event when it arrives again. HTTP idempotency still handles repeated checkout requests, which could otherwise create separate orders and event ids. Whether I keep the scheduler or move to Debezium, I want those responsibilities to remain explicit: record the event with the order, resume delivery from durable state, and make a repeated delivery safe to process.