Apache Kafka with Java: Architecture, Idempotence, and End-to-End Resilience

What actually happens when your application executes kafkaTemplate.send(...)?

If your answer is simply “the message arrives in a queue and a worker processes it”, you are operating Apache Kafka under the mental model of a traditional message broker like RabbitMQ or AWS SQS. Kafka is fundamentally different: it is an immutable, append-only, distributed commit log.

This difference changes everything: how you partition workloads, how consumers scale, how you guarantee delivery order, and, most importantly, how you prevent duplicate processing or data loss when networks glitch or nodes crash.

This article is Part 1 of an in-depth engineering series on Event-Driven Architecture (EDA) and messaging in enterprise systems. We will build an end-to-end financial transaction pipeline using Java 25 (LTS), Spring Boot 4.1, and Apache Kafka 4.3 (KRaft mode). We will implement:

The Transactional Outbox Pattern: bridging relational databases and Kafka without distributed transactions (XA) or two-phase commit.
Kafka Transactions & Exactly-Once Semantics (EOS): achieving end-to-end consistency across read-process-write loops.
Idempotent Consumers: double-entry bookkeeping and deterministic row locking to prevent duplicate debiting under at-least-once delivery.
Resilient Error Handling: blocking exponential retries with Dead Letter Topics (DLT) for strictly ordered feeds, alongside non-blocking retry topics for independent workloads.
Production Observability & Testing: Micrometer metrics, OpenTelemetry tracing, and real integration testing with Testcontainers 2.x.

1. The Mental Model: Why Kafka is Not RabbitMQ or SQS

Before touching configuration or code, we must align on the underlying mechanics.

Traditional Queue (RabbitMQ / SQS):
[ Producer ] ──► [ Queue ] ──► Message consumed and DELETED from broker
                           └──► Multiple workers compete for the SAME messages

Kafka Distributed Log:
[ Producer ] ──► [ Partition Log: [0][1][2][3][4][5]... ] (Never deleted on read)
                           ▲                  ▲
                           │                  │
                    Consumer A (Offset 2)  Consumer B (Offset 5)

In traditional message queues:

The broker manages state: it tracks which consumer received which message, waits for an ack, and deletes or hides the message once confirmed.
Messages disappear after consumption.
Scaling consumption beyond a single thread is simply a matter of adding more worker threads or instances pulling from the same queue.

In Apache Kafka:

The broker is a passive, dumb log: messages are appended to the end of a commit log on disk and retained based on time or size (e.g., 7 days), regardless of whether they have been read.
The consumer manages state: a consumer merely maintains an integer pointer called an offset. Reading an event does not mutate or delete anything on the broker. A consumer can rewind its offset to re-read history from three days ago, or replay events into a brand-new service.
Consumption scales strictly by partitions: within a consumer group, each partition is assigned to at most one consumer thread. If a topic has 12 partitions, you can run at most 12 concurrent consumers in that group. Adding a 13th consumer means it sits idle.

Ordering Guarantees and Partitions

A universal truth of distributed systems: Kafka only guarantees message order within a single partition. There is no global order across an entire topic.

When a producer publishes a record:

If a record key is provided, Kafka hashes the key using MurmurHash2: $$\text{partition} = \text{murmur2}(\text{key}) \pmod{\text{number of partitions}}$$ All events sharing the same key always land on the exact same partition in the exact order they were sent.
If no key is provided, modern Kafka producers batch records and round-robin across partitions to maximize throughput and disk utilization.

For financial systems, orders, or user account streams, selecting an appropriate key (e.g., accountId or orderId) is essential to preserve strict chronological ordering.

2. Core Concepts: Brokers, KRaft, and Consumer Groups

The KRaft Metadata Quorum (No More ZooKeeper)

In Apache Kafka 4.x, ZooKeeper has been completely removed in favor of KRaft (Kafka Raft Metadata Mode). The brokers elect a leader among a designated quorum of controllers using the Raft consensus protocol. Metadata changes (topic creation, partition reassignments) are committed directly to an internal metadata topic log replicated across the controllers. This yields sub-second controller failovers and allows clusters to scale to millions of partitions without third-party daemon dependencies.

Replicas, Leaders, and In-Sync Replicas (ISR)

Every partition has:

One Leader broker: handles all reads and writes for that partition.
Zero or more Follower brokers: continuously fetch log segments from the leader to keep their local copies in sync.

The subset of followers caught up with the leader within a configured lag threshold (replica.lag.time.max.ms) forms the ISR (In-Sync Replicas) set.

Topic: transferencias.solicitada (Partition 0)
┌────────────────────────────────────────────────────────┐
│ Broker 1 (LEADER)     │ [Record 0] [Record 1] [Record 2]│
├───────────────────────┼────────────────────────────────┤
│ Broker 2 (ISR Follower)│ [Record 0] [Record 1] [Record 2]│
├───────────────────────┼────────────────────────────────┤
│ Broker 3 (Lagging)    │ [Record 0]                     │ (Dropped from ISR)
└────────────────────────────────────────────────────────┘

