Apache Kafka con Java: Arquitectura, Idempotencia y Resiliencia de Extremo a Extremo

¿Qué ocurre realmente cuando tu aplicación ejecuta kafkaTemplate.send(...)?

Si tu respuesta inmediata es “el mensaje llega a una cola y un worker lo procesa”, estás operando Apache Kafka bajo el modelo mental de un broker tradicional como RabbitMQ o AWS SQS. Kafka es algo fundamentalmente distinto: es un registro distribuido de confirmaciones (commit log), inmutable y de solo anexado (append-only).

Esta diferencia cambia por completo las reglas: cómo particionas la carga, cómo escalan los consumidores, cómo garantizas el orden de entrega y, lo más crítico de todo, cómo evitas procesar eventos por duplicado o perder datos cuando la red falla o los nodos se reinician.

Este artículo es la primera parte de una serie técnica profunda sobre Arquitectura Dirigida por Eventos (EDA) y sistemas de mensajería empresarial. Construiremos un flujo financiero transaccional de extremo a extremo utilizando Java 25 (LTS), Spring Boot 4.1 y Apache Kafka 4.3 (modo KRaft). Implementaremos:

El patrón Transactional Outbox: integrando bases de datos relacionales y Kafka sin transacciones distribuidas (XA) ni two-phase commit.
Transacciones de Kafka y Exactly-Once Semantics (EOS): consistencia atómica de lectura-proceso-escritura dentro del broker.
Consumidores Idempotentes: contabilidad por partida doble y bloqueo determinista de filas para tolerar reentregas bajo la semántica at-least-once.
Estrategias de Resiliencia ante Errores: reintentos exponenciales bloqueantes con Dead Letter Topics (DLT) para flujos estrictamente ordenados, y tópicos de reintento no bloqueantes para tareas desacopladas.
Observabilidad y Pruebas Reales: métricas con Micrometer, trazabilidad con OpenTelemetry y pruebas de integración con Testcontainers 2.x.

1. El modelo mental: por qué Kafka no es RabbitMQ ni SQS

Antes de escribir configuraciones o código, debemos entender la mecánica subyacente.

Cola Tradicional (RabbitMQ / SQS):
[ Productor ] ──► [ Cola ] ──► El mensaje se consume y se ELIMINA del broker
                           └──► Múltiples workers compiten por el MISMO mensaje

Log Distribuido de Kafka:
[ Productor ] ──► [ Partición del Log: [0][1][2][3][4][5]... ] (Nunca se borra al leer)
                           ▲                  ▲
                           │                  │
                    Consumidor A (Offset 2) Consumidor B (Offset 5)

En los sistemas de colas tradicionales:

El broker gestiona el estado: sabe qué consumidor recibió qué mensaje, espera una confirmación y elimina u oculta el mensaje confirmado.
Los mensajes desaparecen una vez leídos.
Escalar el consumo consiste simplemente en añadir más hilos o instancias leyendo de la misma cola.

En Apache Kafka:

El broker es un log pasivo: los mensajes se anexan al final de un archivo en disco y se retienen por tiempo o tamaño (por ejemplo, 7 días), sin importar cuántas veces se hayan leído.
El consumidor gestiona el estado: un consumidor solo mantiene un puntero numérico llamado offset. Leer un evento no modifica ni borra nada en el broker. Un consumidor puede retroceder su offset para reprocesar eventos de hace tres días o inicializar un microservicio nuevo desde el principio del histórico.
El consumo escala estrictamente por particiones: dentro de un mismo grupo de consumidores, cada partición se asigna a lo sumo a un hilo consumidor. Si un tópico tiene 12 particiones, puedes tener hasta 12 consumidores concurrentes en ese grupo. Un decimotercer consumidor quedará inactivo.

Garantías de orden y particiones

Una verdad ineludible en sistemas distribuidos: Kafka solo garantiza el orden de los mensajes dentro de una misma partición. No existe orden global a nivel de todo el tópico.

Cuando un productor publica un registro:

