Arquitetura de Ingestão de Dados Resiliente com Apache Kafka e Python

Contexto Operacional e Raiz do Problema

Sistemas distribuídos voltados para alto throughput de eventos enfrentam desafios clássicos de infraestrutura e rede: flutuações de latência, picos imprevisíveis de tráfego (flash crowds) e falhas parciais em nós downstream. Em arquiteturas síncronas baseadas em HTTP/REST, a falha do serviço consumidor reflete diretamente na indisponibilidade do produtor, criando um efeito dominó que afeta a estabilidade do sistema.

Para isolar esses domínios, o padrão de arquitetura orientada a eventos (Event-Driven Architecture) utiliza o Apache Kafka como um log de confirmações gravado em disco e distribuído (distributed commit log). O Kafka oferece garantias de ordenação por partição, retenção configurável e semânticas ajustáveis de entrega de mensagens (At-least-once, At-most-once e Exactly-once).

Entretanto, utilizar o Kafka não elimina a responsabilidade do software em tratar falhas na camada da aplicação. Produtores mal configurados podem perder mensagens durante partionamentos de rede, enquanto consumidores sem tratamento explícito de offsets podem gerar leitura duplicada em massa ou descartar dados silenciosamente ao encontrar payloads malformados (poison pills).

O objetivo desta implementação é projetar um pipeline de ingestão em Python que adote garantias rigorosas de entrega At-least-once, suporte a Dead Letter Queue (DLQ) para isolamento de erros, e serialização tipada de contratos via Pydantic.

---

Setup e Pré-requisitos

Para executar o pipeline em ambiente local ou de staging, são necessários os seguintes componentes:

1. Python 3.10+ instalado.
2. Docker e Docker Compose para execução do cluster Kafka.
3. Biblioteca nativa do cliente C/C++ librdkafka empacotada na biblioteca Python confluent-kafka.

Dependências de Software (Python)

Instale as dependências via pip:

pip install confluent-kafka pydantic pytest

Infraestrutura Local via Docker Compose

Crie o arquivo docker-compose.yml para levantar um broker Kafka em modo KRaft (sem necessidade de Zookeeper):

version: '3.8'
services:
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka-broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'

Inicie o broker com o comando:

docker compose up -d

---

Implementação Completa em Python

O código abaixo integra o produtor idempotente, a validação do contrato de dados via Pydantic e o consumidor com commit manual e redirecionamento de falhas para Dead Letter Queue (DLQ).

import json
import logging
import signal
import sys
import time
from typing import Any, Dict, List, Optional
from confluent_kafka import Consumer, KafkaError, KafkaException, Producer
from pydantic import BaseModel, Field, ValidationError

# Configuração de logging estruturado
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("kafka_pipeline")

class MetricEvent(BaseModel):
    """Contrato de dados tipado para validação de payload."""
    event_id: str
    device_id: str
    timestamp: float
    metric_value: float
    status: str = Field(default="OK")

class ResilientKafkaProducer:
    """Produtor idempotente com garantia de ordenação e retentativas configuradas."""

    def __init__(self, bootstrap_servers: str) -> None:
        conf: Dict[str, Any] = {
            "bootstrap.servers": bootstrap_servers,
            "enable.idempotence": True,
            "acks": "all",
            "retries": 5,
            "max.in.flight.requests.per.connection": 5,
            "compression.type": "snappy"
        }
        self.producer = Producer(conf)

    def delivery_report(self, err: Optional[KafkaError], msg: Any) -> None:
        """Callback invocado após a confirmação de escrita pelo broker."""
        if err is not None:
            logger.error(f"Falha na entrega da mensagem: {err}")
        else:
            logger.info(
                f"Mensagem entregue no tópico {msg.topic()} "
                f"[{msg.partition()}] offset {msg.offset()}"
            )

    def produce(self, topic: str, key: str, value: str) -> None:
        """Envia mensagem ao tópico com tratamento de buffer local cheio."""
        try:
            self.producer.produce(
                topic=topic,
                key=key.encode("utf-8"),
                value=value.encode("utf-8"),
                on_delivery=self.delivery_report
            )
            self.producer.poll(0)
        except BufferError:
            logger.warning("Buffer local cheio. Forçando flush síncrono...")
            self.producer.flush(timeout=2.0)
            self.producer.produce(
                topic=topic,
                key=key.encode("utf-8"),
                value=value.encode("utf-8"),
                on_delivery=self.delivery_report
            )

    def flush(self, timeout: float = 10.0) -> None:
        """Esvazia o buffer pendente do produtor."""
        self.producer.flush(timeout)