When a producer sends with acks=all, the write is only acknowledged to the client after every member of the current ISR has committed that record to disk. Coupled with min.insync.replicas=2, if the ISR drops below 2 nodes, the leader refuses further writes, preventing split-brain scenarios and catastrophic data loss.

Consumer Groups and KIP-848

A Consumer Group represents a coordinated logical subscriber. When multiple application instances join the same consumer group, the partitions of subscribed topics are divided among them.

Kafka 4.x introduces the production-ready KIP-848 Next Generation Consumer Rebalance Protocol (group.protocol: consumer). Unlike the legacy “eager” or “cooperative sticky” client-side rebalance protocols—which required global synchronization barriers that temporarily halted consumption across the entire group—KIP-848 delegates partition assignment to the broker’s Group Coordinator. Assignments are communicated incrementally via heartbeats, reducing rebalance latency from tens of seconds to milliseconds without stop-the-world pauses.

3. The Architecture Scenario: Financial Transfer Pipeline

To put these patterns into practice, we design a mission-critical financial processing pipeline. In this domain, losing a message results in missing funds, and processing a message twice results in double charges.

                        ┌──────────────────────────────────────────────────┐
                        │             transferencias-service               │
                        │  (REST API -> PostgreSQL -> Transactional Outbox)│
                        └─────────────────────────┬────────────────────────┘
                                                  │
                                                  ▼
                                 [ Topic: transferencias.solicitada ]
                                                  │
                                                  ▼
                        ┌──────────────────────────────────────────────────┐
                        │               antifraude-service                 │
                        │        (Read -> Evaluate -> Kafka Tx Produce)    │
                        └─────────────┬──────────────────────────┬─────────┘
                                      │                          │
                 (If Risk Score < 80) │                          │ (If Risk Score >= 80)
                                      ▼                          ▼
               [ Topic: transferencias.analisada ]   [ Topic: transferencias.rejeitada ]
                                      │                          │
                                      ▼                          ▼
                        ┌──────────────────────────┐   ┌──────────────────────────┐
                        │      ledger-service      │   │   notificacoes-service   │
                        │ (Idempotent DB Settlement│   │ (Non-blocking retry topic│
                        │   Double-entry Ledger)   │   │     Email/SMS alerts)    │
                        └─────────────┬────────────┘   └──────────────────────────┘
                                      │
                                      ▼
                        [ Topic: transferencias.liquidada ]
                                      │
                                      ▼
                        ┌──────────────────────────────────────────────────┐
                        │  transferencias-service (Status Projection)      │
                        │  notificacoes-service (Customer Push Alerts)     │
                        └──────────────────────────────────────────────────┘

4. Service 1: `transferencias-service` and Transactional Outbox

The Dual-Write Problem

A common architectural trap in microservices is the dual-write:

// ANTI-PATTERN: NEVER DO THIS IN PRODUCTION
@Transactional
public void criarTransferencia(TransferenciaRequest req) {
    transferenciaRepository.save(new Transferencia(...)); // 1. Writes to DB
    kafkaTemplate.send("transferencias.solicitada", req); // 2. Writes to Kafka
}

What happens if step 1 succeeds, but step 2 throws a network timeout? The database holds the record, but Kafka never gets the event. Conversely, what happens if step 2 succeeds, but the database transaction rolls back during commit due to a constraint violation? Kafka received an event for an entity that does not exist.

Distributed 2-Phase Commit (XA) across databases and message brokers introduces crippling latency, coordinator single-points-of-failure, and poor ecosystem support.

The industry solution is the Transactional Outbox Pattern:

[ Client Request ]
       │
       ▼
┌──────────────────────────────────────────────────────────────────┐
│  Atomic Local Database Transaction (ACID)                        │
│                                                                  │
│  1. INSERT INTO transferencia (...)                              │
│  2. INSERT INTO outbox_evento (topico, chave, payload, criado_em)│
└──────────────────────────────────┬───────────────────────────────┘
                                   │ Committed
                                   ▼
┌──────────────────────────────────────────────────────────────────┐
│  Outbox Relay (Asynchronous Poller or Debezium CDC)              │
│                                                                  │
│  SELECT ... FROM outbox_evento WHERE publicado_em IS NULL       │
│  FOR UPDATE SKIP LOCKED                                          │
│                                                                  │
│  kafkaTemplate.send(...) ──► [ Kafka Broker (acks=all) ]         │
│                                   │ ACK                          │
│  UPDATE outbox_evento SET publicado_em = now() WHERE id = ...    │
└──────────────────────────────────────────────────────────────────┘

Outbox Schema and Domain Persistence

-- V1__schema.sql (transferencias-service)
CREATE TABLE transferencia (
    id                UUID PRIMARY KEY,
    conta_origem      VARCHAR(32)    NOT NULL,
    conta_destino     VARCHAR(32)    NOT NULL,
    valor_centavos    BIGINT         NOT NULL CHECK (valor_centavos > 0),
    moeda             CHAR(3)        NOT NULL,
    status            VARCHAR(20)    NOT NULL,
    motivo_rejeicao   TEXT,
    criada_em         TIMESTAMPTZ    NOT NULL DEFAULT now(),
    atualizada_em     TIMESTAMPTZ    NOT NULL DEFAULT now()
);