Si se proporciona una clave (key), Kafka calcula un hash (MurmurHash2): $$\text{partición} = \text{murmur2}(\text{clave}) \pmod{\text{número de particiones}}$$ Todos los eventos con la misma clave van a parar siempre a la misma partición y en el mismo orden cronológico.
Si no se proporciona clave, el productor agrupa registros y los distribuye entre particiones por turnos (round-robin) para maximizar el rendimiento de disco y red.

En un sistema financiero, asignar una clave adecuada (como cuentaId o transferenciaId) es indispensable para garantizar el orden de las operaciones.

2. Conceptos clave: Brokers, KRaft y Consumer Groups

El quórum de metadatos KRaft (sin ZooKeeper)

A partir de Apache Kafka 4.x, ZooKeeper ha sido eliminado por completo en favor de KRaft (Kafka Raft Metadata Mode). Los brokers eligen un líder entre un grupo de controladores utilizando el algoritmo de consenso Raft. Cualquier cambio de metadatos (creación de tópicos, reasignación de particiones) se confirma directamente en un tópico interno replicado entre controladores. Esto permite recuperaciones de fallos en milisegundos y clústeres capaces de gestionar millones de particiones sin dependencias externas.

Réplicas, Líderes e In-Sync Replicas (ISR)

Cada partición posee:

Un Broker Líder: atiende todas las lecturas y escritas para esa partición.
Cero o más Brokers Seguidores (Followers): obtienen continuamente los segmentos de log del líder para mantener sus copias locales sincronizadas.

El conjunto de seguidores que están al día con el líder dentro de un intervalo máximo (replica.lag.time.max.ms) conforma el grupo ISR (In-Sync Replicas).

Tópico: transferencias.solicitada (Partición 0)
┌────────────────────────────────────────────────────────┐
│ Broker 1 (LÍDER)      │ [Registro 0] [Registro 1] [Registro 2]│
├───────────────────────┼────────────────────────────────┤
│ Broker 2 (ISR Seguidor)│ [Registro 0] [Registro 1] [Registro 2]│
├───────────────────────┼────────────────────────────────┤
│ Broker 3 (Atrasado)   │ [Registro 0]                   │ (Fuera del ISR)
└────────────────────────────────────────────────────────┘

Cuando un productor envía con acks=all, el broker solo confirma la recepción al cliente cuando todas las réplicas del ISR han escrito el registro en disco. Combinado con min.insync.replicas=2, si el ISR baja de 2 nodos, el líder rechaza nuevas escrituras, evitando la pérdida silenciosa de información.

Grupos de Consumidores y KIP-848

Un Consumer Group representa un suscriptor lógico. Cuando múltiples instancias se unen al mismo grupo, las particiones de los tópicos suscritos se reparten equitativamente entre ellas.

Kafka 4.x incorpora el protocolo de rebalanceo KIP-848 Next Generation Consumer Rebalance Protocol (group.protocol: consumer). En lugar del protocolo antiguo del cliente —que requería pausas globales sincronizadas que detenían temporalmente el consumo de todo el grupo—, KIP-848 delega la asignación al Group Coordinator del broker. Las asignaciones se transfieren incrementalmente mediante heartbeats, reduciendo los tiempos de rebalanceo de decenas de segundos a pocos milisegundos.

3. El escenario de arquitectura: flujo de transferencias financieras

Diseñaremos un flujo de procesamiento financiero donde perder un mensaje significa dinero desaparecido y procesarlo dos veces significa cobrarle doble a un cliente.

                        ┌──────────────────────────────────────────────────┐
                        │             transferencias-service               │
                        │  (REST API -> PostgreSQL -> Transactional Outbox)│
                        └─────────────────────────┬────────────────────────┘
                                                  │
                                                  ▼
                                 [ Tópico: transferencias.solicitada ]
                                                  │
                                                  ▼
                        ┌──────────────────────────────────────────────────┐
                        │               antifraude-service                 │
                        │        (Lee -> Evalúa -> Transacción Kafka)      │
                        └─────────────┬──────────────────────────┬─────────┘
                                      │                          │
                 (Si Riesgo < 80)     │                          │ (Si Riesgo >= 80)
                                      ▼                          ▼
               [ Tópico: transferencias.analisada ]   [ Tópico: transferencias.rejeitada ]
                                      │                          │
                                      ▼                          ▼
                        ┌──────────────────────────┐   ┌──────────────────────────┐
                        │      ledger-service      │   │   notificacoes-service   │
                        │ (Liquidación Idempotente │   │ (Tópicos de reintento    │
                        │    Partidas Dobles)      │   │    no bloqueantes)       │
                        └─────────────┬────────────┘   └──────────────────────────┘
                                      │
                                      ▼
                        [ Tópico: transferencias.liquidada ]
                                      │
                                      ▼
                        ┌──────────────────────────────────────────────────┐
                        │  transferencias-service (Proyección de Estado)   │
                        │  notificacoes-service (Push a Clientes)          │
                        └──────────────────────────────────────────────────┘

