DEV Community: Alberto Haboba

Transactional outbox + idempotent Kafka consumers in Spring Boot 4

Alberto Haboba — Wed, 07 Oct 2026 16:56:52 +0000

Stop dual-writing to Postgres and Kafka. A small, boring pattern that keeps them in sync, with the real code.

If you've ever written something like this, you've got a bug. It just hasn't happened yet:

orderRepository.save(order);
kafkaTemplate.send("order.events", orderJson); // 💥 what if this fails? or the DB commit fails after it?

That's a dual write. Two systems, no shared transaction. If the DB commit fails after the send, downstream services react to an order that doesn't exist. If the send fails after the commit, the order exists and nobody ever hears about it. Retrying doesn't fix it, it just moves the window around.

The fix most teams end up with is the transactional outbox, plus idempotent consumers on the other side. Here's a version that runs on Spring Boot 4.1 / Java 21 with Spring Kafka, JPA and Flyway. The snippets are taken straight from a working project (an Order → Inventory saga), so they compile and they're covered by tests.

The idea in one paragraph

Don't publish to Kafka inside your business transaction. Insert a row into an outbox_events table in the same DB transaction as your business change. A separate publisher reads unpublished rows and sends them to Kafka. The DB is now the single source of truth: either both the order and its event exist, or neither does. You get at-least-once delivery, so consumers have to tolerate duplicates. That's the second half.

Step 1: one envelope for every event

Each message carries an eventId. That ID becomes the idempotency key later.

/**
 * Envelope for every Kafka message. {@code eventId} is the idempotency key.
 */
public record DomainEvent(
        UUID eventId,
        String type,
        Instant occurredAt,
        String aggregateType,
        UUID aggregateId,
        int version,
        String payloadJson
) {
    public static DomainEvent of(String type, String aggregateType, UUID aggregateId, int version, String payloadJson) {
        return new DomainEvent(UUID.randomUUID(), type, Instant.now(), aggregateType, aggregateId, version, payloadJson);
    }
}

Step 2: business row + outbox row, one transaction

The outbox table (Flyway migration):

CREATE TABLE outbox_events (
    id             UUID PRIMARY KEY,
    topic          VARCHAR(120) NOT NULL,
    partition_key  VARCHAR(120) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP WITH TIME ZONE NOT NULL,
    published_at   TIMESTAMP WITH TIME ZONE
);

CREATE INDEX idx_outbox_created ON outbox_events (created_at);

The writer just serializes the envelope and saves a row. Note that it has no @Transactional of its own. It joins whatever transaction the caller already opened:

public void enqueue(String topic, String partitionKey, DomainEvent event) {
    try {
        String json = objectMapper.writeValueAsString(event);
        repository.save(new OutboxEvent(event.eventId(), topic, partitionKey, json, Instant.now(clock)));
    } catch (JacksonException e) {
        throw new IllegalStateException("Cannot serialize domain event", e);
    }
}