CREATE TABLE outbox_evento (
    id            UUID PRIMARY KEY,
    topico        VARCHAR(255) NOT NULL,
    chave         VARCHAR(64)  NOT NULL,
    tipo          VARCHAR(100) NOT NULL,
    payload       TEXT         NOT NULL,
    criado_em     TIMESTAMPTZ  NOT NULL DEFAULT now(),
    publicado_em  TIMESTAMPTZ
);

CREATE INDEX idx_outbox_pendentes ON outbox_evento (criado_em) WHERE publicado_em IS NULL;

Here is the domain service writing both records inside a single local Spring @Transactional boundary:

package com.exemplo.pagamentos.transferencias.aplicacao;

import com.exemplo.pagamentos.transferencias.dominio.Transferencia;
import com.exemplo.pagamentos.transferencias.eventos.TransferenciaSolicitada;
import com.exemplo.pagamentos.transferencias.infra.outbox.PublicadorDeEventos;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.UUID;

@Service
public class CriarTransferenciaService {

    private final JdbcClient jdbc;
    private final PublicadorDeEventos publicador;

    public CriarTransferenciaService(JdbcClient jdbc, PublicadorDeEventos publicador) {
        this.jdbc = jdbc;
        this.publicador = publicador;
    }

    @Transactional
    public Transferencia executar(String contaOrigem, String contaDestino, long valorEmCentavos, String moeda) {
        var transferencia = new Transferencia(
                UUID.randomUUID(), contaOrigem, contaDestino, valorEmCentavos, moeda, "PENDENTE", Instant.now());

        // 1. Grava a entidade de negócio
        jdbc.sql("""
                INSERT INTO transferencia (id, conta_origem, conta_destino, valor_centavos, moeda, status, criada_em, atualizada_em)
                VALUES (:id, :origem, :destino, :valor, :moeda, :status, :criadaEm, :criadaEm)
                """)
                .param("id", transferencia.id())
                .param("origem", transferencia.contaOrigem())
                .param("destino", transferencia.contaDestino())
                .param("valor", transferencia.valorEmCentavos())
                .param("moeda", transferencia.moeda())
                .param("status", transferencia.status())
                .param("criadaEm", transferencia.criadaEm())
                .update();

        // 2. Grava o evento no outbox na mesma transação
        var evento = new TransferenciaSolicitada(
                UUID.randomUUID(),
                transferencia.id(),
                transferencia.contaOrigem(),
                transferencia.contaDestino(),
                transferencia.valorEmCentavos(),
                transferencia.moeda(),
                transferencia.criadaEm());

        publicador.publicar(
                Topicos.SOLICITADA,
                transferencia.contaOrigem(),
                "TransferenciaSolicitada",
                evento.eventoId(),
                evento);

        return transferencia;
    }
}

The publisher serializes the payload to JSON and inserts into outbox_evento:

package com.exemplo.pagamentos.transferencias.infra.outbox;

import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Component;
import tools.jackson.databind.json.JsonMapper;

import java.util.UUID;

@Component
public class PublicadorDeEventos {

    private final JdbcClient jdbc;
    private final JsonMapper jsonMapper;

    public PublicadorDeEventos(JdbcClient jdbc, JsonMapper jsonMapper) {
        this.jdbc = jdbc;
        this.jsonMapper = jsonMapper;
    }

    public void publicar(String topico, String chave, String tipo, UUID eventoId, Object payload) {
        String json = jsonMapper.writeValueAsString(payload);
        jdbc.sql("""
                INSERT INTO outbox_evento (id, topico, chave, tipo, payload)
                VALUES (:id, :topico, :chave, :tipo, :payload)
                """)
                .param("id", eventoId)
                .param("topico", topico)
                .param("chave", chave)
                .param("tipo", tipo)
                .param("payload", json)
                .update();
    }
}

The Outbox Relay

To deliver pending events to Kafka without race conditions across multiple application instances, we use FOR UPDATE SKIP LOCKED. This allows each polling instance to lock and stream a batch of unpublished records without stepping on concurrent worker nodes:

package com.exemplo.pagamentos.transferencias.infra.outbox;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);
    private static final int TAMANHO_LOTE = 100;
    private static final long TIMEOUT_ENVIO_MS = 10_000;

    private final JdbcClient jdbc;
    private final KafkaTemplate<String, String> kafka;
    private final TransactionTemplate transacao;

    public OutboxRelay(JdbcClient jdbc, KafkaTemplate<String, String> kafka,
                       PlatformTransactionManager transactionManager) {
        this.jdbc = jdbc;
        this.kafka = kafka;
        this.transacao = new TransactionTemplate(transactionManager);
    }

    @Scheduled(fixedDelay = 200)
    public void publicarPendentes() {
        transacao.executeWithoutResult(status -> {
            List<EventoOutbox> lote = jdbc.sql("""
                    SELECT id, topico, chave, tipo, payload
                    FROM outbox_evento
                    WHERE publicado_em IS NULL
                    ORDER BY criado_em
                    LIMIT :limite
                    FOR UPDATE SKIP LOCKED
                    """)
                    .param("limite", TAMANHO_LOTE)
                    .query(EventoOutbox.class)
                    .list();

            if (lote.isEmpty()) {
                return;
            }

            List<CompletableFuture<?>> envios = lote.stream().map(this::enviar).toList();
            aguardar(envios);

            jdbc.sql("UPDATE outbox_evento SET publicado_em = now() WHERE id IN (:ids)")
                    .param("ids", lote.stream().map(EventoOutbox::id).toList())
                    .update();

            log.debug("Outbox: {} evento(s) publicado(s)", lote.size());
        });
    }

    private CompletableFuture<?> enviar(EventoOutbox evento) {
        var registro = new ProducerRecord<>(evento.topico(), evento.chave(), evento.payload());
        registro.headers().add("evento-id", evento.id().toString().getBytes(StandardCharsets.UTF_8));
        registro.headers().add("evento-tipo", evento.tipo().getBytes(StandardCharsets.UTF_8));
        return kafka.send(registro);
    }

    private void aguardar(List<CompletableFuture<?>> envios) {
        try {
            CompletableFuture.allOf(envios.toArray(CompletableFuture[]::new))
                    .get(TIMEOUT_ENVIO_MS, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Publicação do outbox interrompida", e);
        } catch (ExecutionException | TimeoutException e) {
            throw new IllegalStateException("Falha ao publicar lote do outbox", e);
        }
    }

    record EventoOutbox(UUID id, String topico, String chave, String tipo, String payload) {}
}

Key trade-offs of this approach:

Zero Loss Window: The database row is marked as published only after Kafka acknowledges receipt. If an instance crashes mid-send, the uncommitted lock releases, and another node safely picks up the batch.
At-Least-Once Delivery: If the process crashes after Kafka acks the record but before the UPDATE outbox_evento commits, the events will be resent upon recovery. Downstream consumers must be idempotent.
Ordering: With multiple relay instances polling concurrently via SKIP LOCKED, events for the same account might theoretically be sent slightly out of order if batches interleave. If strict sub-millisecond sequential ordering is non-negotiable, run a single leader instance via Spring Cloud Leadership election, or use Change Data Capture (CDC) with Debezium streaming from PostgreSQL’s write-ahead log (WAL).

Producer Configuration for Maximum Durability

# application.yml (transferencias-service)
spring:
  application:
    name: transferencias-service
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    producer:
      acks: all
      compression-type: zstd
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        enable.idempotence: true
        delivery.timeout.ms: 120000
        linger.ms: 5
    template:
      observation-enabled: true
    listener:
      observation-enabled: true

app:
  kafka:
    replicas: 1   # Use 3 in production

Configuration rationale:

acks: all: The leader broker will not acknowledge the write until all In-Sync Replicas have flushed it.
enable.idempotence: true: The Kafka client assigns a unique Producer ID (PID) and sequence numbers to every message. If network retries occur, the broker detects duplicates and discards them without duplicating log entries.
delivery.timeout.ms: 120000: Upper ceiling on total send retries before failing.
linger.ms: 5 and compression-type: zstd: High-performance batching with minimal latency cost, optimizing broker network I/O.

Topic Provisioning

package com.exemplo.pagamentos.transferencias.infra.kafka;

import com.exemplo.pagamentos.transferencias.aplicacao.Topicos;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

import java.time.Duration;

@Configuration
class TopicosConfig {

    @Bean
    NewTopic solicitada(@Value("${app.kafka.replicas:3}") int replicas) {
        return TopicBuilder.name(Topicos.SOLICITADA)
                .partitions(12)
                .replicas(replicas)
                .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, String.valueOf(Math.min(2, replicas)))
                .config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(Duration.ofDays(7).toMillis()))
                .build();
    }
}

Warning on Partitions: Never dynamically increase partition counts on a live topic that relies on keyed ordering. Changing the partition count invalidates the MurmurHash2 mapping, causing new records for an existing key to route to a different partition than historical records!

5. Service 2: `antifraude-service` and Exactly-Once inside Kafka

The antifraude-service performs a pure stream transformation: it reads from transferencias.solicitada, evaluates risk, and writes to either transferencias.analisada or transferencias.rejeitada.

Because both input and output reside within Kafka, we can leverage Kafka Transactions (Exactly-Once Semantics - EOS).

How Kafka Transactions Work

Kafka Transactions bind three distributed operations into a single atomic commit:

Producer Idempotence: eliminates client-side retransmit duplicates.
Transaction Coordinator: an internal state machine managing atomic commits across multiple output topic-partitions.
Consumer Offset Commits inside the Transaction: the consumer offset is committed directly via the transaction coordinator, not written uncoordinated to __consumer_offsets.
Consumer Isolation (isolation-level: read_committed): downstream consumers configured with read_committed buffer incoming records and only expose them to the application once the transaction commit marker is written to the partition log.