4. Servicio 1: `transferencias-service` y el patrón Transactional Outbox

El problema de la doble escritura (Dual-Write)

Un error frecuente en microservicios es la doble escritura:

// ANTIPATRÓN: NUNCA HAGAS ESTO EN PRODUCCIÓN
@Transactional
public void criarTransferencia(TransferenciaRequest req) {
    transferenciaRepository.save(new Transferencia(...)); // 1. Graba en Base de Datos
    kafkaTemplate.send("transferencias.solicitada", req); // 2. Publica en Kafka
}

¿Qué sucede si el paso 1 tiene éxito, pero el paso 2 falla por timeout de red? La base de datos tiene la transferencia, pero Kafka nunca recibe el evento. ¿Y si el paso 2 tiene éxito, pero la transacción de la base de datos falla al confirmar por una restricción de clave foránea? Kafka publicó un evento sobre una transferencia inexistente.

Las transacciones distribuidas (Two-Phase Commit / XA) entre base de datos y broker introducen bloqueos masivos, puntos únicos de fallo y escaso soporte.

La solución es el patrón Transactional Outbox:

[ Solicitud del Cliente ]
           │
           ▼
┌──────────────────────────────────────────────────────────────────┐
│  Transacción Local en Base de Datos (ACID)                       │
│                                                                  │
│  1. INSERT INTO transferencia (...)                              │
│  2. INSERT INTO outbox_evento (topico, chave, payload, criado_em)│
└──────────────────────────────────┬───────────────────────────────┘
                                   │ Confirmada (Commit)
                                   ▼
┌──────────────────────────────────────────────────────────────────┐
│  Outbox Relay (Proceso Programado o Debezium CDC)                │
│                                                                  │
│  SELECT ... FROM outbox_evento WHERE publicado_em IS NULL       │
│  FOR UPDATE SKIP LOCKED                                          │
│                                                                  │
│  kafkaTemplate.send(...) ──► [ Broker Kafka (acks=all) ]         │
│                                   │ ACK                          │
│  UPDATE outbox_evento SET publicado_em = now() WHERE id = ...    │
└──────────────────────────────────────────────────────────────────┘

Esquema del Outbox y Servicio de Negocio

-- 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;

Servicio de aplicación ejecutando ambas inserciones dentro de la misma transacción local:

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. Guarda la entidad de negocio
        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. Guarda el evento en la tabla outbox dentro de la misma transacción
        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;
    }
}

El publicador serializa el payload a JSON y lo guarda en 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();
    }
}

El Relay del Outbox

Para enviar los eventos pendientes sin colisiones entre múltiples instancias del servicio, usamos FOR UPDATE SKIP LOCKED. Cada instancia bloquea y publica un lote independiente sin interferir con las demás:

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("Publicación del outbox interrumpida", e);
        } catch (ExecutionException | TimeoutException e) {
            throw new IllegalStateException("Error al publicar lote del outbox", e);
        }
    }

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

Aspectos a considerar:

Cero Pérdida de Datos: La fila se marca como publicada únicamente después de que Kafka confirma la recepción. Si una instancia cae a la mitad, el bloqueo se libera y otra instancia retoma el lote.
Entrega At-Least-Once: Si el proceso se detiene tras la confirmación de Kafka pero antes del UPDATE outbox_evento, el evento se republicará. Los consumidores deben ser idempotentes.
Orden de Eventos: Si se ejecutan múltiples instancias de relay en paralelo con SKIP LOCKED, lotes de la misma cuenta podrían intercalarse en casos extremos. Si el orden estricto es crítico, utiliza una única instancia líder de relay o una solución CDC como Debezium.

