Do Node.js ao Flink: como migramos o CDC da Woovi

O ponto de partida

O processo, que vou chamar de publisher, abria um change stream por coleção do MongoDB com fullDocument: 'updateLookup' e executava os efeitos dentro dele mesmo. Eram cerca de 20 streams, alguns milhões de eventos por dia, num único pod.

Os efeitos se dividiam assim:

Sempre que este post fala em notificação, é a notificação dentro da própria plataforma da Woovi.

flowchart LR
  M[(MongoDB<br/>oplog)] -->|~20 change streams| P["publisher<br/>1 pod Node.js"]
  P -->|index| ES[(Elasticsearch)]
  P -->|PubSub| R[Redis<br/>subscriptions]
  P -->|publish| Q[MQTT<br/>maquininha e impressora]
  P -->|jobs| B[fila<br/>limpa cache]
  P -->|HTTP| C[Atendimento]
  P -->|send| K[(Kafka)]

Simplificado, o núcleo do publisher era isto:

// versão simplificada do padrão que existia
const stream = Model.watch([], { fullDocument: 'updateLookup' })

stream.on('change', async (change) => {
  // a posição é salva no Redis, com TTL de 60 s, antes do efeito
  await saveResumeToken(change._id)
  await handler(change)
})

async function sendToTopic(topic, payload) {
  try {
    await producer.send({
      topic,
      messages: [{ value: JSON.stringify({ ...payload, eventId: uuidv4() }) }],
    })
  } catch (err) {
    logger.error(err) // o erro morre aqui
  }
}

Cada linha desse trecho esconde um problema.

Por que migrar

O motivo nunca foi custo. O publisher entregava at-most-once por construção e perdia eventos sem deixar rastro, e em vários casos o cliente final foi afetado.

Os defeitos estavam no desenho desde o início, mas enquanto o volume era baixo eles quase não apareciam. Os problemas começaram com o crescimento da empresa e, em especial, depois de um pico no volume de eventos: com mais carga, os handlers concorrentes se acumulavam, a memória estourava, e cada restart abria espaço para perder eventos.

Os mecanismos eram estes:

1. A posição era salva antes do efeito. Se o processo caía depois de gravar o resume token e antes de terminar o handler, o evento sumia.
2. A posição tinha TTL de 60 segundos. Um restart mais lento que isso não achava o token e recomeçava a leitura “de agora”. Essa era a perda mais perigosa, porque não gerava nenhuma linha de log.
3. Erro de envio era engolido. Sem retry e sem dead letter. Ainda havia outros dois caminhos silenciosos: mensagem descartada quando o producer não estava inicializado, e falha de conexão inicial engolida, que deixava o processo sem Kafka até o próximo restart.
4. O eventId era aleatório. Nenhum consumidor conseguia distinguir uma reentrega de um evento novo.
5. Não havia backpressure. O EventEmitter não espera o listener async, então cada evento virava um handler concorrente. O p50 do handler de cobrança ia de dezenas de milissegundos fora do pico a mais de 10 segundos no pico.
6. Não escalava. Change stream não tem consumer group: uma segunda réplica processaria tudo em dobro. Era um pod só, com todos os efeitos no mesmo event loop.

Havia outros problemas menores. Handlers concorrentes não garantiam ordem por documento. Um stream fechado por erro não resumível ficava fechado enquanto o processo seguia vivo e o health check continuava verde. E cada leitor do oplog (o publisher e os jobs Flink que já existiam) fazia o seu próprio updateLookup.

Etapa 1: Elasticsearch

A indexação foi a primeira a sair. Uma dúzia de coleções passou a ser indexada por um job Flink dedicado, e o publisher ganhou uma flag para desligar essa parte.

Busca era um bom primeiro candidato. Ela tolera alguns segundos de atraso, um reprocessamento só reescreve o mesmo documento no índice (idempotente por natureza), e era justamente o efeito que mais contaminava os outros: lentidão ou erro do Elasticsearch atrasava ou derrubava o processo que também cuidava da tela de “pago”.

Com o job separado, o backpressure do Elasticsearch só atrasa a busca. O job lê o oplog, transforma o documento e escreve no índice, e se o cluster fica lento a fonte simplesmente lê mais devagar.

Essa etapa também serviu de treino operacional. Foi com ela que aprendemos que um job Flink parado não avisa ninguém: o job de indexação já ficou quase dois dias em FAILED depois de uma manutenção do banco. Daí vieram os alertas de estado do job e de atraso da fonte que hoje valem para todos.

Antes da etapa 2, fizemos uma limpeza no publisher: removemos o código de Elasticsearch que tinha sobrado e os streams sem efeito nenhum (de cerca de 20 para 10), o que eliminou mais de um milhão de leituras por dia.

Etapa 2: eventos de domínio no Kafka

A decisão central foi separar captura de efeito. A captura é genérica e precisa de estado: ler o oplog, guardar a posição, montar o evento, publicar. O efeito é código de domínio: renderizar o recibo para a impressora, traduzir mensagens, montar o payload da subscription.

