Apache Kafka

Kafka não é uma message queue — é um log distribuído e persistente. A distinção importa: em uma fila, mensagens são deletadas ao serem consumidas. No Kafka, mensagens ficam armazenadas por um período configurável (horas, dias, semanas) e qualquer consumidor pode ler qualquer offset do log a qualquer momento. Isso muda fundamentalmente o que é possível: múltiplos sistemas independentes consumindo o mesmo stream de eventos sem coordenação, replay de eventos históricos, e reprocessamento de dados de qualquer ponto no tempo.

Arquitetura fundamental

O log como abstração central

Um tópico Kafka é um log append-only particionado. Cada mensagem escrita recebe um offset — um número sequencial imutável dentro da partição. Consumidores leem o log a partir de um offset e avançam o cursor (commit do offset) explicitamente. O broker não rastreia o que foi "consumido" — apenas armazena o log. Isso inverte a responsabilidade: o consumidor controla sua posição no log, não o broker.

# Estrutura de um tópico com 3 partições
Topic: orders

Partition 0: [offset 0][offset 1][offset 2][offset 3]...  → replica em broker 1, 2
Partition 1: [offset 0][offset 1][offset 2]...            → replica em broker 2, 3
Partition 2: [offset 0][offset 1][offset 2][offset 3][offset 4]...  → replica em broker 1, 3

# Cada partição tem:
# - Um leader (recebe writes e reads)
# - N-1 followers (réplicas síncronas ou assíncronas)
# - ISR (In-Sync Replicas): conjunto de réplicas consideradas "atualizadas"

Partições e paralelismo

Partições são a unidade de paralelismo do Kafka. O número de partições de um tópico define o paralelismo máximo de consumo — você não pode ter mais consumers ativos em um grupo do que partições. Se um tópico tem 12 partições, o consumer group pode ter no máximo 12 consumers ativos; consumers adicionais ficam ociosos como standby.

Partition key e ordenação

Kafka garante ordenação apenas dentro de uma partição. Para garantir que eventos de um mesmo entidade (pedido, cliente) sejam processados em ordem, use a partition key — o broker mapeia a key para a mesma partição via hash consistente.

# Ordenação por customer_id: todos os eventos do mesmo cliente → mesma partição
producer.send("orders",
    key=customer_id.encode(),    # key determina a partição
    value=event_json.encode()
)

# Sem key → round-robin entre partições (sem garantia de ordem)
producer.send("metrics", value=metric_json.encode())

# Cuidado: partition key com distribuição ruim → hot partition
# Se 80% dos pedidos são de 10 customers → 80% do tráfego em 10 partições
# Solução: compor a key (customer_id + hour) ou usar consistent hashing externo

Consumer Groups

Consumer groups são o mecanismo de escalonamento do lado do consumidor. Cada grupo recebe todas as mensagens do tópico, mas cada partição é atribuída a exatamente um consumer dentro do grupo. Grupos diferentes consomem o tópico independentemente, do seu próprio offset — sem interferência entre si.

# Topic: orders (3 partições)
# Consumer Group A (serviço de inventário): 3 consumers
Consumer A-1 → lê Partition 0
Consumer A-2 → lê Partition 1
Consumer A-3 → lê Partition 2

# Consumer Group B (serviço de faturamento): 1 consumer
Consumer B-1 → lê Partition 0, 1, 2 (todas — consumer único no grupo)

# Consumer Group C (analytics): 5 consumers
Consumer C-1 → lê Partition 0
Consumer C-2 → lê Partition 1
Consumer C-3 → lê Partition 2
Consumer C-4, C-5 → ociosos (mais consumers que partições)

Rebalancing

Quando um consumer entra ou sai do grupo, o Kafka realoca partições entre os consumers ativos — chamado de rebalancing. Durante o rebalance, o processamento para temporariamente (stop-the-world). Em Kafka moderno (v2.4+), o cooperative rebalancing (incremental) minimiza o impacto: apenas as partições que precisam ser movidas são pausadas, não todas.

# Configurações críticas para rebalancing
group.id=inventory-service        # identificador do consumer group
session.timeout.ms=45000          # tempo até considerar consumer morto
heartbeat.interval.ms=3000        # frequência de heartbeat (deve ser < session.timeout/3)
max.poll.interval.ms=300000       # tempo máximo entre polls (aumentar para processamento lento)
partition.assignment.strategy=CooperativeStickyAssignor  # minimiza rebalancing

