O streaming tem um problema de silêncio.

A versão curta

Quando uma partição do Kafka fica em silêncio, ela terminou, ou está apenas lenta? Vistas de fora, as duas situações parecem idênticas — e a resposta determina se você pode agir com segurança sobre o que já tem. Netflix, Uber e Walmart publicaram como lidam com isso, e cada uma abre mão de algo: um segundo pipeline, uma via de paralelismo, a correção durante o replay. Engenharia sólida, em todos os casos. Nenhuma delas pergunta ao Kafka a única coisa que o Kafka já sabe.

Encontrei esse problema em um stream de inventário de uma empresa da Fortune 50: treze tópicos Kafka, três partições cada, centenas de milhões de eventos por dia, cada evento de um dado produto exigindo ordem estrita de event time — lendo sete dias de histórico do zero e, em seguida, entrando diretamente no tráfego de pico ao vivo. O primeiro design colocava uma janela de trinta minutos na frente de tudo, ainda precisava de um segundo job para varrer o que a janela deixava escapar, e ainda assim deixava 5,5% dos registros saírem na ordem errada.

A versão que entrou em produção não espera por um relógio. Ela lê os offsets do Kafka.

Eis como — e por que a ideia é mais antiga que o Kafka.

A troca que todos fazem / §01

Este não é o problema de uma equipe só. Quatro organizações de engenharia fortes, quatro trocas sensatas:

• O Walmart fixa cada par item–loja em uma única partição, de modo que apenas um consumidor o atualize. Ordenação por fixação de chaves — abre mão de paralelismo.
• O Uber executa um pipeline separado de backfill para recomputar após eventos atrasados e fora de ordem, porque o caminho de streaming "pode causar imprecisões ao descartar quaisquer eventos que cheguem após o watermarking". Abre mão de um segundo pipeline.
• A Netflix, ao fazer backfill de pipelines Flink, constatou que o histórico reproduzido gerava uma ordem diferente e estado descontrolado. Abre mão de correção e de estado.
• A orientação da própria Confluent para uma partição Kafka ociosa: esperar por ela e travar, ou ignorá-la e aceitar "muitas mensagens atrasadas quando a partição voltar a ficar ativa". Abre mão com travamentos ou dados atrasados.

Todas são razoáveis. O fio condutor é que cada uma infere completude a partir do tempo — uma janela, um timeout, uma espera — quando os offsets das partições dizem exatamente onde os dados terminam. É essa a lacuna que isto fecha.

Por que o stream fica embaralhado / §02

Um tópico Kafka é dividido em partições, e cada partição é seu próprio trilho. Dentro de uma partição, a ordem é garantida: registro um, depois registro onze, sempre. Entre partições, nada coordena nada — os consumidores que as leem nunca estão no mesmo ponto ao mesmo tempo. O Kafka faz exatamente o que promete, os consumidores fazem exatamente o que prometem, e ainda assim a saída combinada fica embaralhada para qualquer coisa que dependa de sequência.

Você não pode agir sobre "vendido" antes de "em estoque" para a mesma unidade. Então, antes de um pipeline liberar qualquer coisa, ele precisa responder a uma pergunta: tenho tudo até este momento? Quase toda resposta em uso hoje é um temporizador — a janela, o timeout de ociosidade, o job de reconciliação. Cada uma é um palpite sobre o silêncio.

Uma solução antiga: reportar, não esperar / §03

Há um precedente, e ele não é um relógio melhor. As primeiras ferrovias operavam pelo sistema de intervalos de tempo: um trem seguinte partia uma vez decorrido um intervalo fixo, de modo que a linha "devia" estar livre.

Em junho de 1889, um trem parou em uma rampa íngreme nos arredores de Armagh, na Irlanda. Os vagões traseiros foram desacoplados, desceram a colina de volta e atingiram o trem que vinha atrás. Cerca de oitenta pessoas morreram.

O trem seguinte havia partido de Armagh assim que os vinte minutos exigidos se passaram. Vinte minutos haviam se passado, logo a linha devia estar livre. Ninguém em Armagh tinha como saber que não estava.

A Grã-Bretanha não respondeu com um relógio melhor. O Parlamento deu ao Board of Trade o poder de exigir o bloqueio absoluto nas ferrovias de passageiros: nenhum trem entra em uma seção da via até que a cabine de sinalização à frente confirme que a seção está livre. A liberação deixou de ser uma duração que alguém esperava passar e tornou-se um fato que alguém reportava. Um trem parado e um trem que já partiu deixaram de parecer a mesma coisa, porque ninguém mais interpretava nada a partir do silêncio.

Esse é todo o movimento. Ler os offsets do Kafka é a cabine de sinalização.

A construção — três peças / §04

O princípio é curto: o tempo só avança quando a evidência prova que pode. Torná-lo real exigiu três peças, e a parte difícil não estava onde as pessoas esperam.

