Kafka com Python: Pipeline de Ingestão com Validação Pydantic e Dead Letter Queue (DLQ)

Na construção de pipelines de dados orientados a eventos, um dos maiores desafios operacionais é lidar com mensagens malformadas ou incompatíveis com o esquema esperado (as chamadas poison pills). Se o consumidor simplesmente ignorar o erro, dados valiosos são perdidos sem rastreabilidade. Se o consumidor travar, toda a partição fica bloqueada, gerando um gargalo (lag) que degrada a saúde do sistema.

Este artigo apresenta uma abordagem prática para resolver esse problema utilizando o ecossistema Python com a biblioteca confluent-kafka e validação de esquema com Pydantic (v2). O objetivo é demonstrar a implementação de um fluxo de Dead Letter Queue (DLQ) com tratamento de erros categorizado em três camadas distintas, garantindo resiliência e consistência operacional.

Semântica At-Least-Once e o Mito da Atomicidade

Antes de prosseguirmos para o código, é crucial estabelecer um princípio fundamental de sistemas distribuídos: o envio de uma mensagem para a DLQ e a confirmação (commit) do offset da mensagem original não constituem uma operação atômica.

Como o Kafka não suporta transações nativas entre leitura de um tópico e escrita em outro de forma distribuída sem o uso da API transacional específica (que exige configurações complexas de Transaction Coordinator), adotamos a semântica de entrega At-Least-Once (Pelo Menos Uma Vez).

Nesse modelo, o fluxo funciona sob o padrão de commit condicionado à confirmação de entrega na DLQ:
1. O consumidor identifica um payload inválido.
2. O consumidor envia o payload corrompido para o tópico de DLQ.
3. O consumidor aguarda a confirmação síncrona de recebimento do broker da DLQ (através de um callback de entrega e um comando flush).
4. Somente após a confirmação positiva da DLQ, o consumidor realiza o commit manual do offset da mensagem original no tópico de origem.

Se a aplicação falhar ou perder a conectividade com o cluster após a mensagem ser gravada na DLQ, mas antes que o commit do offset seja registrado no tópico interno __consumer_offsets, a mensagem original será relida na reinicialização da aplicação ou no rebalanceamento de partições. Consequentemente, a mensagem poderá ser duplicada na DLQ. Esse trade-off é aceitável e preferível à perda silenciosa de dados, exigindo apenas que os processos que consomem a DLQ sejam tolerantes a duplicatas (idempotentes).

---

Configuração do Ambiente Local

Para reproduzir o laboratório, utilizaremos o Docker Compose com a imagem do Apache Kafka mantida pela Bitnami, configurada em modo KRaft (Kafka Raft Metadata Mode), eliminando a necessidade de um container dedicado para o ZooKeeper.

Arquivo docker-compose.yml

Crie um diretório para o projeto e salve o seguinte conteúdo como docker-compose.yml:

version: '3.8'

services:
  kafka-kraft:
    image: docker.io/bitnami/kafka:3.7
    container_name: kafka-kraft
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-kraft:9093
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
      - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - ALLOW_PLAINTEXT_LISTENER=yes
    volumes:
      - kafka_data:/bitnami/kafka

volumes:
  kafka_data:
    driver: local

Instalação das Dependências

Instale as bibliotecas necessárias no seu ambiente virtual Python. Certifique-se de utilizar as versões especificadas para evitar incompatibilidades de API:

pip install confluent-kafka==2.3.0 pydantic==2.6.4

---

Implementação do Pipeline de Ingestão

A implementação do consumidor exige uma arquitetura de código capaz de isolar e tratar erros com base em sua natureza. Dividimos as falhas em três camadas estritas:

1. Camada 1: Defeito Intrínseco do Dado (Poison Pill): Erros de decodificação JSON (JSONDecodeError) ou falhas de validação de esquema (ValidationError do Pydantic). A mensagem é enviada à DLQ com metadados nos cabeçalhos (headers), seguida de flush() síncrono e commit manual do offset.
2. Camada 2: Falha Transitória de Infraestrutura: Indisponibilidade de APIs externas, timeouts ou quedas de banco de dados downstream. O consumidor não deve enviar a mensagem para a DLQ, pois o dado é válido. A aplicação deve registrar o erro, interromper o consumo (levantar exceção) e encerrar sem realizar o commit, forçando o reprocessamento quando o serviço for restabelecido.
3. Camada 3: Falha de Commit: Erros operacionais ao tentar registrar o offset no Kafka. Devem ser tratados de forma isolada, registrando logs de alerta sem reencaminhar mensagens já processadas para a DLQ.

