Kafka Streams e Connect

Conceito

O Kafka em si só move bytes. As duas peças que transformam isso numa plataforma são Connect (integrar sem código) e Streams (processar com estado).

Parte 1 — Kafka Connect

Um serviço para mover dados entre Kafka e o resto do mundo, por configuração JSON em vez de código.

              ┌──────────── Kafka Connect (cluster de workers) ────────────┐
banco ──────→ │ SOURCE connector  →  tópico  →  SINK connector │ ──────→ S3
(Debezium)    └────────────────────────────────────────────────┘         Elastic

CDC com Debezium: a peça que fecha o circuito

Debezium é um source connector que lê o log de replicação do banco — o WAL lógico do Postgres, o binlog do MySQL, LogMiner/XStream no Oracle, o oplog/change stream do MongoDB — e publica cada mudança como evento.

{
  "name": "pedidos-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-primary",
    "database.dbname": "loja",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_pedidos",
    "table.include.list": "public.pedidos,public.outbox",
    "topic.prefix": "loja",
    "tombstones.on.delete": "true",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter"
  }
}

Isto resolve o outbox pattern de forma automática (4. Garantias de Entrega e Idempotência): a aplicação grava na tabela outbox na mesma transação do dado, e o Debezium publica no Kafka lendo o WAL. Nada de escrita dupla, nada de publicador próprio para manter.

Parte 2 — Kafka Streams

Uma biblioteca (não um cluster) para processar streams com estado dentro do seu próprio serviço. Sem Flink, sem Spark, sem YARN: é um .jar na sua aplicação, que escala adicionando instâncias.

A dualidade stream/table

O conceito central, e o mais elegante do Kafka:

Um stream é a derivada de uma tabela; uma tabela é a integral de um stream.

É a mesma ideia do log de replicação de um banco: aplicar todos os eventos em ordem reconstrói o estado. O Kafka expõe as duas visões da mesma coisa (compactação de log).

State store e changelog

Operações com estado (count, aggregate, join, janelas) guardam estado local em RocksDB, e cada state store tem um tópico de changelog no Kafka:

instância A                             Kafka
├── RocksDB (estado local, no disco) ──→ tópico-changelog (compactado)
└── se a instância morrer: outra instância RELÊ o changelog e reconstrói o estado

Isso dá tolerância a falha sem banco externo: o estado vive local (rápido) e é durável no Kafka (recuperável). O custo é o tempo de reconstrução após falha — mitigado por standby replicas (num.standby.replicas), que mantêm uma cópia quente.

Repartição: o custo escondido

Se você agrupa por uma chave diferente da chave da mensagem, o Streams precisa reparticionar — o que significa escrever num tópico interno e ler de volta:

stream.groupBy((k, v) -> v.getClienteId())   // chave nova!
      .count();
// → cria um tópico interno "app-KSTREAM-...-repartition": +1 write +1 read por evento

É correto (garante que a mesma chave caia na mesma partição) e é o principal custo oculto de uma topologia. Quando possível, produza já com a chave que você vai agregar (3. Partições, Offsets e Consumer Groups).

Tempo: event time vs processing time

event time      = quando o fato aconteceu (o timestamp no evento)
processing time  = quando o seu código viu

Streams trabalha por event time por padrão, com janelas e um período de tolerância (grace period) para eventos atrasados:

TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1))

Isso é o que permite resultados corretos e reproduzíveis num replay: reprocessar o histórico dá o mesmo resultado, porque as janelas são definidas pelo tempo do evento, não pelo relógio de execução.

Exactly-once

processing.guarantee=exactly_once_v2

Uma linha, e o Streams passa a usar transações do Kafka para ligar atomicamente consumir → atualizar estado → produzir. Vale porque tudo está dentro do Kafka — a ressalva de 4. Garantias de Entrega e Idempotência (efeito colateral externo não é coberto) continua valendo se você chamar uma API ou gravar num banco de dentro do processamento.

E se meu serviço é em Go?

Kafka Streams é Java/JVM apenas. Em Go, as opções são:

Para a maioria dos casos em Go, a resposta honesta é: consumidor idempotente + estado num banco (4. Garantias de Entrega e Idempotência, 5. Go em Produção). Streams só se paga quando há agregação com estado, janelas e joins de verdade.

Trade-offs