Retenção e compactação

Retenção por tempo e tamanho

Mensagens no Kafka expiram após um período configurável — não quando são consumidas. Isso é fundamental: você pode ter um consumer que processa tudo em 1 segundo e outro que processa em 1 hora, ambos no mesmo tópico, sem que um afete o outro.

# Configuração de retenção por tópico
kafka-configs --bootstrap-server kafka:9092 \
  --entity-type topics --entity-name orders \
  --alter \
  --add-config retention.ms=604800000  # 7 dias
              retention.bytes=10737418240  # 10GB por partição (o que vier primeiro)

Log compaction

Para tópicos onde o que importa é o estado atual de cada key (não o histórico completo), log compaction mantém apenas a mensagem mais recente por partition key. O Kafka remove mensagens antigas com a mesma key, mas garante que a última versão nunca é deletada. Perfeito para changelogs de banco de dados e tabelas de lookup.

# Tópico compactado — changelog de clientes
# Kafka mantém apenas o registro mais recente por customer_id
[customer-123: {name: "Ana", plan: "basic"}]     ← antigo, será removido
[customer-456: {name: "João", plan: "pro"}]
[customer-123: {name: "Ana Silva", plan: "pro"}] ← este fica (mais recente)

# Configurar compaction
kafka-configs --bootstrap-server kafka:9092 \
  --entity-type topics --entity-name customers-state \
  --alter --add-config cleanup.policy=compact

# Tombstone: publicar mensagem com value=null → deleta a key do log compactado
producer.send("customers-state", key="customer-123", value=None)

Garantias de entrega no produtor

# acks=0: fire-and-forget — sem confirmação
# acks=1: leader confirma → risco de perda se leader cair antes de replicar
# acks=all (ou -1): todos os ISRs confirmam → máxima durabilidade

# Configuração de produtor para durabilidade máxima
producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    acks='all',                # aguardar todos os ISRs
    retries=2147483647,        # retry infinito até timeout
    max_in_flight_requests_per_connection=1,  # manter ordem com retries
    enable_idempotence=True,   # exactly-once no produtor (dedup por sequence number)
    delivery_timeout_ms=120000,
    linger_ms=5,               # aguardar 5ms para batching
    batch_size=16384,          # tamanho máximo do batch
    compression_type='snappy', # compressão → menos I/O
)

Kafka Streams e ksqlDB

Kafka Streams é uma biblioteca Java/Scala para processamento de streams diretamente dentro de uma aplicação — sem cluster separado como Spark ou Flink. O processamento acontece nos mesmos processos que consomem os tópicos, com estado local em RocksDB e fault-tolerance via changelog topics.

# Kafka Streams — exemplo em pseudocódigo (Java API)
StreamsBuilder builder = new StreamsBuilder();

KStream<String, Order> orders = builder.stream("orders");

// Filtrar apenas pedidos aprovados
KStream<String, Order> approved = orders
    .filter((key, value) -> value.getStatus() == OrderStatus.APPROVED);

// Contar pedidos por cliente em janela de 1 hora (windowed aggregation)
KTable<Windowed<String>, Long> ordersByCustomer = approved
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .count();

// Publicar resultado em outro tópico
ordersByCustomer
    .toStream()
    .map((key, count) -> KeyValue.pair(key.key(), count.toString()))
    .to("orders-per-customer-hourly");

// Estado local consultável via interactive queries (sem round-trip ao Kafka)
ReadOnlyWindowStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType("orders-count-store",
        QueryableStoreTypes.windowStore()));
long count = store.fetch("customer-123", startTime, endTime);

ksqlDB — SQL sobre streams

# ksqlDB — processar streams com SQL
# Criar stream a partir de tópico Kafka
CREATE STREAM orders_stream (
  order_id VARCHAR,
  customer_id VARCHAR,
  total DOUBLE,
  status VARCHAR,
  created_at TIMESTAMP
) WITH (
  KAFKA_TOPIC='orders',
  VALUE_FORMAT='JSON',
  TIMESTAMP='created_at'
);

# Agregação contínua — atualiza conforme novos eventos chegam
CREATE TABLE orders_summary AS
  SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(total) AS total_spent
  FROM orders_stream
  WHERE status = 'APPROVED'
  GROUP BY customer_id
  EMIT CHANGES;