# application.yml (antifraude-service)
spring:
  application:
    name: antifraude-service
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    producer:
      acks: all
      transaction-id-prefix: tx-antifraude-${HOSTNAME:local}-
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: antifraude
      enable-auto-commit: false
      auto-offset-reset: earliest
      isolation-level: read_committed
      properties:
        group.protocol: consumer

The Transactional Listener

package com.exemplo.pagamentos.antifraude;

import com.exemplo.pagamentos.antifraude.eventos.TransferenciaAnalisada;
import com.exemplo.pagamentos.antifraude.eventos.TransferenciaRejeitada;
import com.exemplo.pagamentos.antifraude.eventos.TransferenciaSolicitada;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import tools.jackson.databind.json.JsonMapper;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.UUID;

@Component
public class AnaliseAntifraudeListener {

    private static final int LIMITE_BLOQUEIO = 80;

    private final KafkaTemplate<String, String> kafka;
    private final JsonMapper jsonMapper;
    private final MotorDeRisco motor;

    public AnaliseAntifraudeListener(KafkaTemplate<String, String> kafka,
                                     JsonMapper jsonMapper, MotorDeRisco motor) {
        this.kafka = kafka;
        this.jsonMapper = jsonMapper;
        this.motor = motor;
    }

    @KafkaListener(topics = Topicos.SOLICITADA, groupId = "antifraude")
    public void analisar(ConsumerRecord<String, String> registro) {
        TransferenciaSolicitada t = jsonMapper.readValue(registro.value(), TransferenciaSolicitada.class);
        int score = motor.calcularRisco(t);

        if (score >= LIMITE_BLOQUEIO) {
            var rejeitada = new TransferenciaRejeitada(
                    idDerivado("rejeitada-antifraude", t.transferenciaId()),
                    t.transferenciaId(), "BLOQUEADA_ANTIFRAUDE", Instant.now());
            kafka.send(Topicos.REJEITADA, t.contaOrigem(), jsonMapper.writeValueAsString(rejeitada));
            return;
        }

        var analisada = new TransferenciaAnalisada(
                idDerivado("analisada", t.transferenciaId()),
                t.transferenciaId(), t.contaOrigem(), t.contaDestino(),
                t.valorEmCentavos(), t.moeda(), score, Instant.now());
        kafka.send(Topicos.ANALISADA, t.contaOrigem(), jsonMapper.writeValueAsString(analisada));
    }

    private static UUID idDerivado(String tipo, UUID transferenciaId) {
        return UUID.nameUUIDFromBytes((tipo + ":" + transferenciaId).getBytes(StandardCharsets.UTF_8));
    }
}

Notice the idDerivado pattern: rather than generating a random UUID for the downstream event, we compute a deterministic UUID derived from the original business transaction ID. If the outbox relay previously resent a duplicate record, the output event will carry the exact same UUID, enabling downstream consumers to detect duplicates effortlessly.

6. Service 3: `ledger-service` and the Idempotent Consumer

The ledger-service is the heart of the system: it mutates monetary account balances.

In distributed systems, network partitions, consumer restarts, or broker rebalances mean messages will eventually be redelivered. Therefore, correctness cannot depend on the broker alone. The consumer itself must enforce strict idempotence.

Database Schema: Double-Entry Bookkeeping

-- V1__schema.sql (ledger-service)
CREATE TABLE conta (
    id              VARCHAR(32) PRIMARY KEY,
    saldo_centavos  BIGINT   NOT NULL CHECK (saldo_centavos >= 0),
    moeda           CHAR(3)  NOT NULL
);

CREATE TABLE lancamento (
    id                BIGSERIAL PRIMARY KEY,
    transferencia_id  UUID         NOT NULL,
    conta_id          VARCHAR(32)  NOT NULL REFERENCES conta (id),
    tipo              CHAR(1)      NOT NULL CHECK (tipo IN ('D', 'C')),
    valor_centavos    BIGINT       NOT NULL CHECK (valor_centavos > 0),
    criado_em         TIMESTAMPTZ  NOT NULL DEFAULT now()
);

CREATE TABLE transferencia_processada (
    transferencia_id  UUID PRIMARY KEY,
    resultado         VARCHAR(20)  NOT NULL,
    processada_em     TIMESTAMPTZ  NOT NULL DEFAULT now()
);

CREATE TABLE outbox_evento (
    id            UUID PRIMARY KEY,
    topico        VARCHAR(255) NOT NULL,
    chave         VARCHAR(64)  NOT NULL,
    tipo          VARCHAR(100) NOT NULL,
    payload       TEXT         NOT NULL,
    criado_em     TIMESTAMPTZ  NOT NULL DEFAULT now(),
    publicado_em  TIMESTAMPTZ
);

CREATE INDEX idx_outbox_pendentes ON outbox_evento (criado_em) WHERE publicado_em IS NULL;

Key aspects of this financial ledger:

Every transfer produces two lancamento rows: one debit (D) and one credit (C). Any account balance can be verified by summing all historical ledger entries.
CHECK (saldo_centavos >= 0) enforces that accounts cannot go negative at the database level.
transferencia_processada uses the business transferencia_id as its primary key. An idempotency key must always be a business identifier, never an arbitrary message ID or offset.

Idempotent Settlement Service

package com.exemplo.pagamentos.ledger.aplicacao;

import com.exemplo.pagamentos.ledger.eventos.TransferenciaAnalisada;
import com.exemplo.pagamentos.ledger.eventos.TransferenciaLiquidada;
import com.exemplo.pagamentos.ledger.eventos.TransferenciaRejeitada;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.UUID;

@Service
public class LiquidacaoService {

    private final JdbcClient jdbc;
    private final PublicadorDeEventos publicador;

    public LiquidacaoService(JdbcClient jdbc, PublicadorDeEventos publicador) {
        this.jdbc = jdbc;
        this.publicador = publicador;
    }

    @Transactional
    public ResultadoLiquidacao liquidar(TransferenciaAnalisada t) {
        // 1. Idempotência: Atomic INSERT com ON CONFLICT DO NOTHING
        int novas = jdbc.sql("""
                INSERT INTO transferencia_processada (transferencia_id, resultado)
                VALUES (:id, 'EM_PROCESSAMENTO')
                ON CONFLICT (transferencia_id) DO NOTHING
                """)
                .param("id", t.transferenciaId())
                .update();

        if (novas == 0) {
            return new ResultadoLiquidacao.Duplicada(t.transferenciaId());
        }

        // 2. Trava as contas em ordem determinística (evita deadlock cruzado)
        List<Conta> contas = jdbc.sql("""
                SELECT id, saldo_centavos, moeda FROM conta
                WHERE id IN (:ids)
                ORDER BY id
                FOR UPDATE
                """)
                .param("ids", List.of(t.contaOrigem(), t.contaDestino()))
                .query(Conta.class)
                .list();

        Conta origem = contas.stream().filter(c -> c.id().equals(t.contaOrigem())).findFirst().orElse(null);
        Conta destino = contas.stream().filter(c -> c.id().equals(t.contaDestino())).findFirst().orElse(null);

        // 3. Validação de Regras de Negócio (Rejeição é resultado normal, não exceção técnica)
        if (origem == null || destino == null) {
            return rejeitar(t, "CONTA_INEXISTENTE");
        }
        if (!origem.moeda().equals(t.moeda()) || !destino.moeda().equals(t.moeda())) {
            return rejeitar(t, "MOEDA_INCOMPATIVEL");
        }
        if (origem.saldoCentavos() < t.valorEmCentavos()) {
            return rejeitar(t, "SALDO_INSUFICIENTE");
        }

        // 4. Lançamento em partidas dobradas
        lancar(t.transferenciaId(), origem.id(), "D", -t.valorEmCentavos(), t.valorEmCentavos());
        lancar(t.transferenciaId(), destino.id(), "C", t.valorEmCentavos(), t.valorEmCentavos());

        // 5. Publica evento no Outbox na mesma transação local
        var evento = new TransferenciaLiquidada(UUID.randomUUID(), t.transferenciaId(), Instant.now());
        publicador.publicar(Topicos.LIQUIDADA, t.contaOrigem(), "TransferenciaLiquidada",
                evento.eventoId(), evento);
        marcarResultado(t.transferenciaId(), "LIQUIDADA");

        return new ResultadoLiquidacao.Liquidada(t.transferenciaId());
    }

    private void lancar(UUID transferenciaId, String contaId, String tipo, long delta, long valor) {
        jdbc.sql("UPDATE conta SET saldo_centavos = saldo_centavos + :delta WHERE id = :id")
                .param("delta", delta)
                .param("id", contaId)
                .update();

        jdbc.sql("""
                INSERT INTO lancamento (transferencia_id, conta_id, tipo, valor_centavos)
                VALUES (:transferenciaId, :contaId, :tipo, :valor)
                """)
                .param("transferenciaId", transferenciaId)
                .param("contaId", contaId)
                .param("tipo", tipo)
                .param("valor", valor)
                .update();
    }

    private ResultadoLiquidacao rejeitar(TransferenciaAnalisada t, String motivo) {
        var evento = new TransferenciaRejeitada(UUID.randomUUID(), t.transferenciaId(), motivo, Instant.now());
        publicador.publicar(Topicos.REJEITADA, t.contaOrigem(), "TransferenciaRejeitada",
                evento.eventoId(), evento);
        marcarResultado(t.transferenciaId(), "REJEITADA");
        return new ResultadoLiquidacao.Rejeitada(t.transferenciaId(), motivo);
    }

    private void marcarResultado(UUID transferenciaId, String resultado) {
        jdbc.sql("UPDATE transferencia_processada SET resultado = :resultado WHERE transferencia_id = :id")
                .param("resultado", resultado)
                .param("id", transferenciaId)
                .update();
    }

    record Conta(String id, long saldoCentavos, String moeda) {}
}