O Flink ficou com a captura. Ele publica um tópico <entidade>.upserted por coleção. Os efeitos continuaram em TypeScript, como consumidores Kafka.

flowchart LR
  M[(MongoDB<br/>oplog)] -->|"1 fonte CDC por coleção"| FE["job Flink<br/>eventos"]
  FE -->|"entidade.upserted<br/>chave = _id"| K[(Kafka)]
  K --> W["consumidores Node<br/>tópicos gerais"]
  K --> WC["consumidor Node<br/>cobranças, 6 partições"]
  K --> O["outros serviços<br/>notificação na plataforma, analytics..."]
  W --> R[Redis PubSub] & Q[MQTT] & B[fila de cache] & C[Atendimento]
  WC --> R & Q
  M -->|CDC| FES["job Flink<br/>indexação"] --> ES[(Elasticsearch)]

Portar os efeitos para dentro do Flink teria significado reescrever em Java a renderização ESC/POS da impressora, a internacionalização e a geração do BR Code do tempo real, além de dar ao job credenciais de MQTT e do Redis do produto. E qualquer restart do job pararia a tela de “pago”.

Também consideramos as alternativas de sempre:

KafkaSink<DomainEvent> sink = KafkaSink.<DomainEvent>builder()
    .setBootstrapServers(bootstrapServers)
    .setRecordSerializer(new DomainEventSerializer()) // chave = _id em UTF-8
    .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
    .setKafkaProducerConfig(producerProps) // acks=all, idempotência, compression=none
    .build();

// versão simplificada do claim
const claimed = await redis.set(`event:${event.eventId}`, '1', 'EX', 3600, 'NX')
if (!claimed) return // reentrega, já processado

await handleChange(rebuildChange(event))

O cutover

O primeiro desenho usava janelas de tempo por efeito: cada lado sabia, por configuração, a partir de quando um efeito era dele. Funcionaria, mas exigia parsers dos dois lados e vários deployments coordenados. Trocamos por duas flags:

• uma no Flink, que passa a emitir também os tópicos que o publisher emitia;
• outra compartilhada pelo publisher (que deixa de abrir os streams) e pelos consumidores (que passam a processar).

Isso implica uma janela em que os dois lados fazem os mesmos efeitos. Antes de ligar, auditamos cada consumidor externo dos tópicos: o de notificações da plataforma deduplica por índice único, o de analytics usa ids de job fixos e recalcula, outro faz upsert. A exceção conhecida era a notificação push de “cobrança paga” na plataforma, sem deduplicação, que saiu duas vezes durante a sobreposição.

A ordem importa. Liga os consumidores primeiro e depois desliga o publisher, aceitando duplicata. Na ordem contrária, o intervalo entre os dois deploys perderia efeitos. Em produção, foram cerca de 20 minutos de efeitos duplicados. O rollback é o inverso, com um cuidado: antes de religar os consumidores, os consumer groups precisam ser reposicionados no fim do tópico, ou reprocessam tudo o que ficou para trás.

A verificação teve três partes. Em staging, comparamos os payloads do publisher e do Flink para cobranças reais, em insert e update, e eram idênticos exceto por alguns campos novos. Em produção, comparamos por janelas de 10 minutos a contagem do APM do publisher, as mudanças lidas pelo Flink e o avanço de offset dos consumer groups, e os três bateram com cerca de 1% de diferença. Por fim, acompanhamos os comandos no Redis: os SET NX de claim apareceram na hora da virada, e os PUBLISH do PubSub subiram durante a sobreposição e voltaram ao patamar anterior depois.

Do início da análise à remoção do código e da infraestrutura do publisher foram quatro dias. A remoção tirou cerca de 10 mil linhas do monorepo.

Antes e depois

Em staging, um consumidor ficou 25 minutos fora do ar por um erro de deploy. Quando voltou, consumiu o backlog sem perda e sem nenhuma falha de handler. No modelo antigo isso seria impossível.

O que piorou

Lições

1. A posição de leitura faz parte do dado. Salvar o resume token antes do efeito, ou com TTL, transforma qualquer falha em perda silenciosa.
2. EventEmitter não é fila. Listener async sem await é concorrência ilimitada, e a carga aparece como latência, depois OOM, depois perda.
3. Separar captura de efeito deixou cada parte com o runtime que faz sentido para ela, e o Kafka no meio desacopla as duas.
4. Porte para a outra linguagem só o mínimo que precisa estar no evento, e proteja com fixtures geradas pelo código original.
5. eventId determinístico mais deduplicação no consumidor é o que torna at-least-once tolerável.
6. Sobreposição proposital é mais simples que janelas de tempo, desde que a idempotência de cada consumidor tenha sido auditada antes. O único consumidor que não deduplicava era justamente o que o cliente via.
7. Meça a vazão por coleção contra o esperado. O stall que perdeu pedidos não gerou erro nenhum.
8. Atraso não é perda, mas precisa de dono. O caminho novo acabou com a perda e trouxe atrasos novos, que agora são trabalho de tuning na leitura do oplog e no consumidor.