Código do Consumidor (consumer.py)

Salve o código abaixo como consumer.py. Este script implementa o loop de consumo, a validação Pydantic e o roteamento condicional de erros.

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

# Configuração de logs para visibilidade operacional
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("IngestionPipeline")

# --- Modelagem de Dados com Pydantic ---
class TelemetryPayload(BaseModel):
    device_id: str = Field(..., min_length=3, max_length=50)
    temperature: float = Field(..., ge=-50.0, le=100.0)
    timestamp: int = Field(..., description="Epoch timestamp em segundos")

# --- Exceção de Infraestrutura Simulada ---
class DatabaseConnectionError(Exception):
    """Simula uma falha de conexão com o banco de dados downstream."""
    pass

def save_to_database_mock(payload: TelemetryPayload) -> None:
    """
    Simula a persistência de dados. 
    Se o device_id for 'db-fail', simula uma falha de infraestrutura.
    """
    if payload.device_id == "db-fail":
        raise DatabaseConnectionError("Timeout ao conectar ao PostgreSQL principal.")
    logger.info(f"Sucesso: Payload do dispositivo {payload.device_id} persistido.")

# --- Callback de Confirmação da DLQ ---
class DLQDeliveryCallback:
    def __init__(self):
        self.success = False
        self.error_msg = None

    def __call__(self, err, msg):
        if err is not None:
            self.error_msg = str(err)
            self.success = False
            logger.error(f"Callback DLQ: Falha no envio: {err}")
        else:
            self.success = True
            logger.info(f"Callback DLQ: Mensagem entregue ao tópico {msg.topic()} na partição [{msg.partition()}]")

# --- Fluxo Principal do Pipeline ---
def run_pipeline():
    consumer_conf = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'telemetry-ingestion-group',
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False  # Controle manual do ponteiro de leitura
    }

    producer_conf = {
        'bootstrap.servers': 'localhost:9092',
        'acks': 'all'  # Garante escrita em todas as réplicas em sincronia
    }

    consumer = Consumer(consumer_conf)
    dlq_producer = Producer(producer_conf)

    raw_topic = "telemetry-raw"
    dlq_topic = "telemetry-dlq"

    consumer.subscribe([raw_topic])
    logger.info(f"Consumidor iniciado e inscrito no tópico: {raw_topic}")

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

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    raise KafkaException(msg.error())

            # Recuperação dos dados brutos
            raw_data = msg.value()
            logger.info("Mensagem recebida do offset %s: %r", msg.offset(), raw_data)

            try:
                # --- CAMADA 1: Validação e Tratamento de Poison Pills ---
                try:
                    if raw_data is None or len(raw_data) == 0:
                        raise ValueError('Payload nulo ou vazio')
                    decoded_text = raw_data.decode('utf-8')
                    parsed_json = json.loads(decoded_text)
                    telemetry = TelemetryPayload.model_validate(parsed_json)
                except (UnicodeDecodeError, json.JSONDecodeError, ValidationError, TypeError, ValueError) as validation_err:
                    logger.warning(f"Poison Pill detectada no offset {msg.offset()}. Erro: {validation_err}")

                    # Preparação dos cabeçalhos de erro para rastreabilidade na DLQ
                    headers = [
                        ("original-topic", msg.topic().encode('utf-8')),
                        ("original-partition", str(msg.partition()).encode('utf-8')),
                        ("original-offset", str(msg.offset()).encode('utf-8')),
                        ("error-message", str(validation_err).encode('utf-8'))
                    ]

                    # Envio síncrono para a DLQ
                    callback = DLQDeliveryCallback()
                    dlq_producer.produce(
                        topic=dlq_topic,
                        key=msg.key(),
                        value=raw_data if raw_data is not None else b'',
                        headers=headers,
                        callback=callback
                    )

                    # Força a entrega imediata das mensagens pendentes no buffer do produtor
                    dlq_producer.flush(timeout=5.0)

                    if not callback.success:
                        # Se a DLQ falhar, interrompemos o consumidor sem comitar a mensagem original
                        raise RuntimeError(f"Falha ao gravar na DLQ: {callback.error_msg}. Abortando processamento.")

                    # Commit condicionado à confirmação de entrega na DLQ
                    try:
                        consumer.commit(message=msg, asynchronous=False)
                        logger.info(f"Offset {msg.offset()} comitado após desvio para DLQ.")
                    except Exception as commit_err:
                        # --- CAMADA 3: Falha de Commit do Desvio ---
                        logger.error(f"Erro ao comitar offset {msg.offset()} após desvio para DLQ: {commit_err}")
                    continue

                # --- CAMADA 2: Processamento de Negócio e Falhas de Infraestrutura ---
                try:
                    save_to_database_mock(telemetry)
                except DatabaseConnectionError as infra_err:
                    # Falha transitória: NÃO envia para a DLQ. Para o consumidor.
                    logger.critical(f"Falha de infraestrutura detectada: {infra_err}. Encerrando consumidor.")
                    raise infra_err

                # Fluxo de Sucesso: Commit do offset original
                try:
                    consumer.commit(message=msg, asynchronous=False)
                    logger.info(f"Offset {msg.offset()} comitado com sucesso.")
                except Exception as commit_err:
                    # --- CAMADA 3: Falha de Commit de Sucesso ---
                    logger.error(f"Erro operacional ao comitar offset {msg.offset()}: {commit_err}")

            except DatabaseConnectionError:
                # Propaga o erro de infraestrutura para interromper o loop principal
                break

    except KeyboardInterrupt:
        logger.info("Interrupção manual recebida pelo operador.")
    finally:
        # Garante o fechamento limpo dos recursos e a liberação de partições
        consumer.close()
        dlq_producer.flush(timeout=2.0)
        logger.info("Recursos do pipeline liberados com segurança.")