Um: a fila de timers é a ordenação. Cada registro é estacionado em estado keyed sob seu próprio event time, com um timer de event time para aquele milissegundo exato. Nada é emitido ainda. Quando o watermark alcança aquele timestamp, o timer dispara e tudo que estava estacionado ali sai junto. O Flink dispara os timers de uma chave em ordem crescente de timestamp, então nunca ordenamos nada. É um método e um pedaço de estado — a dificuldade nunca esteve aqui.

Dois: o stream sem event time torna-se o relógio. Um dos treze tópicos não tinha event time algum — ele apenas publicava produtos válidos para que todos os outros se enriquecessem com eles. Portanto, não tinha nada a ordenar, e podia fazer o que os outros não podiam: emitir watermarks a partir do tempo de processamento, sempre o timestamp mais recente possível. O porém: o relógio de um pipeline é o mínimo entre suas entradas, então um stream rodando no "agora" não muda nada enquanto os streams de histórico ainda estão sendo contabilizados. Construí essa peça primeiro e ela não teve efeito algum — não estava errada, estava na ordem errada. O stream de controle só importa quando os outros saem do mínimo, e um stream sai do mínimo quando é marcado como ocioso.

Três: o Kafka decide quando um stream terminou — não um timeout. A ociosidade nativa do Flink é um temporizador: sem dados por um tempo, marque como ocioso. Mas um timeout não consegue distinguir "inicializando" de "terminado", e a inicialização é o pior caso — as conexões se formam em velocidades diferentes, um tópico fica em silêncio, o timeout dispara, os outros avançam o relógio para além de event times que ele ainda não leu, e seu primeiro registro chega já atrasado. Isso não é um atraso, é uma violação. Então o timeout tinha que sair. Na inicialização, antes de ler um registro, o job pergunta ao Kafka o último offset de cada partição e o guarda. Um tópico só está ocioso quando cada partição alcançou esse número — não a maioria, todas. Nenhuma duração para ajustar, porque o portão é um fato vindo do Kafka.

// once, at job startup
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

// as records arrive
if (offset + 1 >= startupEndOffsets.get(partition)) markCaughtUp(partition);

O que isso rendeu / §05

A janela retinha cada registro até fechar — trinta minutos só para ler o histórico, de novo a cada reinício. Agora um registro é liberado assim que o watermark o ultrapassa: em média menos de 100 milissegundos, incluindo os registros retidos para que não saíssem fora de ordem. Um registro que chega no minuto um sai no minuto um. A ordenação seguiu o mesmo caminho: duas camadas de ociosidade por timeout deixavam 5,5% dos registros fora de ordem; uma camada, 1,5%; ociosidade decidida pelos offsets de inicialização do Kafka, zero. A janela se foi e o job de varredura também — um único job lê sete dias de histórico e depois roda ao vivo.

O que isso não promete / §06

Determinístico não é clarividente. Um evento só existe para o pipeline quando chega ao Kafka. Se um produtor retém um registro e ele chega depois que o relógio já passou do seu event time, ele sai imediatamente, fora da vez, e o contador de violações o registra. Nenhum processador de streams consegue ordenar um evento que ainda não pode ver. Dentro dessa fronteira a garantia é exata, e o lag do watermark define a largura da fronteira.

A mesma pergunta, em toda parte / §07

Depois de vê-la uma vez, você a vê em todo sistema que decide o que é verdade a partir de streams que não chegam juntos.

Um medidor fluvial fica em silêncio durante uma enchente. O rio está estável, ou a água alcançou o sensor? Em 2018, o USGS relatou medidores de vazão derrubados pelo furacão Florence — justamente os que mais importavam. "Sem notícias" é a leitura errada para esse silêncio.

Um conjunto de câmeras. Capture um atleta com várias câmeras e cada uma é seu próprio stream, como uma partição; você não pode fundir uma pose 3D para um instante até que cada câmera o tenha reportado. Uma câmera silenciosa pode ter perdido um quadro ou estar observando alguém imóvel. Na i3d.ai é a mesma pergunta, dezenas de vezes por segundo.

Um agente. Um agente de IA chama uma ferramenta e não ouve nada de volta. Agentes são muito bons em tratar silêncio como sucesso — nenhum erro retornou, então siga para o próximo passo em velocidade de máquina. Um agente agindo sobre a conta de um cliente como ela estava há uma hora é um trem despachado por um temporizador. A solução é a que as ferrovias encontraram: nenhum passo avança com base no silêncio; todo passo avança com base em uma confirmação que ele pode apontar.

O ganho de confiabilidade mais barato em streaming não é um engine mais rápido ou uma janela maior. É parar de inferir "terminado" a partir do silêncio e perguntar ao sistema que já sabe. A janela espera pelo pior caso para cada registro; a verificação de offset espera exatamente o quanto a evidência exige, e nem um milissegundo a mais.

A implementação de referência enxuta — gerador, handler, rastreador de recuperação — é a próxima coisa que vou publicar.