# Push query — retorna resultados continuamente (como subscribe)
SELECT * FROM orders_summary WHERE customer_id = 'cust-123' EMIT CHANGES;

Kafka vs RabbitMQ — quando usar cada um

Comparação por linguagem

Produção e consumo de eventos no Kafka com tratamento correto de offsets, erros e rebalancing em cada linguagem.

// Producer com idempotência e confirmação
using Confluent.Kafka;
using System.Text.Json;

var producerConfig = new ProducerConfig
{
    BootstrapServers = "kafka:9092",
    Acks = Acks.All,
    EnableIdempotence = true,
    MessageTimeoutMs = 30000,
    LingerMs = 5,
    BatchSize = 16384,
    CompressionType = CompressionType.Snappy,
};

using var producer = new ProducerBuilder<string, string>(producerConfig).Build();

var orderEvent = new OrderCreated(orderId, customerId, total);
var message = new Message<string, string>
{
    Key = customerId,               // partition key — mesmo cliente → mesma partição
    Value = JsonSerializer.Serialize(orderEvent),
    Headers = new Headers
    {
        { "event_type", System.Text.Encoding.UTF8.GetBytes("order.created") },
        { "schema_version", System.Text.Encoding.UTF8.GetBytes("1") },
    },
};

// Publicar com confirmação assíncrona
var result = await producer.ProduceAsync("orders", message, cancellationToken);
logger.LogInformation(
    "Publicado: topic={Topic} partition={Partition} offset={Offset}",
    result.Topic, result.Partition.Value, result.Offset.Value);

// ---------------------------------------------------------------
// Consumer com commit manual e graceful shutdown
var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "kafka:9092",
    GroupId = "inventory-service",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnableAutoCommit = false,       // commit manual — após processar
    EnableAutoOffsetStore = false,  // controle explícito do offset store
    MaxPollIntervalMs = 300000,
    SessionTimeoutMs = 45000,
    HeartbeatIntervalMs = 3000,
    PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky,
};

using var consumer = new ConsumerBuilder<string, string>(consumerConfig)
    .SetPartitionsAssignedHandler((c, partitions) =>
    {
        logger.LogInformation("Partições atribuídas: {Partitions}",
            string.Join(", ", partitions));
    })
    .SetPartitionsRevokedHandler((c, partitions) =>
    {
        logger.LogInformation("Partições revogadas — fazendo commit antes do rebalance");
        c.Commit(partitions);
    })
    .Build();

consumer.Subscribe("orders");

try
{
    while (!cancellationToken.IsCancellationRequested)
    {
        var consumeResult = consumer.Consume(cancellationToken);
        if (consumeResult is null) continue;

        try
        {
            var order = JsonSerializer.Deserialize<OrderCreated>(consumeResult.Message.Value)!;
            await inventoryService.ReserveStockAsync(order, cancellationToken);

            // Commit apenas após processar com sucesso
            consumer.StoreOffset(consumeResult);  // store para commit em batch
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            logger.LogError(ex, "Erro ao processar offset {Offset}", consumeResult.Offset);
            // Estratégia de falha: logar + continuar (Kafka não tem DLQ nativo)
            // Publicar em tópico de dead-letter manualmente
            await PublishToDeadLetter(consumeResult, ex);
            consumer.StoreOffset(consumeResult); // não travar — avançar offset
        }

        // Commit em batch a cada 5s (via AutoCommit interno com EnableAutoOffsetStore)
    }
}
finally
{
    consumer.Commit();
    consumer.Close();
}

from confluent_kafka import Consumer, Producer, KafkaError, KafkaException
import json
import logging
import signal
import sys

logger = logging.getLogger(__name__)

# Producer
producer = Producer({
    'bootstrap.servers': 'kafka:9092',
    'acks': 'all',
    'enable.idempotence': True,
    'linger.ms': 5,
    'batch.size': 16384,
    'compression.type': 'snappy',
    'message.timeout.ms': 30000,
})

def delivery_report(err, msg):
    if err:
        logger.error("Falha ao publicar: %s", err)
    else:
        logger.debug("Publicado: %s [%d] @%d", msg.topic(), msg.partition(), msg.offset())

def publish_order_created(order: dict) -> None:
    producer.produce(
        topic='orders',
        key=order['customer_id'],      # partition key
        value=json.dumps(order).encode(),
        headers={'event_type': 'order.created', 'version': '1'},
        callback=delivery_report,
    )
    producer.poll(0)  # disparar callbacks pendentes sem bloquear