if __name__ == "__main__":
    run_pipeline()

---

Script de Teste e Geração de Carga

Para testar o comportamento do pipeline sob diferentes cenários de erro, usaremos um script produtor que enviará quatro tipos de mensagens:

1. Um payload válido (processamento regular).
2. Um payload com JSON corrompido (falha de decodificação – Camada 1).
3. Um payload com violação de restrições do esquema (comprimento e limites numéricos no Pydantic – Camada 1).
4. Um payload que aciona a simulação de erro de banco de dados (falha de infraestrutura – Camada 2).

Código do Produtor (producer_test.py)

Salve o arquivo abaixo como producer_test.py:

import json
import time
from confluent_kafka import Producer

def delivery_report(err, msg):
    if err is not None:
        print(f"Falha ao enviar mensagem de teste: {err}")
    else:
        print(f"Mensagem enviada com sucesso para {msg.topic()} [{msg.partition()}]")

def run_test_producer():
    conf = {'bootstrap.servers': 'localhost:9092'}
    producer = Producer(conf)
    topic = "telemetry-raw"

    # 1. Payload Válido
    payload_valid = {
        "device_id": "sensor-alpha",
        "temperature": 23.4,
        "timestamp": int(time.time())
    }

    # 2. Payload Inválido: JSON Malformado (Poison Pill)
    payload_corrupted_json = "{device_id: 'sensor-beta', temperature: NaN}"

    # 3. Payload Inválido: Esquema Violado (Pydantic ValidationError)
    payload_invalid_schema = {
        "device_id": "x",  # Curto demais (min_length=3)
        "temperature": 150.0,  # Fora do limite (le=100.0)
        "timestamp": int(time.time())
    }

    # 4. Payload Válido que força erro de Infraestrutura Downstream
    payload_infra_fail = {
        "device_id": "db-fail",
        "temperature": 21.1,
        "timestamp": int(time.time())
    }

    print("Enviando mensagens de teste...")

    # Envia os cenários
    producer.produce(topic, json.dumps(payload_valid).encode('utf-8'), callback=delivery_report)
    producer.produce(topic, payload_corrupted_json.encode('utf-8'), callback=delivery_report)
    producer.produce(topic, json.dumps(payload_invalid_schema).encode('utf-8'), callback=delivery_report)
    producer.produce(topic, json.dumps(payload_infra_fail).encode('utf-8'), callback=delivery_report)

    producer.flush()

if __name__ == "__main__":
    run_test_producer()

---

Roteiro Passo a Passo de Execução e Inspeção