(Small Boot 4 detail: that's Jackson 3, so the imports are tools.jackson.* and the exception is JacksonException.)

And the service that places an order:

@Transactional
public Order placeOrder(String productSku, int quantity) {
    if (quantity < 1) {
        throw new IllegalArgumentException("quantity must be >= 1");
    }
    Instant now = Instant.now(clock);
    Order order = new Order(UUID.randomUUID(), productSku, quantity, now);
    orderRepository.save(order);

    String payload = toJson(new OrderPayloads.OrderCreated(order.getId(), productSku, quantity));
    DomainEvent event = DomainEvent.of(
            EventTypes.ORDER_CREATED, "Order", order.getId(), 1, payload);
    outboxWriter.enqueue(props.topics().orders(), order.getId().toString(), event);
    return order;
}

Two things worth noticing:

No Kafka call here at all. The HTTP request returns 201 with the order in PENDING, and the event goes out asynchronously.
The partition key is the order ID. All events for the same order land on the same partition, so they're consumed in order.

Step 3: the publisher

A scheduled job polls unpublished rows, sends them, and marks them published:

@Scheduled(fixedDelayString = "${app.outbox.poll-interval-ms:500}")
@Transactional
public void publishPending() {
    List<OutboxEvent> batch = repository.findUnpublished().stream()
            .limit(props.outbox().batchSize())
            .toList();
    for (OutboxEvent event : batch) {
        try {
            kafkaTemplate.send(event.getTopic(), event.getPartitionKey(), event.getPayload()).get();
            event.markPublished(Instant.now(clock));
            log.info("Outbox published eventId={} topic={}", event.getId(), event.getTopic());
        } catch (Exception ex) {
            log.warn("Outbox publish failed eventId={}: {}", event.getId(), ex.getMessage());
            break;
        }
    }
}

@Query("""
        select e from OutboxEvent e
        where e.publishedAt is null
        order by e.createdAt asc
        """)
List<OutboxEvent> findUnpublished();

Why it's written this way:

.get() makes the send synchronous. A row is only marked published after the broker acknowledged it.
break on the first failure keeps ordering. We don't skip a failed event and publish the next one for the same order.
Where duplicates come from: if the send succeeds but the DB commit that sets published_at fails (crash, connection drop), the row is still "unpublished" and gets sent again next tick. That's the at-least-once part, and it's expected.

The producer side is set up to avoid making things worse:

spring:
  kafka:
    producer:
      acks: all
      properties:
        enable.idempotence: true

The idempotent producer prevents duplicates caused by the producer's own retries. It can't prevent the "sent, then crashed before commit" duplicate. Only the consumer can deal with that one.

Step 4: idempotent consumers

Every consumer records which events it already handled, keyed by (event_id, consumer_name):

CREATE TABLE processed_events (
    event_id       UUID         NOT NULL,
    consumer_name  VARCHAR(80)  NOT NULL,
    processed_at   TIMESTAMP WITH TIME ZONE NOT NULL,
    PRIMARY KEY (event_id, consumer_name)
);

The consumer name is part of the key on purpose. The same event can legitimately be processed by inventory-service and by some other consumer later.

The core of it is small:

@Transactional
public void processOnce(String consumerName, DomainEvent event, Consumer<DomainEvent> handler) {
    if (processedEventRepository.existsByEventIdAndConsumerName(event.eventId(), consumerName)) {
        log.info("Skipping duplicate eventId={} consumer={}", event.eventId(), consumerName);
        return;
    }
    handler.accept(event);
    processedEventRepository.save(new ProcessedEvent(event.eventId(), consumerName, Instant.now(clock)));
}

And a listener uses it like this:

@KafkaListener(topics = "${app.topics.orders}", groupId = "inventory-service")
public void onOrderEvent(String payload) {
    DomainEvent event = processor.parse(payload);
    processor.processOnce(CONSUMER, event, this::handle);
}

The important part is what's inside that one transaction: the duplicate check, the handler's DB changes, and the processed_events insert. In the sample, the inventory handler reserves stock and then writes its own inventory.reserved (or inventory.failed) event to the outbox, all inside the same transaction. So "consume → update state → emit next event" is atomic from the database's point of view:

Crash before commit? Nothing happened, and Kafka redelivers.
Crash after commit but before the offset is committed? Kafka redelivers, existsBy... returns true, and it's skipped.
Two copies processed concurrently? The composite primary key makes the second insert fail, its transaction rolls back (including the handler's changes), and on retry it's skipped as a duplicate.

Offsets are only committed after the listener returns (enable-auto-commit: false, so the Spring Kafka container handles commits), which is exactly the ordering you want here.

Step 5: retries and a dead letter topic

Not every failure is transient. The listener container gets a DefaultErrorHandler with exponential backoff and a dead-letter recoverer:

DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
        (record, ex) -> {
            // routes to order.events.DLT or inventory.events.DLT based on record.topic()
            // ...
        });

ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
backOff.setInitialInterval(200L);
backOff.setMultiplier(2.0);
backOff.setMaxInterval(2_000L);

DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, backOff);
errorHandler.addNotRetryableExceptions(IllegalArgumentException.class);
factory.setCommonErrorHandler(errorHandler);

So that's 3 retries at roughly 200 ms, 400 ms and 800 ms, then the record goes to the DLT, and the partition keeps moving instead of blocking on one poison message. IllegalArgumentException is marked non-retryable, and that's what the parser throws for malformed JSON. Retrying a broken payload three times just wastes time.