class ResilientKafkaConsumer:
    """Consumidor com controle manual de offset e isolamento em Dead Letter Queue."""

    def __init__(
        self,
        bootstrap_servers: str,
        group_id: str,
        topics: List[str],
        dlq_topic: str
    ) -> None:
        conf: Dict[str, Any] = {
            "bootstrap.servers": bootstrap_servers,
            "group.id": group_id,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
            "max.poll.interval.ms": 300000,
            "session.timeout.ms": 45000
        }
        self.consumer = Consumer(conf)
        self.producer = ResilientKafkaProducer(bootstrap_servers)
        self.topics = topics
        self.dlq_topic = dlq_topic
        self.running = True

    def stop(self) -> None:
        """Sinaliza parada graciosa do loop de consumo."""
        self.running = False

    def process_message(self, raw_value: bytes) -> bool:
        """Valida e processa a mensagem consumida."""
        try:
            data = json.loads(raw_value.decode("utf-8"))
            event = MetricEvent(**data)
            logger.info(f"Evento processado com sucesso: ID={event.event_id}, Valor={event.metric_value}")
            return True
        except (json.JSONDecodeError, ValidationError) as err:
            logger.error(f"Falha na validação do payload: {err}")
            return False
        except Exception as err:
            logger.error(f"Erro inesperado no processamento do evento: {err}")
            return False

    def send_to_dlq(self, raw_value: bytes, error_reason: str) -> None:
        """Encaminha mensagens incorretas ou não processáveis para a DLQ."""
        dlq_payload = json.dumps({
            "raw_payload": raw_value.decode("utf-8", errors="replace"),
            "error": error_reason,
            "timestamp": time.time()
        })
        self.producer.produce(self.dlq_topic, key="dlq_error", value=dlq_payload)
        self.producer.flush(timeout=5.0)
        logger.warning(f"Mensagem enviada para a DLQ '{self.dlq_topic}'. Motivo: {error_reason}")

    def consume_loop(self) -> None:
        """Loop principal de ingestão e confirmação de offset."""
        self.consumer.subscribe(self.topics)
        logger.info(f"Consumidor subscrito nos tópicos: {self.topics}")

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

                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    logger.error(f"Erro do consumidor Kafka: {msg.error()}")
                    break

                raw_value = msg.value()
                if raw_value is None:
                    continue

                success = self.process_message(raw_value)
                if success:
                    self.consumer.commit(msg, asynchronous=False)
                else:
                    self.send_to_dlq(raw_value, "Falha de validação ou processamento")
                    self.consumer.commit(msg, asynchronous=False)
        finally:
            self.consumer.close()
            self.producer.flush()
            logger.info("Consumidor encerrado e offsets commitados.")

def run_pipeline() -> None:
    """Função executável de demonstração do pipeline completo."""
    bootstrap_servers = "localhost:9092"
    main_topic = "telemetry-metrics"
    dlq_topic = "telemetry-metrics-dlq"

    # 1. Produção de dados de teste
    producer = ResilientKafkaProducer(bootstrap_servers)

    # Payload Válido
    valid_data = json.dumps({
        "event_id": "evt-101",
        "device_id": "sensor-alpha-01",
        "timestamp": time.time(),
        "metric_value": 87.4,
        "status": "OK"
    })
    producer.produce(main_topic, key="sensor-alpha-01", value=valid_data)

    # Payload Inválido (Poison Pill)
    invalid_data = json.dumps({
        "event_id": "evt-102",
        "device_id": "sensor-beta-02",
        "timestamp": "invalid_timestamp_string",
        "metric_value": "NOT_A_FLOAT"
    })
    producer.produce(main_topic, key="sensor-beta-02", value=invalid_data)
    producer.flush()

    # 2. Consumo e tratamento dos eventos
    consumer = ResilientKafkaConsumer(
        bootstrap_servers=bootstrap_servers,
        group_id="telemetry-processor-group",
        topics=[main_topic],
        dlq_topic=dlq_topic
    )

    def handle_signal(sig: int, frame: Any) -> None:
        logger.info("Interrupção recebida. Solicitando encerramento...")
        consumer.stop()

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

    # Executa o loop por tempo limitado para validação local
    logger.info("Iniciando consumo (pressione Ctrl+C para encerrar)...")
    consumer.consume_loop()