Siga os passos abaixo para executar o ambiente, rodar os testes e verificar o comportamento das filas.

Passo 1: Iniciar o Cluster Kafka

Inicie o container do Kafka em modo KRaft:

docker compose up -d

Verifique se o container está em execução:

docker ps

Passo 2: Executar o Consumidor (Pipeline)

Abra um terminal dedicado e execute o script do consumidor:

python consumer.py

O terminal exibirá a mensagem de inicialização e aguardará eventos.

Passo 3: Executar o Produtor de Testes

Abra um segundo terminal e execute o produtor para injetar as mensagens de teste no cluster:

python producer_test.py

Passo 4: Analisar a Saída do Consumidor

Volte ao terminal onde o consumer.py está rodando. A saída gerada (ilustrativa para acompanhamento dos offsets) demonstra cada camada em ação:

2024-10-24 10:00:00 [INFO] Consumidor iniciado e inscrito no tópico: telemetry-raw
2024-10-24 10:00:05 [INFO] Mensagem recebida do offset 0: {"device_id": "sensor-alpha", "temperature": 23.4, "timestamp": 1713960000}
2024-10-24 10:00:05 [INFO] Sucesso: Payload do dispositivo sensor-alpha persistido.
2024-10-24 10:00:05 [INFO] Offset 0 comitado com sucesso.

2024-10-24 10:00:05 [INFO] Mensagem recebida do offset 1: {device_id: 'sensor-beta', temperature: NaN}
2024-10-24 10:00:05 [WARNING] Poison Pill detectada no offset 1. Erro: Failed to decode JSON: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)
2024-10-24 10:00:05 [INFO] Callback DLQ: Mensagem entregue ao tópico telemetry-dlq na partição [0]
2024-10-24 10:00:05 [INFO] Offset 1 comitado após desvio para DLQ.

2024-10-24 10:00:05 [INFO] Mensagem recebida do offset 2: {"device_id": "x", "temperature": 150.0, "timestamp": 1713960000}
2024-10-24 10:00:05 [WARNING] Poison Pill detectada no offset 2. Erro: 2 validation errors for TelemetryPayload
device_id
  String should have at least 3 characters [type=string_too_short, input_value='x', input_type=str]
temperature
  Input should be less than or equal to 100 [type=less_than_equal, input_value=150.0, input_type=float]
2024-10-24 10:00:05 [INFO] Callback DLQ: Mensagem entregue ao tópico telemetry-dlq na partição [0]
2024-10-24 10:00:05 [INFO] Offset 2 comitado após desvio para DLQ.

2024-10-24 10:00:05 [INFO] Mensagem recebida do offset 3: {"device_id": "db-fail", "temperature": 21.1, "timestamp": 1713960000}
2024-10-24 10:00:05 [CRITICAL] Falha de infraestrutura detectada: Timeout ao conectar ao PostgreSQL principal.. Encerrando consumidor.
2024-10-24 10:00:05 [INFO] Recursos do pipeline liberados com segurança.

Análise do comportamento:
– O offset 0 foi processado e comitado normalmente.
– Os offsets 1 e 2 falharam na validação. Foram enviados para a DLQ e seus offsets foram comitados individualmente após a confirmação de entrega.
– O offset 3 acionou a simulação de falha de banco de dados. O consumidor registrou o erro crítico e encerrou a execução sem comitar o offset 3. Se você reiniciar o consumidor, ele lerá novamente a partir do offset 3.

Passo 5: Inspecionar o Tópico de DLQ

Para auditar as mensagens enviadas à DLQ com seus respectivos cabeçalhos de erro, execute o utilitário oficial do Kafka diretamente dentro do container Docker:

docker exec -it kafka-kraft /opt/bitnami/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic telemetry-dlq \
  --from-beginning \
  --property print.headers=true \
  --property print.key=true \
  --property print.timestamp=true

Exemplo de saída esperada (simplificada para visualização):

CreateTime:1713960005123 \
original-topic:telemetry-raw,original-partition:0,original-offset:1,error-message:Failed to decode JSON... \
null \
{device_id: 'sensor-beta', temperature: NaN}

CreateTime:1713960005456 \
original-topic:telemetry-raw,original-partition:0,original-offset:2,error-message:2 validation errors for TelemetryPayload... \
null \
{"device_id": "x", "temperature": 150.0, "timestamp": 1713960000}