Two practical notes if you adapt this:

Choose your exception types on purpose. Whatever you mark non-retryable skips the backoff entirely. In the sample, the handlers wrap their own failures in IllegalArgumentException. If your handler can hit transient problems (a DB timeout, a downstream call), let those escape as a different exception type so they get retried.
Watch DLT partition counts. If your resolver targets the same partition number as the source record, the DLT needs at least as many partitions as the source topic. Otherwise pick a partition that exists.

Testing it without Docker

The unit test for the dedupe logic is almost boring, which is the point:

@Test
void skipsDuplicateEvent() {
    DomainEvent event = DomainEvent.of("order.created", "Order", UUID.randomUUID(), 1, "{}");
    when(repository.existsByEventIdAndConsumerName(event.eventId(), "c1")).thenReturn(true);
    AtomicInteger calls = new AtomicInteger();

    processor.processOnce("c1", event, e -> calls.incrementAndGet());

    assertThat(calls.get()).isZero();
    verify(repository, never()).save(any());
}

For the full flow, an @EmbeddedKafka + H2 integration test posts an order over HTTP and uses Awaitility to wait until it becomes CONFIRMED (or REJECTED when stock is insufficient). It exercises the outbox, the publisher, both listeners and the saga, and ./mvnw verify runs it without Docker.

Trade-offs (read before copying into prod)

This is the simple, polling flavor of the outbox. It's a good default, but be honest with yourself about scale:

Polling vs CDC. Polling every 500 ms is easy to reason about. At high volume, log-based CDC (e.g. Debezium reading the outbox table) cuts latency and DB load.
Batch size. Here the batch limit is applied after the query runs. With a large backlog, push the limit into the query itself (paging).
Multiple instances. Two app instances can pick up the same unpublished rows. Consumers dedupe, so it's correct, but it's wasteful. SELECT ... FOR UPDATE SKIP LOCKED or a single active publisher fixes that.
The transaction stays open during the sends. That's fine for small batches. Keep batches small.
Housekeeping. outbox_events and processed_events grow forever unless you prune them. A periodic delete of old published rows is enough.

None of this is exotic. It's just the stuff that's easy to skip on day one and painful to retrofit on day ninety.

I packaged this whole setup (outbox, idempotent consumers, retries + DLT, the Order → Inventory saga sample, Docker Compose with Postgres 17 + Kafka KRaft, CI, and 16 tests) as a paid source-code starter, if you'd rather not wire it from scratch: Spring Boot 4 Kafka Event Starter (25 USD). Built by me, a Java backend engineer in Mexico City (GitHub: kodkodmx). The article stands on its own, and you don't need to buy anything to use the pattern.

Rotating refresh tokens with reuse detection in Spring Boot 4 (no extra JWT library)

Alberto Haboba — Wed, 07 Oct 2026 16:50:01 +0000

Short-lived JWTs, opaque hashed refresh tokens, and what to do when an old token shows up again.

Most JWT tutorials stop at "sign a token, put it in a header." Then real life shows up. Access tokens that live for days can't be revoked, and refresh tokens that never change are a permanent key for anyone who copies one.

Here's the setup I use in Spring Boot 4.1 / Java 21:

Access token: a JWT (HS256) that lives for 15 minutes. It's signed with Spring Security's own Nimbus encoder, so there's no extra JWT library.
Refresh token: a random opaque string (not a JWT) that lives for 30 days. The database stores only its SHA-256 hash.
Rotation: every refresh returns a new refresh token and revokes the old one.
Reuse detection: if a revoked refresh token is ever presented again, assume it was stolen and revoke the whole token family.

All snippets below come from a working, tested project. Let's go piece by piece.

Why rotation + reuse detection?

Say an attacker copies a user's refresh token. Without rotation, they can mint access tokens for 30 days and you'll never know.

With rotation, the attacker and the real user now hold the same token, and only one of them can use it. Whoever refreshes second presents a token that's already been rotated. That's the signal. You can't tell which party is the attacker, so you kill the whole chain (the "family") and force a fresh login. The real user gets logged out once. The attacker's access ends right there.

The table

