Apache Kafka com Java: Eventos, Consistência e Confiabilidade

Aplicação
   │
   │ publica
   ▼
Producer ──▶ Broker ──▶ Topic ──▶ Partition ──▶ Consumer
                    │                 │
                    │                 └── Offset
                    └── Réplicas / ISR

Tópico: pagamentos.transferencia.analisada.v1   (chave = conta de origem)

Partição 0: [CC-104 #0] [CC-104 #1] [CC-104 #2] ...
Partição 1: [CC-220 #0] [CC-220 #1] [CC-220 #2] ...
Partição 2: [CC-555 #0] [CC-042 #1] ...

Topic: transferencias

Partition 0 ──▶ Consumer A
Partition 1 ──▶ Consumer B
Partition 2 ──▶ Consumer C
Partition 3 ──▶ Consumer A

Grupo "ledger"           Grupo "notificacoes"
Partition 0 ─▶ A         Partition 0 ─▶ X
Partition 1 ─▶ B         Partition 1 ─▶ Y
Partition 2 ─▶ C         Partition 2 ─▶ Z

Partition 0

Broker 1  ── líder
Broker 2  ── réplica
Broker 3  ── réplica

ISR = [Broker 1, Broker 2, Broker 3]

Cliente ──POST /transferencias──▶ transferencias-service
                                        │  (Postgres: transferencia + outbox)
                                        ▼  relay do outbox
                          [pagamentos.transferencia.solicitada.v1]
                                        │
                                        ▼
                               antifraude-service        Kafka → Kafka, transacional
                                  │             │
                    (aprovada)    │             │  (bloqueada)
                                  ▼             │
             [pagamentos.transferencia.analisada.v1]
                                  │             │
                                  ▼             │
                            ledger-service      │        Postgres: contas, lançamentos, outbox
                             │          │       │
                  (liquidada)│          │(rejeitada)
                             ▼          ▼       ▼
        [pagamentos.transferencia.liquidada.v1]  [pagamentos.transferencia.rejeitada.v1]
                             │                          │
             ┌───────────────┴──────────────┬───────────┘
             ▼                              ▼
   notificacoes-service            transferencias-service
   (retry topics + DLT)            (projeção do status)

docker pull apache/kafka:4.3.1

docker run --name kafka -p 9092:9092 apache/kafka:4.3.1

# docker-compose.yml (desenvolvimento)
services:
  kafka:
    image: apache/kafka:4.3.1
    container_name: kafka
    ports:
      - "9092:9092"

  postgres:
    image: postgres:18-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: app
      POSTGRES_DB: transferencias
    ports:
      - "5432:5432"

docker compose up -d

docker compose ps

docker compose down

tar -xzf kafka_2.13-4.3.1.tgz
cd kafka_2.13-4.3.1

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t "$KAFKA_CLUSTER_ID" -c config/server.properties
bin/kafka-server-start.sh config/server.properties

docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --create \
  --topic pagamentos.transferencia.solicitada.v1 \
  --partitions 3 \
  --replication-factor 1

docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --list

docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --topic pagamentos.transferencia.solicitada.v1

docker compose exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka:9092 \
  --topic pagamentos.transferencia.solicitada.v1

docker compose exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka:9092 \
  --topic pagamentos.transferencia.solicitada.v1 \
  --property parse.key=true \
  --property key.separator=:

CC-104:{"transferenciaId":"001","valorCentavos":30000}
CC-104:{"transferenciaId":"002","valorCentavos":10000}
CC-220:{"transferenciaId":"003","valorCentavos":5000}

docker compose exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic pagamentos.transferencia.solicitada.v1 \
  --from-beginning

docker compose exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --list

docker compose exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --group ledger

docker compose exec kafka /opt/kafka/bin/kafka-configs.sh \
  --bootstrap-server kafka:9092 \
  --entity-type topics \
  --entity-name pagamentos.transferencia.solicitada.v1 \
  --alter \
  --add-config retention.ms=604800000

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.1.1</version>
    <relativePath/>
</parent>

<properties>
    <java.version>25</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webmvc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-flyway</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

package com.exemplo.pagamentos.transferencias.eventos;

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

public record TransferenciaSolicitada(
        UUID eventoId,
        UUID transferenciaId,
        String contaOrigem,
        String contaDestino,
        long valorEmCentavos,
        String moeda,
        Instant ocorridoEm) {}

public record TransferenciaAnalisada(
        UUID eventoId,
        UUID transferenciaId,
        String contaOrigem,
        String contaDestino,
        long valorEmCentavos,
        String moeda,
        int scoreRisco,
        Instant ocorridoEm) {}

public record TransferenciaLiquidada(UUID eventoId, UUID transferenciaId, Instant ocorridoEm) {}

public record TransferenciaRejeitada(
        UUID eventoId, UUID transferenciaId, String motivo, Instant ocorridoEm) {}

-- 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   VARCHAR(255),
    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;

package com.exemplo.pagamentos.transferencias.aplicacao;

import java.util.UUID;

public interface PublicadorDeEventos {

    void publicar(String topico, String chave, String tipo, UUID eventoId, Object evento);
}

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

import com.exemplo.pagamentos.transferencias.aplicacao.PublicadorDeEventos;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import tools.jackson.databind.json.JsonMapper;

import java.util.UUID;

@Component
class OutboxPublicador implements PublicadorDeEventos {

    private final JdbcClient jdbc;
    private final JsonMapper jsonMapper;

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

    @Override
    @Transactional(propagation = Propagation.MANDATORY)
    public void publicar(String topico, String chave, String tipo, UUID eventoId, Object evento) {
        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", jsonMapper.writeValueAsString(evento))
                .update();
    }
}

package com.exemplo.pagamentos.transferencias.aplicacao;

import com.exemplo.pagamentos.transferencias.eventos.TransferenciaSolicitada;
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.Optional;
import java.util.UUID;

@Service
public class TransferenciaService {

    private final JdbcClient jdbc;
    private final PublicadorDeEventos publicador;

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

    @Transactional
    public Transferencia solicitar(UUID id, String origem, String destino, long valor, String moeda) {
        if (origem.equals(destino)) {
            throw new TransferenciaInvalidaException("Conta de origem e destino devem ser diferentes");
        }

        int inseridas = jdbc.sql("""
                INSERT INTO transferencia (id, conta_origem, conta_destino, valor_centavos, moeda, status)
                VALUES (:id, :origem, :destino, :valor, :moeda, 'PENDENTE')
                ON CONFLICT (id) DO NOTHING
                """)
                .param("id", id)
                .param("origem", origem)
                .param("destino", destino)
                .param("valor", valor)
                .param("moeda", moeda)
                .update();

        if (inseridas == 1) {
            var evento = new TransferenciaSolicitada(
                    UUID.randomUUID(), id, origem, destino, valor, moeda, Instant.now());
            publicador.publicar(Topicos.SOLICITADA, origem, "TransferenciaSolicitada",
                    evento.eventoId(), evento);
        }

        return buscar(id).orElseThrow();
    }

    @Transactional(readOnly = true)
    public Optional<Transferencia> buscar(UUID id) {
        return jdbc.sql("""
                SELECT id, conta_origem, conta_destino, valor_centavos, moeda, status
                FROM transferencia WHERE id = :id
                """)
                .param("id", id)
                .query(Transferencia.class)
                .optional();
    }
}

public record Transferencia(
        UUID id, String contaOrigem, String contaDestino,
        long valorCentavos, String moeda, String status) {}

public final class Topicos {

    public static final String SOLICITADA = "pagamentos.transferencia.solicitada.v1";
    public static final String ANALISADA = "pagamentos.transferencia.analisada.v1";
    public static final String LIQUIDADA = "pagamentos.transferencia.liquidada.v1";
    public static final String REJEITADA = "pagamentos.transferencia.rejeitada.v1";

    private Topicos() {}
}

public class TransferenciaInvalidaException extends RuntimeException {

    public TransferenciaInvalidaException(String mensagem) {
        super(mensagem);
    }
}

package com.exemplo.pagamentos.transferencias.web;

@RestController
@RequestMapping("/transferencias")
public class TransferenciaController {

    private final TransferenciaService service;

    public TransferenciaController(TransferenciaService service) {
        this.service = service;
    }

    @PostMapping
    public ResponseEntity<TransferenciaResponse> solicitar(
            @RequestHeader("Idempotency-Key") UUID chaveIdempotencia,
            @RequestBody SolicitarTransferenciaRequest request) {

        Transferencia t = service.solicitar(chaveIdempotencia, request.contaOrigem(),
                request.contaDestino(), request.valorEmCentavos(), request.moeda());

        return ResponseEntity.accepted().body(TransferenciaResponse.de(t));
    }

    @GetMapping("/{id}")
    public ResponseEntity<TransferenciaResponse> consultar(@PathVariable UUID id) {
        return service.buscar(id)
                .map(t -> ResponseEntity.ok(TransferenciaResponse.de(t)))
                .orElseGet(() -> ResponseEntity.notFound().build());
    }

    public record SolicitarTransferenciaRequest(
            String contaOrigem, String contaDestino, long valorEmCentavos, String moeda) {}

    public record TransferenciaResponse(UUID id, String status) {

        static TransferenciaResponse de(Transferencia t) {
            return new TransferenciaResponse(t.id(), t.status());
        }
    }
}

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

# 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   # 3 em produção

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

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

import com.exemplo.pagamentos.transferencias.eventos.TransferenciaLiquidada;
import com.exemplo.pagamentos.transferencias.eventos.TransferenciaRejeitada;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import tools.jackson.databind.json.JsonMapper;

@Component
class ResultadoTransferenciaListener {

    private final JdbcClient jdbc;
    private final JsonMapper jsonMapper;

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

    @KafkaListener(topics = Topicos.LIQUIDADA, groupId = "transferencias-projecao")
    void aoLiquidar(String json) {
        var evento = jsonMapper.readValue(json, TransferenciaLiquidada.class);
        jdbc.sql("""
                UPDATE transferencia SET status = 'LIQUIDADA', atualizada_em = now()
                WHERE id = :id AND status = 'PENDENTE'
                """)
                .param("id", evento.transferenciaId())
                .update();
    }

    @KafkaListener(topics = Topicos.REJEITADA, groupId = "transferencias-projecao")
    void aoRejeitar(String json) {
        var evento = jsonMapper.readValue(json, TransferenciaRejeitada.class);
        jdbc.sql("""
                UPDATE transferencia
                SET status = 'REJEITADA', motivo_rejeicao = :motivo, atualizada_em = now()
                WHERE id = :id AND status = 'PENDENTE'
                """)
                .param("id", evento.transferenciaId())
                .param("motivo", evento.motivo())
                .update();
    }
}

# 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

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

@Component
public class MotorDeRisco {

    private static final long LIMITE_VALOR_ALTO_CENTAVOS = 10_000_000L; // R$ 100.000,00

    // Regra de exemplo; um motor real combinaria histórico, dispositivo, geografia etc.
    public int calcularRisco(TransferenciaSolicitada t) {
        return t.valorEmCentavos() >= LIMITE_VALOR_ALTO_CENTAVOS ? 95 : 10;
    }
}

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

package com.exemplo.pagamentos.ledger.aplicacao;

import java.util.UUID;

public sealed interface ResultadoLiquidacao {

    record Liquidada(UUID transferenciaId) implements ResultadoLiquidacao {}

    record Rejeitada(UUID transferenciaId, String motivo) implements ResultadoLiquidacao {}

    record Duplicada(UUID transferenciaId) implements ResultadoLiquidacao {}
}

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: só um processamento por transferência consegue inserir esta linha.
        //    Uma entrega concorrente da mesma transferência espera o commit deste e depois não insere nada.
        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 duas contas em ordem determinística (evita deadlock entre transferências cruzadas).
        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. Regras de negócio: rejeição é um resultado normal, não um erro.
        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. Partidas dobradas: débito na origem, crédito no destino.
        lancar(t.transferenciaId(), origem.id(), "D", -t.valorEmCentavos(), t.valorEmCentavos());
        lancar(t.transferenciaId(), destino.id(), "C", t.valorEmCentavos(), t.valorEmCentavos());

        // 5. O evento de resultado entra no outbox, na mesma transação dos lançamentos.
        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) {}
}

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

import com.exemplo.pagamentos.ledger.aplicacao.LiquidacaoService;
import com.exemplo.pagamentos.ledger.aplicacao.ResultadoLiquidacao;
import com.exemplo.pagamentos.ledger.eventos.TransferenciaAnalisada;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import tools.jackson.databind.json.JsonMapper;

@Component
class LiquidacaoListener {

    private static final Logger log = LoggerFactory.getLogger(LiquidacaoListener.class);

    private final LiquidacaoService service;
    private final JsonMapper jsonMapper;

    LiquidacaoListener(LiquidacaoService service, JsonMapper jsonMapper) {
        this.service = service;
        this.jsonMapper = jsonMapper;
    }

    @KafkaListener(topics = Topicos.ANALISADA, groupId = "ledger")
    void aoReceber(ConsumerRecord<String, String> registro) {
        TransferenciaAnalisada evento =
                jsonMapper.readValue(registro.value(), TransferenciaAnalisada.class);

        ResultadoLiquidacao resultado = service.liquidar(evento);

        switch (resultado) {
            case ResultadoLiquidacao.Liquidada l ->
                    log.info("Transferência {} liquidada (partição {}, offset {})",
                            l.transferenciaId(), registro.partition(), registro.offset());
            case ResultadoLiquidacao.Rejeitada r ->
                    log.info("Transferência {} rejeitada: {}", r.transferenciaId(), r.motivo());
            case ResultadoLiquidacao.Duplicada d ->
                    log.info("Transferência {} já processada; entrega duplicada ignorada", d.transferenciaId());
        }
    }
}

# application.yml (ledger-service)
spring:
  application:
    name: ledger-service
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        enable.idempotence: true
    consumer:
      group-id: ledger
      enable-auto-commit: false
      auto-offset-reset: earliest
      isolation-level: read_committed
      max-poll-records: 50
      properties:
        group.protocol: consumer
    listener:
      ack-mode: record
      concurrency: 3
      observation-enabled: true
    template:
      observation-enabled: true

app:
  kafka:
    replicas: 1   # 3 em produção

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

@Bean
NewTopic analisadaDlt(@Value("${app.kafka.replicas:3}") int replicas) {
    return TopicBuilder.name(Topicos.ANALISADA + "-dlt")
            .partitions(12)
            .replicas(replicas)
            .config(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, String.valueOf(Math.min(2, replicas)))
            .build();
}

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

import io.micrometer.core.instrument.MeterRegistry;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
class MonitorDeDeadLetter {

    private static final Logger log = LoggerFactory.getLogger(MonitorDeDeadLetter.class);

    private final MeterRegistry meterRegistry;

    MonitorDeDeadLetter(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    @KafkaListener(topics = Topicos.ANALISADA + "-dlt", groupId = "ledger-dlt-monitor")
    void registrar(ConsumerRecord<String, String> registro) {
        String causa = cabecalho(registro, KafkaHeaders.DLT_EXCEPTION_MESSAGE);
        String origem = cabecalho(registro, KafkaHeaders.DLT_ORIGINAL_TOPIC);

        meterRegistry.counter("ledger.dlt.mensagens", "topico_origem", origem).increment();
        log.error("Mensagem enviada ao DLT. chave={} origem={} causa={}", registro.key(), origem, causa);
    }

    private static String cabecalho(ConsumerRecord<String, String> registro, String nome) {
        Header header = registro.headers().lastHeader(nome);
        return header == null ? "desconhecido" : new String(header.value(), StandardCharsets.UTF_8);
    }
}

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

@Component
class NotificacaoListener {

    private final NotificacaoService notificacoes;
    private final JsonMapper jsonMapper;

    NotificacaoListener(NotificacaoService notificacoes, JsonMapper jsonMapper) {
        this.notificacoes = notificacoes;
        this.jsonMapper = jsonMapper;
    }

    @KafkaListener(topics = Topicos.LIQUIDADA, groupId = "notificacoes")
    void aoLiquidar(String json) {
        var evento = jsonMapper.readValue(json, TransferenciaLiquidada.class);
        notificacoes.notificarLiquidacao(evento.transferenciaId());
    }

    @KafkaListener(topics = Topicos.REJEITADA, groupId = "notificacoes")
    void aoRejeitar(String json) {
        var evento = jsonMapper.readValue(json, TransferenciaRejeitada.class);
        notificacoes.notificarRejeicao(evento.transferenciaId(), evento.motivo());
    }
}

@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 aguardando publicação no Kafka")
                .register(registry);
    }
}

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers-junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers-kafka</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers-postgresql</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

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 a entrega duplicada típica do 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);
        });
    }

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