Configuración del Productor para Máxima Durabilidad

# 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   # En producción: 3

acks: all: El líder solo confirma cuando todas las réplicas del ISR han escrito el registro.
enable.idempotence: true: Asigna identificadores y números de secuencia que evitan duplicados en el log del broker ante reintentos de red del cliente.
delivery.timeout.ms: 120000: Tiempo máximo total de intentos antes de fallar.
linger.ms: 5 y compression-type: zstd: Agrupación eficiente de mensajes que optimiza el ancho de banda y la E/S de disco.

Declaración de Tópicos

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();
    }
}

5. Servicio 2: `antifraude-service` y Exactly-Once dentro de Kafka

El antifraude-service lee de transferencias.solicitada, calcula el nivel de riesgo y escribe en transferencias.analisada o transferencias.rejeitada. Al ser un procesamiento exclusivo de Kafka a Kafka, podemos aplicar Transacciones de Kafka (Exactly-Once Semantics - EOS).

Funcionamiento de las Transacciones en Kafka

Las transacciones de Kafka coordinan atómicamente:

Idempotencia del Productor: evita duplicados por reintentos de red.
Coordinador de Transacciones: coordina el commit atómico entre tópicos de salida y particiones.
Commit del Offset dentro de la Transacción: el avance del offset de consumo se confirma dentro de la misma transacción de Kafka.
Aislamiento isolation-level: read_committed: los consumidores downstream solo leen registros de transacciones confirmadas.

# 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

Listener Transaccional

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));
    }
}

Observa idDerivado: en lugar de generar un UUID aleatorio, calculamos un UUID determinista a partir del ID de la transferencia original. Si el relay del outbox reenvía un evento duplicado, el evento derivado tendrá exactamente el mismo ID, permitiendo que los consumidores downstream detecten la duplicidad de inmediato.

6. Servicio 3: `ledger-service` y el Consumidor Idempotente

El ledger-service altera saldos monetarios reales. Bajo la semántica at-least-once, la red puede reentregar mensajes. La consistencia absoluta depende de que el consumidor sea estrictamente idempotente.

Esquema: Contabilidad por Partida Doble

-- 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;

Principios del diseño:

Cada transferencia genera dos registros en lancamento: un débito (D) y un crédito (C). El saldo se puede auditar sumando las partidas.
CHECK (saldo_centavos >= 0) previene saldos negativos a nivel de base de datos.
La tabla transferencia_processada utiliza transferencia_id como clave primaria. La clave de idempotencia debe ser un identificador de negocio, nunca un offset técnico.

Servicio de Liquidación Idempotente

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. Idempotencia: Inserción atómica con 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. Bloqueo de cuentas en orden determinista (evita deadlocks cruzados)
        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. Reglas de negocio (El rechazo es un resultado válido, no un error técnico)
        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. Asientos contables por partida doble
        lancar(t.transferenciaId(), origem.id(), "D", -t.valorEmCentavos(), t.valorEmCentavos());
        lancar(t.transferenciaId(), destino.id(), "C", t.valorEmCentavos(), t.valorEmCentavos());

        // 5. El evento de liquidación entra al Outbox en la misma transacción
        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) {}
}

Decisiones arquitectónicas:

INSERT ... ON CONFLICT DO NOTHING: Si dos entregas idénticas ocurren en paralelo, PostgreSQL bloquea el índice de la clave primaria. La primera confirma y la segunda detecta novas == 0, retornando inmediatamente sin alterar saldos.
ORDER BY id FOR UPDATE: Bloquear cuentas en orden alfabético estricto elimina los deadlocks entre transferencias inversas (A→B y B→A).
Los Rechazos de Negocio no son Excepciones: Un saldo insuficiente es un estado de negocio legítimo. Se confirma la transacción, se guarda el motivo y se emite el evento de rechazo al outbox.

Gestión de Fallos: Reintentos Bloqueantes y DLT

Tipo de Error	Ejemplo	Acción Correcta
Transitorio	Pérdida momentánea de conexión a la BD	Reintentar con espera exponencial
Veneno (Poison Pill)	JSON corrupto, campo obligatorio nulo	Enviar directo al DLT (sin reintentos)
Resultado de Negocio	Saldo insuficiente	Confirmación normal; publicar rechazo