CREATE TABLE refresh_tokens (
    id           UUID        PRIMARY KEY,
    user_id      UUID        NOT NULL REFERENCES users (id) ON DELETE CASCADE,
    token_hash   VARCHAR(64) NOT NULL UNIQUE,
    family_id    UUID        NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL,
    expires_at   TIMESTAMPTZ NOT NULL,
    revoked_at   TIMESTAMPTZ,
    replaced_by  UUID
);
CREATE INDEX ix_refresh_tokens_family  ON refresh_tokens (family_id);
CREATE INDEX ix_refresh_tokens_user    ON refresh_tokens (user_id);
CREATE INDEX ix_refresh_tokens_expires ON refresh_tokens (expires_at);

token_hash: we never store the raw token. A DB dump doesn't give anyone usable tokens.
family_id: every token created from the same login shares it.
revoked_at / replaced_by: the rotation chain. A revoked token points to the token that replaced it.

Issuing tokens

The refresh token is 32 random bytes, URL-safe:

/** 256 bits of randomness, URL-safe, no padding. */
public String newRefreshTokenValue() {
    byte[] bytes = new byte[32];
    RANDOM.nextBytes(bytes);
    return Base64.getUrlEncoder().withoutPadding().encodeToString(bytes);
}

public static String sha256(String value) {
    try {
        MessageDigest digest = MessageDigest.getInstance("SHA-256");
        return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
    } catch (NoSuchAlgorithmException e) {
        throw new IllegalStateException("SHA-256 not available", e);
    }
}

"Why SHA-256 and not bcrypt like passwords?" Passwords are low-entropy, so they need a slow, salted hash. A 256-bit random token can't be brute-forced, so a fast unsalted hash is enough. It's also deterministic, which means we can look the token up by its hash with a unique index. You can't do that with salted bcrypt.

The access token uses Spring Security's JwtEncoder:

public String createAccessToken(User user) {
    Instant now = clock.instant();
    JwtClaimsSet claims = JwtClaimsSet.builder()
            .issuer(props.jwt().issuer())
            .subject(user.getId().toString())
            .issuedAt(now)
            .expiresAt(now.plus(props.jwt().accessTokenTtl()))
            .claim("email", user.getEmail())
            .claim(SecurityConfig.ROLES_CLAIM, List.of(user.getRole().name()))
            .build();
    JwsHeader header = JwsHeader.with(MacAlgorithm.HS256).build();
    return encoder.encode(JwtEncoderParameters.from(header, claims)).getTokenValue();
}

The refresh flow

This is the heart of it:

/**
 * Rotates a refresh token. Presenting a token that was already rotated/revoked is treated as
 * theft: the whole token family is revoked and the client must log in again.
 */
@Transactional(noRollbackFor = ApiException.class)
public TokenResponse refresh(String rawRefreshToken) {
    Instant now = clock.instant();
    RefreshToken current = refreshTokens.findByTokenHash(TokenService.sha256(rawRefreshToken))
            .orElseThrow(AuthService::invalidRefreshToken);

    if (current.isRevoked()) {
        int revoked = refreshTokens.revokeFamily(current.getFamilyId(), now);
        log.warn("Refresh token reuse detected for user {} (family {}); revoked {} active token(s)",
                current.getUserId(), current.getFamilyId(), revoked);
        throw invalidRefreshToken();
    }
    if (current.isExpired(now)) {
        throw invalidRefreshToken();
    }
    User user = users.findById(current.getUserId())
            .filter(User::isEnabled)
            .orElseThrow(AuthService::invalidRefreshToken);

    IssuedRefreshToken next = createRefreshToken(user, current.getFamilyId(), now);
    current.revoke(now, next.entity().getId());
    return response(user, next.rawValue());
}

Three details are easy to miss:

1. `noRollbackFor = ApiException.class`

When reuse is detected, we run revokeFamily(...) and then throw so the client gets a 401. Spring rolls back transactions on runtime exceptions by default, which would quietly undo the revocation. That's the one write we really care about. noRollbackFor keeps it.

2. A row lock on lookup

/** Row lock prevents two concurrent refreshes with the same token from both succeeding. */
@Lock(LockModeType.PESSIMISTIC_WRITE)
Optional<RefreshToken> findByTokenHash(String tokenHash);