if __name__ == "__main__":
    run_pipeline()

---

Validação Prática e Logs

Ao executar o arquivo acima com o Kafka rodando localmente, os logs impressos no console confirmam a ingestão, o desvio da mensagem corrompida para a DLQ e a confirmação manual de offsets:

2026-03-30 14:00:01,102 [INFO] kafka_pipeline: Mensagem entregue no tópico telemetry-metrics [0] offset 0
2026-03-30 14:00:01,105 [INFO] kafka_pipeline: Mensagem entregue no tópico telemetry-metrics [0] offset 1
2026-03-30 14:00:01,110 [INFO] kafka_pipeline: Consumidor subscrito nos tópicos: ['telemetry-metrics']
2026-03-30 14:00:01,550 [INFO] kafka_pipeline: Evento processado com sucesso: ID=evt-101, Valor=87.4
2026-03-30 14:00:01,558 [ERROR] kafka_pipeline: Falha na validação do payload: 2 validation errors for MetricEvent...
2026-03-30 14:00:01,562 [INFO] kafka_pipeline: Mensagem entregue no tópico telemetry-metrics-dlq [0] offset 0
2026-03-30 14:00:01,563 [WARNING] kafka_pipeline: Mensagem enviada para a DLQ 'telemetry-metrics-dlq'. Motivo: Falha de validação ou processamento

Teste Unitário da Etapa de Ingestão

Para garantir a qualidade em pipelines de CI/CD, validações de esquema devem ser testadas isoladamente:

import pytest
from pydantic import ValidationError

def test_metric_event_schema_validation():
    valid_payload = {
        "event_id": "evt-200",
        "device_id": "device-xyz",
        "timestamp": 1672531199.0,
        "metric_value": 12.34,
        "status": "OK"
    }
    event = MetricEvent(**valid_payload)
    assert event.event_id == "evt-200"

    invalid_payload = {
        "event_id": "evt-201",
        "device_id": "device-xyz",
        "timestamp": "invalid_date",
        "metric_value": "abc"
    }
    with pytest.raises(ValidationError):
        MetricEvent(**invalid_payload)

---

Armadilhas Operacionais, Resiliência e Observabilidade

1. Consumer Group Rebalance Storms

Quando um consumidor demora mais tempo para processar um lote do que o limite definido em max.poll.interval.ms, o broker Kafka assume que a aplicação travou e remove o consumidor do grupo. Isso dispara um Rebalance, pausando o consumo de outros membros do grupo.
* Solução: Otimize o processamento interno, reduza o número de mensagens retornadas por poll, ou utilize multithreading desacoplando a thread do consumidor Kafka da thread de processamento da regra de negócio.

2. Commit Estratégico de Offsets: Síncrono vs Assíncrono

O commit síncrono (asynchronous=False) bloqueia a thread de execução até que o broker confirme o registro do offset. Ele reduz riscos de duplicidade, mas introduz latência. O commit assíncrono melhora o throughput, porém pode gerar leitura duplicada em caso de falha no meio do processamento. Em pipelines críticos onde perda de dados não é tolerada, prefira commits síncronos por mensagem ou por lote finalizado com tratamento rigoroso de exceções.

3. Idempotência do Produtor

Habilitar enable.idempotence=True no produtor garante que retentativas enviadas pela biblioteca devido a timeouts de rede não gerem mensagens duplicadas na partição. O broker atribui identificadores sequenciais às conexões do produtor para desconsiderar confirmações duplicadas.

4. Observabilidade e Métricas Chave

Para operar pipelines Kafka em produção, monitore continuamente:

Métrica	Ferramenta / Origem	Descrição e Ação
Consumer Lag	Prometheus / Kafka Exporter	Quantidade de mensagens não lidas na partição. Lag crescente indica gargalo no processamento.
DLQ Message Rate	Métricas da Aplicação	Frequência de mensagens descartadas. Picos indicam mudanças não sinalizadas no contrato de origem.
Rebalance Count	JMX Broker / Logs do Consumidor	Número de reequilíbrios no grupo de consumo. Frequência alta indica timeouts ou instabilidade de rede.
Buffer Exhaustion	Client Logs / librdkafka	Indica que o produtor está gerando eventos em taxa superior à capacidade da rede/broker.