# Flush ao encerrar
def shutdown():
    logger.info("Flushing produtor...")
    producer.flush(timeout=10)

# Consumer com commit manual
consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'inventory-service',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False,      # commit manual
    'max.poll.interval.ms': 300000,
    'session.timeout.ms': 45000,
    'heartbeat.interval.ms': 3000,
    'partition.assignment.strategy': 'cooperative-sticky',
})

running = True

def handle_signal(signum, frame):
    global running
    running = False

signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)

def on_assign(consumer, partitions):
    logger.info("Partições atribuídas: %s", [p.partition for p in partitions])

def on_revoke(consumer, partitions):
    logger.info("Commit antes do rebalance")
    consumer.commit(offsets=partitions)

consumer.subscribe(['orders'], on_assign=on_assign, on_revoke=on_revoke)

dead_letter_producer = Producer({'bootstrap.servers': 'kafka:9092'})

try:
    while running:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue

        if msg.error():
            if msg.error().code() == KafkaError.PARTITION_EOF:
                logger.debug("Fim da partição %d", msg.partition())
            else:
                raise KafkaException(msg.error())
            continue

        try:
            event = json.loads(msg.value())
            process_inventory(event)

            # Commit após processar
            consumer.commit(message=msg, asynchronous=False)

        except ProcessingError as e:
            logger.error("Erro de negócio: %s — enviando para DLT", e)
            # Dead letter topic manual (Kafka não tem DLQ nativo)
            dead_letter_producer.produce(
                'orders.dead-letter',
                key=msg.key(),
                value=msg.value(),
                headers={
                    'original_topic': 'orders',
                    'error': str(e),
                    'original_offset': str(msg.offset()),
                },
            )
            consumer.commit(message=msg, asynchronous=False)

finally:
    consumer.close()
    dead_letter_producer.flush()

package kafka

import (
    "context"
    "encoding/json"
    "fmt"
    "log/slog"

    "github.com/twmb/franz-go/pkg/kgo"
)

// Producer com idempotência
func NewProducer(brokers []string) (*kgo.Client, error) {
    return kgo.NewClient(
        kgo.SeedBrokers(brokers...),
        kgo.RequiredAcks(kgo.AllISRAcks()),   // acks=all
        kgo.RecordDeliveryTimeout(30e9),       // 30s timeout
        kgo.ProducerBatchCompression(kgo.SnappyCompression()),
        kgo.ProducerLinger(5e6),               // 5ms linger para batching
        kgo.ProducerBatchMaxBytes(16384),
        kgo.WithLogger(kgo.BasicLogger(nil, kgo.LogLevelInfo, nil)),
    )
}

type OrderEvent struct {
    OrderID    string  `json:"order_id"`
    CustomerID string  `json:"customer_id"`
    Total      float64 `json:"total"`
}

func PublishOrderCreated(ctx context.Context, client *kgo.Client, order OrderEvent) error {
    body, _ := json.Marshal(order)

    record := &kgo.Record{
        Topic: "orders",
        Key:   []byte(order.CustomerID), // partition key
        Value: body,
        Headers: []kgo.RecordHeader{
            {Key: "event_type", Value: []byte("order.created")},
            {Key: "version", Value: []byte("1")},
        },
    }

    // ProduceSync aguarda confirmação do broker
    results := client.ProduceSync(ctx, record)
    for _, res := range results {
        if res.Err != nil {
            return fmt.Errorf("falha ao publicar: %w", res.Err)
        }
        slog.Info("publicado",
            "topic", res.Record.Topic,
            "partition", res.Record.Partition,
            "offset", res.Record.Offset)
    }
    return nil
}

// Consumer com commit manual e graceful shutdown
func NewConsumer(brokers []string, group string) (*kgo.Client, error) {
    return kgo.NewClient(
        kgo.SeedBrokers(brokers...),
        kgo.ConsumerGroup(group),
        kgo.ConsumeTopics("orders"),
        kgo.DisableAutoCommit(),                   // commit manual
        kgo.BlockRebalanceOnPoll(),                // não rebalancear durante poll
        kgo.OnPartitionsAssigned(func(ctx context.Context, c *kgo.Client, assigned map[string][]int32) {
            slog.Info("partições atribuídas", "count", len(assigned))
        }),
        kgo.OnPartitionsRevoked(func(ctx context.Context, c *kgo.Client, revoked map[string][]int32) {
            // Commit antes do rebalance
            if err := c.CommitUncommittedOffsets(ctx); err != nil {
                slog.Error("falha ao commitar antes do rebalance", "err", err)
            }
        }),
    )
}