Critical architectural decisions in this method:

INSERT ... ON CONFLICT DO NOTHING: If two identical events arrive concurrently across different partitions or threads, PostgreSQL locks the primary key index row. The first transaction commits; the second detects novas == 0 and safely bails out without modifying balances.
ORDER BY id FOR UPDATE: Prevents classic distributed database deadlocks. If Transaction 1 transfers from A to B while Transaction 2 transfers from B to A, locking rows in ascending alphabetical order (id) guarantees both transactions acquire locks in the exact same sequence.
Business Rejections are Not Technical Exceptions: Insufficient balance is a valid business outcome. Rolling back the transaction or throwing an unhandled exception would cause Kafka to retry the message forever. Instead, we commit the rejection state and publish an event to transferencias.rejeitada.

Handling Consumer Failures: Blocking Retries and DLT

In consumer processing, errors fall into three categories:

Error Category	Example	Appropriate Action
Transient Error	Database momentary connection loss	Retry with exponential backoff
Poison Pill	Malformed JSON, schema violation	Send immediately to DLT (non-retryable)
Business Outcome	Insufficient account balance	Normal commit; publish rejection event

In the ledger, preserving account ordering is paramount. Therefore, we use blocking retries: the consumer pauses and retries the current record up to 5 times. If it still fails, the record is forwarded to a Dead Letter Topic (-dlt) with failure diagnostic headers attached:

package com.exemplo.pagamentos.ledger.infra.kafka;

import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
import tools.jackson.core.JacksonException;

@Configuration
class TratamentoDeErrosConfig {

    @Bean
    DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> template) {
        var recuperador = new DeadLetterPublishingRecoverer(template,
                (registro, excecao) -> new TopicPartition(registro.topic() + "-dlt", registro.partition()));

        var espera = new ExponentialBackOffWithMaxRetries(5);
        espera.setInitialInterval(500L);
        espera.setMultiplier(2.0);
        espera.setMaxInterval(15_000L);

        var tratador = new DefaultErrorHandler(recuperador, espera);
        tratador.addNotRetryableExceptions(JacksonException.class, IllegalArgumentException.class);
        return tratador;
    }
}

7. Service 4: `notificacoes-service` and Non-Blocking Retries

Unlike the financial ledger, external notifications (emails, SMS) do not require strict chronological serialization. If sending an email to Customer A fails due to a third-party gateway timeout, processing notifications for Customer B should not be blocked.

Here, Non-Blocking Retry Topics are the ideal architecture:

[ Topic: transferencias.liquidada ]
                 │
                 ▼
         (Attempt 1 Fails)
                 │
                 ▼
[ Topic: transferencias.liquidada-retry-1000 ] (Delayed by 1s)
                 │
                 ▼
         (Attempt 2 Fails)
                 │
                 ▼
[ Topic: transferencias.liquidada-retry-2000 ] (Delayed by 2s)
                 │
                 ▼
         (Attempt 3 Fails)
                 │
                 ▼
[ Topic: transferencias.liquidada-dlt ]        (Dead Letter Topic)

Spring Kafka configures this entire topology automatically using RetryTopicConfiguration:

package com.exemplo.pagamentos.notificacoes;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
import org.springframework.kafka.retrytopic.RetryTopicConfigurationBuilder;
import tools.jackson.core.JacksonException;

import java.util.List;

@Configuration
class RetryNotificacoesConfig {

    @Bean
    RetryTopicConfiguration retryNotificacoes(KafkaTemplate<String, String> template) {
        return RetryTopicConfigurationBuilder.newInstance()
                .exponentialBackoff(1_000, 2, 30_000)
                .maxAttempts(4)
                .includeTopics(List.of(Topicos.LIQUIDADA, Topicos.REJEITADA))
                .notRetryOn(JacksonException.class)
                .create(template);
    }
}

Apache Kafka 4.2 declared Share Groups (KIP-932) production-ready.

Historically, scaling Kafka consumers was strictly constrained by partition count: 12 partitions meant a hard ceiling of 12 active consumer threads. Share Groups introduce native queue semantics on top of Kafka log partitions: multiple consumers within a share group can pull records from the same partition concurrently, acknowledging individual records via ShareAckMode without blocking the offset progression of other consumers.

This is ideal for compute-heavy worker workloads and email notifications where message ordering is irrelevant. For ordered financial event streams, standard consumer groups remain the gold standard.

9. Observability and Monitoring

In an event-driven architecture, silent failures become production emergencies. With Spring Boot Actuator, Micrometer, and OpenTelemetry, tracing contexts are automatically injected into Kafka record headers, allowing you to follow a single transaction across all microservices in Grafana Tempo or Jaeger.

Key production metrics to monitor:

Consumer Group Lag: The delta between the latest produced offset and the committed consumer offset. Growing lag indicates downstream bottlenecks or halted consumers.
DLT Message Ingestion Rate: Any message routed to a -dlt topic must trigger an immediate operational alert.
Outbox Backlog Gauge: Monitors the count of unpublished records in the database outbox table. If the relay halts, Kafka metrics remain calm while business events quietly stall.

