Kafka CooperativeStickyAssignor: Reduzindo Rebalances

Quando picos súbitos de rebalance quebram SLOs em tempo real, o Kafka CooperativeStickyAssignor elimina pausas gerais no consumo durante deploys. Em uma plataforma de eventos de alto throughput processando 180.000 mensagens por segundo, um deploy rotineiro no Kubernetes ou uma falha transitória de nó pode disparar uma cascata imediata: pods consumidores revogam todas as suas partições, limpam buffers em memória e paralisam o processamento do consumer group inteiro por 35 a 90 segundos. Enquanto o cluster fica ocioso aguardando o coordenador do grupo recalcular as atribuições, filas a montante transbordam, pipelines analíticos ficam desatualizados e engenheiros de plantão enfrentam falsos alarmes de lag. Substituir estratégias tradicionais pelo protocolo cooperativo elimina esses gargalos globais ao restringir a revogação apenas às partições que realmente precisam mudar de nó.

Por Que Rebalances Eager Causam Quedas no Ingestion

Por anos, as estratégias padrão de atribuição de partição no Apache Kafka—como o RangeAssignor e o RoundRobinAssignor—operaram sob um modelo estritamente "eager". Sob um rebalanceamento eager, cada consumidor do grupo revoga todas as suas partições atribuídas assim que o ciclo de rebalanceamento começa. Mesmo que um cluster tenha 64 partições distribuídas de maneira uniforme entre oito nós, a reinicialização de um único pod força todos os oito consumidores a descartar suas cargas ativas, enviar requisições de JoinGroup para o broker coordenador e interromper todo o consumo até que o grupo receba a confirmação SyncGroup final.

Esse comportamento cria riscos operacionais severos em ambientes Kubernetes que utilizam autoscaling horizontal (HPA) ou rolling updates frequentes. Durante um ciclo eager, o consumo é paralisado globalmente em todas as partições do tópico, e não apenas nas partições que estão migrando de pod. Quando os consumidores gerenciam caches locais ou operam conexões transacionais pesadas—como nos padrões discutidos no artigo sobre Kafka connect sink backpressure isolation—o processo de flush de estado durante a revogação total impõe uma sobrecarga imensa. O coordenador é obrigado a esperar pelo consumidor mais lento, multiplicando o tempo de parada e gerando gigabytes de lag acumulado.

Fluxo de Rebalance Eager:
[Consumidor A: P0, P1] ---> Evento ---> Revoga TODAS [P0, P1] ---> Entra no Grupo (Trabalho parado)
[Consumidor B: P2, P3] ---> Evento ---> Revoga TODAS [P2, P3] ---> Entra no Grupo (Trabalho parado)
Coordenador: Calcula nova distribuição enquanto TODAS as partições ficam ociosas.

Fluxo de Rebalance Cooperative Sticky:
[Consumidor A: P0, P1] ---> Evento ---> Mantém [P0], marca [P1] para migrar ---> Continua P0 ativo
[Consumidor B: P2, P3] ---> Evento ---> Mantém [P2, P3] sem interrupção    ---> Continua P2, P3 ativo
Coordenador: Reatribui apenas [P1] para o Consumidor C na Fase 2.

A causa raiz dessa fragilidade é a premissa de que a atribuição precisa ser uma operação única e atômica, exigindo a limpeza global do estado para evitar leituras duplicadas. Em plataformas de streaming modernas, essa premissa impõe uma penalidade inaceitável à disponibilidade operacional do cluster.

Como o Protocolo Cooperative Sticky Modifica a Revogação de Partições

O CooperativeStickyAssignor transforma a coordenação do Kafka de um modelo síncrono e destrutivo em um protocolo incremental de múltiplas etapas. Em vez de exigir a revogação preventiva de todo o catálogo de partições, o protocolo cooperativo permite que consumidores saudáveis continuem processando registros em suas partições atuais enquanto o coordenador calcula quais partições específicas precisam ser realocadas.

Esse fluxo ocorre em duas etapas estruturadas:

1. Etapa de Descoberta e Planejamento: Quando um consumidor entra ou sai do grupo, os consumidores ativos enviam uma chamada JoinGroup informando suas partições ativas. Diferente do protocolo eager, os consumidores não abrem mão de suas partições antes dessa chamada. O líder do grupo analisa a topologia atual, calcula o balanceamento alvo e identifica exclusivamente o subconjunto de partições que precisa migrar. As partições que continuam com seus proprietários atuais operam normalmente sem interrupções.
2. Etapa de Revogação Direcionada e Reatribuição: Os consumidores com partições marcadas para migração recebem a notificação para liberar apenas essas partições específicas. Eles descarregam o estado local correspondente, realizam o commit dos offsets pendentes e disparam um segundo ciclo rápido de rebalanceamento para transferir as partições desocupadas aos novos nós. Partições que não mudaram de dono nunca entram em pausa.

Ao assegurar que a atribuição seja "sticky" (aderente)—preservando as partições onde elas já estão alocadas—o protocolo minimiza o volume de dados migrados. Se um pod de processamento falhar em um grupo de dez instâncias, noventa por cento do tráfego do tópico continuará sendo entregue e processado sem um único milissegundo de interrupção global.

Configuração do Consumidor em Produção e Tratamento de Falhas

A implementação correta do CooperativeStickyAssignor exige alinhar as propriedades dos clientes com os tempos de vida do ciclo de orquestração do Kubernetes. Erros no dimensionamento de threads de heartbeat e intervalos de poll podem disparar ciclos repetitivos de rebalanceamento, anulando os ganhos do protocolo.

O exemplo a seguir demonstra a configuração do assignor utilizando a biblioteca confluent-kafka em Python para serviços de alta taxa de transferência, como o componente de ingestão no projeto Streaming Radar API engine:

import logging
from confluent_kafka import Consumer, KafkaError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("stream_worker")

consumer_config = {
    # Brokers e identificação do grupo
    "bootstrap.servers": "kafka-broker-1.internal:9092,kafka-broker-2.internal:9092",
    "group.id": "order-processing-v2",
    "client.id": "order-worker-k8s-pod-7",
    
    # Configuração de estratégia: atribuição cooperativa e aderente
    "partition.assignment.strategy": "cooperative-sticky",
    
    # Gerenciamento de offsets
    "enable.auto.commit": False,
    "auto.offset.reset": "earliest",
    
    # Tolerância a falhas para pods em Kubernetes
    "session.timeout.ms": 45000,          # Tempo máximo sem heartbeat antes do broker evictar
    "heartbeat.interval.ms": 10000,       # Enviado por thread secundária (1/3 do session timeout)
    "max.poll.interval.ms": 300000,       # Tempo limite de execução na iteração poll()
    
    # Controle de lote de leitura
    "max.poll.records": 500,
    "fetch.max.bytes": 52428800           # Limite de 50 MB de buffer
}

consumer = Consumer(consumer_config)

def on_assign(c, partitions):
    logger.info(f"Partições atribuídas incrementalmente: {[p.partition for p in partitions]}")

def on_revoke(c, partitions):
    logger.info(f"Partições revogadas incrementalmente: {[p.partition for p in partitions]}")
    # Flush de buffers locais e commit síncrono apenas das partições revogadas
    c.commit(offsets=partitions, asynchronous=False)

def on_lost(c, partitions):
    # Acionado quando o nó perdeu as partições de forma abrupta pelo broker
    logger.warning(f"Partições perdidas sem encerramento gracioso: {[p.partition for p in partitions]}")

consumer.subscribe(
    topics=["telemetry.events.raw"],
    on_assign=on_assign,
    on_revoke=on_revoke,
    on_lost=on_lost
)

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
            logger.error(f"Erro de consumo no Kafka: {msg.error()}")
            break
            
        # Processamento do payload do evento
        process_payload(msg.value())
        
        # Commit assíncrono periódico por lote
        consumer.commit(msg, asynchronous=True)
finally:
    consumer.close()

Ao escrever callbacks de ciclo de vida, é fundamental implementar o tratamento de on_lost. Se um pod demorar mais que o limite definido em max.poll.interval.ms devido a uma consulta lenta no banco de dados, o coordenador revoga a partição compulsoriamente. Chamar commits manuais síncronos dentro de um método on_revoke convencional quando a partição já foi marcada como perdida resulta em exceções do tipo CommitFailedException e impede o desligamento limpo do worker.

Métricas Críticas para Monitorar Lag e Latência do JoinGroup

A migração para o particionamento cooperativo exige a evolução das práticas de observabilidade. Focar apenas em métricas genéricas de total de lag mascara atrasos críticos durante as fases incrementais de negociação.

Em plataformas como Prometheus e Datadog, três métricas JMX expostas pelo cliente Kafka devem ser priorizadas em painéis operacionais:

• kafka.consumer:type=consumer-coordinator-metrics,client-id=*,name=rebalance-latency-avg: Acompanha o tempo médio em milissegundos gasto em negociações de atribuição. Em estratégias eager, essa métrica apresenta picos longos durante qualquer restart. Com o assignor cooperativo, as fases preliminares concluem entre 200 e 500 milissegundos sem congelar a leitura.
• kafka.consumer:type=consumer-coordinator-metrics,client-id=*,name=rebalance-rate-per-hour: Quantifica a frequência de alterações no grupo. Se deploys rotineiros geram mais que o dobro de eventos de rebalance esperados, pods estão falhando em responder à segunda fase de confirmação dentro do tempo limite.
• kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,name=records-lag-max: O principal indicador de conformidade com SLOs. Sob a estratégia cooperativa, a taxa de lag em partições não revogadas deve permanecer estável mesmo durante rolling updates de pods no Kubernetes.

O monitoramento detalhado dessas métricas permite isolar falhas de infraestrutura antes que o acúmulo de mensagens afete pipelines de carga como o Real-Time CDC Analytics Pipeline.

Decisões Arquiteturais e Trade-Offs em Produção

A adoção do CooperativeStickyAssignor traz ganhos expressivos, mas introduz particularidades operacionais que exigem atenção dos times de engenharia.

O primeiro desafio envolve migrações em frotas heterogêneas. Não é possível misturar consumidores utilizando estratégias puramente eager com consumidores operando em modo cooperativo sem um plano de transição gradual. O grupo não consegue convergir se parte das instâncias insistir em revogar tudo antes de chamar JoinGroup. A transição suave requer a declaração de múltiplas estratégias na configuração dos clientes legados (exemplo: partition.assignment.strategy = [CooperativeStickyAssignor, RangeAssignor]), permitindo que a versão atualizada assuma o controle após duas etapas sucessivas de rollout.

O segundo aspecto diz respeito a mensagens duplicadas durante falhas de rede. Como o processo ocorre em fases incrementais, um consumidor com alta latência de resposta pode continuar lendo de sua partição por algumas centenas de milissegundos enquanto o coordenador instrui a revogação. Se os sistemas de destino downstream não forem estritamente idempotentes, essa sobreposição pode gerar registros repetidos. Bancos de dados relacionais e data warehouses devem garantir chaves de desduplicação ou operações de upsert transacional.

Em arquiteturas de processamento distribuído baseadas em microsserviços e contêineres dinâmicos, esses cuidados compensam com folga as desvantagens de interrupções globais. Implementar o assignor cooperativo transforma tarefas rotineiras de deploy em operações invisíveis, preservando os acordos de nível de serviço e a consistência dos dados.