Como o conector Kafka do Flink faz o commit de offsets no Kafka

O ponto de partida é como o Flink enxerga o Kafka. A unidade de trabalho do source é o split: uma partição mais o próximo offset a ler (e, em splits limitados, o offset de parada). Esse estado vai para o checkpoint junto com o resto do job, e é isso que sustenta o exactly-once. O commit de offsets no Kafka é opcional para o Flink. Serve para dashboards de lag e alertas, como em qualquer consumidor Kafka.

O problema (FLINK-33484) aparecia em tópicos transacionais com isolation.level=read_committed. O emitter definia o offset do split como o offset do último registro emitido mais 1. Só que marcadores de transação e registros de transações abortadas ocupam offsets no log e nunca chegam à aplicação. Com um registro no offset 0 e o marcador no 1, o consumidor ficava na posição 2, mas o Flink commitava 1, e o Kafka reportava lag de 1 num job totalmente em dia. Em tópico quieto, esse lag falso podia durar até chegar dado novo.

A correção usa a posição do consumidor (consumer.position), a única fonte que sabe quantas entradas invisíveis existem. Mas não dá para commitar essa posição às cegas, porque o consumidor busca registros à frente do que o Flink já processou. A reconciliação roda na thread do split fetcher: avança o offset só quando nada fetched está pendente fora do checkpoint. As revisões do PR trouxeram vários casos extras: callback que registrava os offsets pedidos e não os efetivamente commitados (a métrica committedOffset divergia do Kafka), partições sem nenhum registro entregue (lag de 7 numa partição só com transação abortada), o placeholder EARLIEST_OFFSET (-2), splits limitados e o risco de o offset commitado voltar atrás. Agora o split reader guarda o último offset commitado por partição e omite a partição em vez de commitar um valor menor.

O autor afirma que nada mudou nas garantias de processamento, porque o conteúdo do checkpoint é o mesmo. Ele também ressalva que não garante offsets certos em todos os casos: o PR trata de qual offset é commitado, não de quando o commit dispara. O FLINK-36532 segue aberto: no stop-with-savepoint o reader fecha antes de a tarefa de commit enfileirada rodar, e o consumer group fica com offsets antigos ou sem nenhum.

Para quem opera Flink com Kafka transacional, isso significa que alerta de lag baseado em consumer group passa a ser mais confiável, com menos falso positivo em tópicos quietos. Ainda assim, o offset commitado continua sendo canal lateral, de melhor esforço. Para recuperação, vale o estado do checkpoint, não o que o Kafka mostra.