@Component
class MetricasOutbox {

    MetricasOutbox(MeterRegistry registry, JdbcClient jdbc) {
        Gauge.builder("outbox.pendentes", () ->
                        jdbc.sql("SELECT count(*) FROM outbox_evento WHERE publicado_em IS NULL")
                                .query(Long.class).single())
                .description("Pending events awaiting Kafka publication")
                .register(registry);
    }
}

10. Integration Testing with Testcontainers

Never rely on mocks to validate distributed messaging logic. Mocks will not catch serialization glitches, partition rebalance deadlocks, or transaction rollback edge cases.

With Testcontainers 2.x, we spin up genuine Apache Kafka and PostgreSQL containers inside JUnit 5 tests in milliseconds:

package com.exemplo.pagamentos.ledger;

import com.exemplo.pagamentos.ledger.eventos.TransferenciaAnalisada;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.kafka.KafkaContainer;
import org.testcontainers.postgresql.PostgreSQLContainer;
import org.testcontainers.utility.DockerImageName;
import tools.jackson.databind.json.JsonMapper;

import java.time.Duration;
import java.time.Instant;
import java.util.UUID;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

@SpringBootTest
@Testcontainers
class LiquidacaoIdempotenciaIT {

    @Container
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("apache/kafka:4.3.1"));

    @Container
    static PostgreSQLContainer postgres =
            new PostgreSQLContainer(DockerImageName.parse("postgres:18-alpine"));

    @DynamicPropertySource
    static void propriedades(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("spring.datasource.username", postgres::getUsername);
        registry.add("spring.datasource.password", postgres::getPassword);
        registry.add("app.kafka.replicas", () -> "1");
    }

    @Autowired KafkaTemplate<String, String> kafkaTemplate;
    @Autowired JdbcClient jdbc;
    @Autowired JsonMapper jsonMapper;

    @Test
    void mesmaTransferenciaEntregueDuasVezesDebitaUmaUnicaVez() {
        jdbc.sql("INSERT INTO conta (id, saldo_centavos, moeda) VALUES ('CC-001', 100000, 'BRL')").update();
        jdbc.sql("INSERT INTO conta (id, saldo_centavos, moeda) VALUES ('CC-002', 0, 'BRL')").update();

        var evento = new TransferenciaAnalisada(
                UUID.randomUUID(), UUID.randomUUID(), "CC-001", "CC-002",
                30_000, "BRL", 10, Instant.now());
        String json = jsonMapper.writeValueAsString(evento);

        // Simula entrega duplicada típica de cenários at-least-once
        kafkaTemplate.send(Topicos.ANALISADA, "CC-001", json);
        kafkaTemplate.send(Topicos.ANALISADA, "CC-001", json);

        await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
            long saldoOrigem = saldo("CC-001");
            long saldoDestino = saldo("CC-002");
            long lancamentos = jdbc.sql("SELECT count(*) FROM lancamento").query(Long.class).single();

            assertThat(saldoOrigem).isEqualTo(70_000);
            assertThat(saldoDestino).isEqualTo(30_000);
            assertThat(lancamentos).isEqualTo(2); // Exatamente um débito e um crédito
        });
    }

    private long saldo(String conta) {
        return jdbc.sql("SELECT saldo_centavos FROM conta WHERE id = :id")
                .param("id", conta)
                .query(Long.class)
                .single();
    }
}

11. Production Pitfalls Checklist

Before deploying your Kafka workloads to production, review this critical checklist:

Default Error Handler: Avoid using default unconfigured error handlers, which drop messages after several failed attempts without routing to a DLT.
Auto-Commit: Never enable enable-auto-commit: true when consumers perform stateful operations or external side effects.
Dual Writes: Never save to a database and publish to Kafka in separate uncoordinated statements without an outbox.
Idempotency Keys: Never use message offsets or random UUIDs as idempotency keys; always anchor to a business identifier.
Auto Offset Reset: Avoid auto.offset.reset: latest for financial systems; missing messages during initial deployments can cause severe accounting discrepancies.
Partition Resizing: Never scale partition counts on topics that require keyed ordering without executing an orderly migration strategy.
Unbounded Retries: Avoid infinite retry loops that freeze partition processing and mask root issues.

12. Conclusion

Building resilient distributed systems with Apache Kafka does not rely on a single magical configuration flag.

Durability comes from acks=all combined with min.insync.replicas=2.
Zero data loss between databases and brokers comes from the Transactional Outbox pattern.
Protection against duplicate executions comes from idempotent consumers anchored on immutable business keys.
Sequential account integrity comes from partition keys and deterministic database row locking.
Operational survivability comes from deliberate error handling, Dead Letter Topics, and end-to-end distributed tracing.

Kafka is an outstanding engine for propagating business facts across distributed boundaries, but it does not absolve engineers from architecting for failure. Ask yourself this question when designing your next pipeline: if any message in your system is delivered twice tomorrow, what happens to the money? If your answer relies on good luck, your architecture is not yet complete.