@Modifying
@Query("update RefreshToken t set t.revokedAt = :now where t.familyId = :familyId and t.revokedAt is null")
int revokeFamily(@Param("familyId") UUID familyId, @Param("now") Instant now);

Without the lock, two requests carrying the same token could both read it as "not revoked" and both rotate it. You'd end up with two valid descendants. With SELECT ... FOR UPDATE, the second request waits, then sees the token as revoked.

3. Every failure gives the same answer

Unknown token, expired, revoked, disabled user: the client always gets the same 401 with the same error code (INVALID_REFRESH_TOKEN). The response doesn't tell anyone which case they hit.

Logout

Logout revokes the family, and it's idempotent:

/** Revokes the token family of the given refresh token. Idempotent: unknown tokens are ignored. */
@Transactional
public void logout(String rawRefreshToken) {
    refreshTokens.findByTokenHash(TokenService.sha256(rawRefreshToken))
            .ifPresent(t -> refreshTokens.revokeFamily(t.getFamilyId(), clock.instant()));
}

The access token itself stays valid until it expires (at most 15 minutes). That's the trade-off you accept with stateless JWTs, and it's why the access TTL is short.

The trade-off you should know about

Strict reuse detection has a known side effect: legit races look like theft. If a client fires two refreshes with the same token at once (two browser tabs, or a retry after a network timeout), the second one finds a revoked token and the family gets killed. The user just has to log in again.

This setup chooses strictness. Some systems add a short grace window where the immediately previous token is still accepted. That's more forgiving, but it's also a small window an attacker can use. Whichever you pick, make the client refresh from one place (a single in-flight refresh promise, for example) and you'll rarely hit this.

Proving it with a test

The integration test runs against a real PostgreSQL (Testcontainers) and checks exactly the behavior described above:

@Test
void refreshRotatesTokenAndReuseRevokesTheWholeFamily() throws Exception {
    Tokens first = register(uniqueEmail());

    Tokens second = refresh(first.refreshToken());
    assertThat(second.refreshToken()).isNotEqualTo(first.refreshToken());

    // Replaying the already-rotated token = suspected theft -> 401 ...
    refreshRaw(first.refreshToken())
            .andExpect(status().isUnauthorized())
            .andExpect(jsonPath("$.code").value("INVALID_REFRESH_TOKEN"));
    // ... and the legitimately rotated token is revoked too.
    refreshRaw(second.refreshToken()).andExpect(status().isUnauthorized());
}

That last line matters. A test that only checks "the old token fails" would pass even without family revocation.

Housekeeping

Expired tokens pile up, so an hourly job deletes the ones that expired more than a day ago:

@Scheduled(cron = "${app.jobs.refresh-token-cleanup-cron:0 0 * * * *}")
@Transactional
public void purgeExpired() {
    int deleted = refreshTokens.deleteExpiredBefore(clock.instant().minus(Duration.ofDays(1)));
    if (deleted > 0) {
        log.info("Purged {} expired refresh token(s)", deleted);
    }
}

Bonus: don't leak which emails exist

Not refresh-specific, but it lives in the same service. On login, an unknown email still runs a password check against a dummy hash, so it takes about as long as a wrong password:

if (user == null) {
    passwordEncoder.matches(request.password(), dummyHash);
    throw invalidCredentials();
}

Both cases return the same 401 INVALID_CREDENTIALS, and there's a test for that too.

Recap

Short-lived JWT access tokens, long-lived opaque refresh tokens.
Store only a SHA-256 hash of the refresh token and look it up by hash.
Rotate on every refresh, and group tokens into families.
A revoked token showing up again means revoke the family, and make sure that write isn't rolled back.
Lock the row during refresh so concurrent requests can't both win.
Test the second half of reuse detection, not just the first.

This is one piece of a bigger baseline I packaged as a paid source-code starter: register/login/refresh/logout, roles, PostgreSQL + Flyway, RFC 9457 error responses, OpenAPI, 30 tests with Testcontainers, Docker and GitHub Actions. If you want the whole thing instead of wiring it yourself: Spring Boot 4 API Starter (29 USD). Built by me, a Java backend engineer in Mexico City (GitHub: kodkodmx). Everything above works without buying anything.