Connect elimina código e adiciona um cluster. Workers precisam ser dimensionados, monitorados e atualizados; um connector com má configuração de tasks.max fica ocioso ou sobrecarrega a origem. Em compensação, ele resolve retry, offset e paralelismo de forma testada — reescrever isso em cada serviço é pior.

Streams é simples de subir e sutil de operar. Escalar é adicionar instância (o grupo rebalanceia), mas cada instância carrega estado local: um deploy que reinicia tudo ao mesmo tempo dispara reconstrução em massa do changelog. Deploy rolling com num.standby.replicas ≥ 1 é requisito, não refinamento.

Schema Registry deixa de ser opcional aqui. Em CDC e streams, mensagens vivem por semanas e são reprocessadas. Sem contrato versionado com compatibilidade retroativa, um campo renomeado quebra o replay do histórico — e o histórico é justamente o ativo que você estava construindo.

O trade-off central: Connect e Streams trocam flexibilidade de código por infraestrutura pronta. Valem quando o problema é padrão (mover dado, agregar por chave, janelar) e viram um peso quando a lógica é específica — aí um consumidor comum, escrito por você, é mais fácil de entender e operar.

Exemplo prático

Contagem por janela, com exactly-once

Properties p = new Properties();
p.put(StreamsConfig.APPLICATION_ID_CONFIG, "pedidos-agregador");  // = consumer group
p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
p.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);   // failover sem reconstruir

StreamsBuilder b = new StreamsBuilder();

// receita por cliente, em janelas de 5 minutos (por EVENT TIME)
b.<String, Pedido>stream("pedidos.criados")
 .groupByKey()                                      // já vem com a chave certa: sem repartição
 .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5), Duration.ofMinutes(1)))
 .aggregate(
     () -> 0.0,
     (chave, pedido, total) -> total + pedido.getTotal(),
     Materialized.as("receita-por-cliente")          // state store + changelog
 )
 .toStream()
 .to("pedidos.receita-5min");

new KafkaStreams(b.build(), p).start();

Sink para S3 sem uma linha de código

{
  "name": "pedidos-to-s3",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "pedidos.criados",
    "tasks.max": "4",
    "s3.bucket.name": "data-lake-prod",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'dt'=YYYY-MM-dd/'hr'=HH",
    "partition.duration.ms": "3600000",
    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Monitoramento

# saúde dos connectors: o campo "state" precisa ser RUNNING (tasks incluídas!)
curl -s localhost:8083/connectors?expand=status | jq '
  to_entries[] | {name: .key,
                  connector: .value.status.connector.state,
                  tasks: [.value.status.tasks[].state]}'

# uma task FAILED não aparece no state do connector: veja o trace
curl -s localhost:8083/connectors/pedidos-cdc/status | jq '.tasks[].trace'

# reiniciar só a task que falhou
curl -X POST localhost:8083/connectors/pedidos-cdc/tasks/2/restart

Como isso falha em produção

Aprofundamento aplicado

Estado local precisa ser reconstruível

Uma agregação por cliente mantém estado. Quando a tarefa muda de instância, o novo processo precisa recuperar esse estado antes de continuar corretamente. Changelogs e mecanismos de recuperação participam desse ciclo; o tempo de restauração também faz parte do tempo de recuperação da aplicação.

Tempo do evento e tempo de chegada

Uma compra ocorreu às 10h, mas chegou às 10h15. Um relatório por horário da compra precisa decidir quanto atraso aceita e como trata eventos que chegam depois do fechamento da janela. Escolher uma janela é escolher uma regra de resultado, além de uma estrutura de processamento. Referência: conceitos de Streams.

Exercício: processe eventos de 10h02 e 10h04, depois receba um evento de 9h59. Resultado esperado: justifique se o evento altera uma janela antiga, é descartado ou segue outro caminho, conforme a regra configurada.

No Connect, descreva as garantias do conector específico e do sistema externo. Registrar offsets no framework não torna toda escrita no destino automaticamente atômica com esse registro. Confirme comportamento de retries, duplicatas e evolução de schema.

Relacionado

• 2. Kafka
• 3. Partições, Offsets e Consumer Groups
• 4. Garantias de Entrega e Idempotência
• 5. Armazenamento, ISR e KRaft
• 1. Mensageria e Desacoplamento
• 3. WAL, Checkpoint e Replicação
• 5. Replicação e Alta Disponibilidade

---

Parte de MOC Kafka · Kafka Streams (docs) · Kafka Connect · debezium.io