Dado que el ledger exige preservar el orden por cuenta, aplicamos reintentos bloqueantes: se reintenta el mensaje hasta 5 veces. Si continúa fallando, se envía al Dead Letter Topic (-dlt):

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. Servicio 4: `notificacoes-service` y Reintentos No Bloqueantes

A diferencia de la contabilidad, el envío de notificaciones (emails, SMS) no requiere un ordenamiento estricto entre usuarios distintos. Si la notificación del Cliente A falla temporalmente, no debe bloquear el envío al Cliente B.

En este caso, la arquitectura adecuada son los Tópicos de Reintento No Bloqueantes:

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 declaró los Share Groups (KIP-932) listos para producción.

Tradicionalmente, la concurrencia de consumo estaba restringida por el número de particiones. Los Share Groups añaden semántica de cola clásica directamente sobre Kafka: múltiples consumidores de un mismo grupo pueden consumir de la misma partición en paralelo, confirmando mensajes de manera individual (ShareAckMode) sin bloquear a los demás consumidores. Son ideales para tareas pesadas de procesamiento asíncrono y notificaciones.

9. Observabilidad

En una arquitectura dirigida por eventos, lo que no se mide se convierte en un incidente invisible. Mediante Spring Boot Actuator, Micrometer y OpenTelemetry, los identificadores de trazabilidad se propagan en las cabeceras de los registros de Kafka.

Métricas operativas esenciales:

Lag de Consumidores: Desfase entre el último offset publicado y el confirmado. Un lag creciente evidencia cuellos de botella.
Mensajes en el DLT: Cualquier valor mayor a cero debe generar una alerta inmediata.
Acumulación en el Outbox: Mide los eventos pendientes de envío en la tabla de outbox.

@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("Eventos pendientes para publicar en Kafka")
                .register(registry);
    }
}

10. Pruebas de Integración con Testcontainers

Las pruebas con mocks no detectan errores de serialización, rebalanceos de particiones ni fallos de transacción.

Con Testcontainers 2.x, levantamos instancias reales de Kafka y PostgreSQL para validar la idempotencia de extremo a extremo:

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 la entrega duplicada habitual en escenarios 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); // Exactamente un débito y un 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. Lista de Verificación de Errores Comunes

Gestor de Errores por Defecto: Evita gestores no configurados que descartan mensajes fallidos sin enviarlos a un DLT.
Auto-Commit: No habilites enable-auto-commit: true en consumidores que realizan efectos secundarios persistentes.
Doble Escritura: Nunca guardes en la base de datos y publiques en Kafka en operaciones desconectadas sin un outbox.
Claves de Idempotencia: Usa siempre un identificador de negocio, nunca un offset técnico o un UUID generado aleatoriamente.
Auto Offset Reset: Evita latest en sistemas críticos donde perder eventos iniciales cause inconsistencias contables.
Redimensión de Particiones: No aumentes el número de particiones en tópicos con ordenamiento por clave sin planificar la migración.
Reintentos Infinitos: Evita bucles infinitos que congelen el consumo de la partición.

12. Conclusión

La resiliencia en Apache Kafka no depende de un único parámetro mágico:

La durabilidad proviene de acks=all junto con min.insync.replicas=2.
La ausencia de pérdida entre base de datos y broker la asegura el patrón Transactional Outbox.
La inmunidad ante ejecuciones duplicadas la garantizan consumidores idempotentes basados en identificadores de negocio inmutables.
El orden secuencial lo resuelven las claves de partición y los bloqueos deterministas de filas en la base de datos.
La capacidad operativa surge de un diseño intencionado de reintentos, DLTs y observabilidad distribuida.

Kafka es una herramienta formidable para comunicar eventos de negocio, pero exige asumir la responsabilidad de diseñar ante el fallo. Si tuvieras que llevarte una sola pregunta para reflexionar sobre tu arquitectura actual: si cualquier mensaje de tu sistema se entrega dos veces mañana, ¿qué ocurre exactamente con el dinero? Si la respuesta depende de la suerte, el diseño aún no está completo.