func ConsumeOrders(ctx context.Context, client *kgo.Client, handler func(context.Context, OrderEvent) error) error {
    deadLetterProducer, _ := NewProducer([]string{"kafka:9092"})
    defer deadLetterProducer.Close()

    for {
        fetches := client.PollFetches(ctx)
        if errs := fetches.Errors(); len(errs) > 0 {
            for _, err := range errs {
                if err.Err == context.Canceled {
                    return nil
                }
                slog.Error("erro no fetch", "err", err.Err)
            }
            continue
        }

        fetches.EachRecord(func(record *kgo.Record) {
            var event OrderEvent
            if err := json.Unmarshal(record.Value, &event); err != nil {
                slog.Error("mensagem inválida", "err", err)
                publishDeadLetter(ctx, deadLetterProducer, record, err)
                return
            }

            if err := handler(ctx, event); err != nil {
                slog.Error("erro ao processar", "order_id", event.OrderID, "err", err)
                publishDeadLetter(ctx, deadLetterProducer, record, err)
                // Avançar offset mesmo em erro para não travar a partição
            }
        })

        // Commit em batch após processar todos os records do fetch
        if err := client.CommitUncommittedOffsets(ctx); err != nil {
            slog.Error("erro no commit", "err", err)
        }
    }
}

func publishDeadLetter(ctx context.Context, p *kgo.Client, original *kgo.Record, cause error) {
    p.ProduceSync(ctx, &kgo.Record{
        Topic: original.Topic + ".dead-letter",
        Key:   original.Key,
        Value: original.Value,
        Headers: append(original.Headers,
            kgo.RecordHeader{Key: "error", Value: []byte(cause.Error())},
            kgo.RecordHeader{Key: "original_offset", Value: []byte(fmt.Sprint(original.Offset))},
        ),
    })
}

Exercícios práticos

1. Configure um cluster Kafka local com 3 brokers (Docker Compose) e crie um tópico com 6 partições e fator de replicação 3. Use kafka-producer-perf-test para medir throughput. Critério: throughput com linger.ms=0 documentado vs linger.ms=20 — diferença de ≥2× esperada; ao matar 1 broker durante a produção (com acks=all), nenhuma mensagem é perdida (verificar via kafka-consumer-groups --describe e contagem de offsets); cluster se recupera quando o broker é reiniciado.
2. Demonstre consumer groups: crie um tópico com 4 partições, inicie 2 consumers no mesmo grupo e publique 100 mensagens. Adicione um 3º consumer e observe o rebalancing. Critério: com 2 consumers, kafka-consumer-groups --describe mostra 2 partições por consumer; ao adicionar o 3º, distribuição muda para 2/1/1 partições; mensagens publicadas durante o rebalancing não são perdidas (total consumido = 100); consumer ocioso quando grupo tem 5 consumers para 4 partições.
3. Implemente dead letter topic manual: falha após 3 tentativas move a mensagem para orders.dead-letter com headers de diagnóstico. Critério: mensagem que sempre lança exceção aparece em orders.dead-letter com headers original_topic, original_offset e error; consumer da DLT loga severity WARN com os dados; o offset da fila principal avança mesmo após a falha (não trava a partição); script de replay republica da DLT para a fila original e processa com sucesso após corrigir o handler.
4. Configure log compaction em customer-profiles. Publique 10 atualizações do mesmo customer_id e force compaction com min.cleanable.dirty.ratio=0.01. Critério: kafka-console-consumer --from-beginning após compaction mostra apenas 1 registro por customer_id (10 customers → 10 registros, não 100); publicar tombstone (value=null) seguido de compaction remove o customer_id completamente do output; verificar via kafka-log-dirs que o tamanho do log reduziu.
5. Implemente Kafka Streams para contar pedidos por cliente em janelas de 5 minutos, publicando resultados em orders-count. Critério: consumer do tópico orders-count mostra contagem atualizada a cada nova mensagem publicada; ao publicar 3 pedidos do cliente A e 2 do cliente B em menos de 5 minutos, o output mostra A:3 e B:2; após 5 minutos, nova janela começa do zero; eventos chegando fora de ordem (com timestamp antigo) são incluídos na janela correta.