---

Teste de Falha na DLQ

O que acontece se a infraestrutura de DLQ falhar ou a rede cair no momento do encaminhamento? Para evitar a perda irreversível de dados, o consumidor jamais deve avançar o offset da mensagem original se a confirmação da DLQ não for recebida.

Para realizar um teste controlado no ambiente local sem depender de ACLs ou permissões complexas, simule a indisponibilidade da DLQ apontando o produtor para a porta 9999 (sem serviço ativo):

1. No consumer.py, altere temporariamente a inicialização do dlq_producer para uma porta fechada com timeout curto:

   # Simulação de indisponibilidade de rede na DLQ
   dlq_producer = Producer({'bootstrap.servers': 'localhost:9999', 'message.timeout.ms': 2000})
   

2. Inicie o consumer.py e envie os dados de teste com python producer_test.py.
3. Ao receber o offset 1 (poison pill), o consumidor tenta publicar na DLQ. Como o broker na porta 9999 é inalcançável, a entrega falha e o callback não sinaliza entrega bem-sucedida (callback.success == False). O pipeline dispara uma exceção RuntimeError (acusando a falha de entrega da DLQ) e encerra o processo sem executar o commit do offset 1.
4. Inspecione o estado do grupo de consumidores no Kafka com o comando oficial:

   docker exec -it kafka-kraft /opt/bitnami/kafka/bin/kafka-consumer-groups.sh \
     --bootstrap-server localhost:9092 \
     --describe --group telemetry-ingestion-group
   

   Na saída exibida, compare as colunas:

   TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
   telemetry-raw   0          1               4               3
   

   O CURRENT-OFFSET permanece em 1. A mensagem válida do offset 0 foi comitada, mas a mensagem problemática não avançou o checkpoint.

5. Recuperação: Restaure 'bootstrap.servers': 'localhost:9092' no consumer.py e reinicie o consumidor. Ele relerá a mensagem a partir do offset 1, completará o desvio para a DLQ com sucesso e comitará o offset, demonstrando na prática a garantia de reprocessamento at-least-once.

(Nota de execução: A tabela de offsets acima reflete um teste iniciado com uma partição limpa ou novo grupo consumidor. Se o consumidor já tiver sido executado anteriormente, ele terá avançado até o offset 3 e será interrompido pela simulação db-fail antes de avaliar novamente o offset 1. Para isolar o teste da DLQ, limpe os tópicos ou utilize um group.id novo).

Recuperação da falha de infraestrutura downstream: Da mesma forma, para retomar o consumo após a interrupção gerada pelo device_id: 'db-fail' no offset 3, desative o gatilho da falha na função save_to_database_mock antes de reiniciar o consumidor.

---

Armadilhas Operacionais e Resiliência em Produção

Ao migrar este pipeline para produção, atente-se às seguintes configurações críticas do Kafka:

1. Consumer Lag vs. max.poll.interval.ms

O tempo de processamento de cada mensagem (incluindo retentativas de banco de dados e envio à DLQ) não deve exceder o limite configurado em max.poll.interval.ms (padrão de 300000 ms ou 5 minutos). Se o processamento exceder esse tempo, o broker considerará o consumidor morto e iniciará um rebalanceamento de partição, o que gera duplicidade de consumo e instabilidade na topologia.

2. A Configuração de acks=all em Produção

No nosso ambiente local de nó único, a diretiva 'acks': 'all' garante que a mensagem seja gravada localmente. No entanto, em um cluster de produção com múltiplos brokers, essa configuração exige que a gravação seja confirmada pelo líder e por todas as réplicas em sincronia (In-Sync Replicas – ISR).

Se o número de réplicas ativas for menor que o parâmetro min.insync.replicas configurado no tópico, o produtor falhará ao tentar gravar na DLQ. Certifique-se de tratar essa falha operacional como um erro de infraestrutura temporário (Camada 2).

3. Evitando Loops Infinitos na DLQ

Ao consumir mensagens da DLQ para reprocessamento ou análise, certifique-se de que o consumidor da DLQ não reenvie mensagens diretamente para o tópico original sem uma estratégia de correção de dados ou um limite máximo de tentativas (retry count) gravado nos headers da mensagem. Caso contrário, mensagens permanentemente inválidas entrarão em um loop infinito de processamento e desvio.