From Node.js to Flink: how we migrated Woovi's CDC

The starting point

The process, which I’ll call the publisher, opened one change stream per MongoDB collection with fullDocument: 'updateLookup' and ran the side effects inside itself. There were about 20 streams and a few million events per day, all in a single pod.

The side effects broke down like this:

Whenever this post talks about notifications, it means notifications inside Woovi’s own platform.

flowchart LR
  M[(MongoDB<br/>oplog)] -->|~20 change streams| P["publisher<br/>1 Node.js pod"]
  P -->|index| ES[(Elasticsearch)]
  P -->|PubSub| R[Redis<br/>subscriptions]
  P -->|publish| Q[MQTT<br/>card machine and printer]
  P -->|jobs| B[queue<br/>clears cache]
  P -->|HTTP| C[Customer support]
  P -->|send| K[(Kafka)]

Simplified, the core of the publisher was this:

// simplified version of the pattern that existed
const stream = Model.watch([], { fullDocument: 'updateLookup' })

stream.on('change', async (change) => {
  // the position is saved in Redis, with a 60 s TTL, before the effect
  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) // the error dies here
  }
}

Every line of that snippet hides a problem.

Why migrate

The reason was never cost. The publisher delivered at-most-once by construction and lost events without a trace, and in several cases the end customer was affected.

The defects were in the design from the start, but while volume was low they barely showed. The problems began as the company grew, and especially after a spike in event volume: with more load, concurrent handlers piled up, memory blew up, and every restart opened room for lost events.

These were the mechanisms:

1. The position was saved before the effect. If the process died after writing the resume token and before the handler finished, the event was gone.
2. The position had a 60 second TTL. A restart slower than that couldn’t find the token and started reading again “from now”. This was the most dangerous loss, because it didn’t produce a single log line.
3. Send errors were swallowed. No retry and no dead letter. There were two more silent paths: messages dropped when the producer wasn’t initialized, and a swallowed initial connection failure that left the process without Kafka until the next restart.
4. The eventId was random. No consumer could tell a redelivery from a new event.
5. There was no backpressure. EventEmitter doesn’t wait for an async listener, so every event became a concurrent handler. The p50 of the charge handler went from tens of milliseconds off peak to over 10 seconds at peak.
6. It didn’t scale. A change stream has no consumer group: a second replica would process everything twice. It was a single pod, with every effect on the same event loop.

There were smaller problems too. Concurrent handlers didn’t guarantee per-document ordering. A stream closed by a non-resumable error stayed closed while the process kept running and the health check stayed green. And each oplog reader (the publisher and the Flink jobs that already existed) did its own updateLookup.

Stage 1: Elasticsearch

Indexing was the first thing to move out. A dozen collections started being indexed by a dedicated Flink job, and the publisher got a flag to turn that part off.

Search was a good first candidate. It tolerates a few seconds of delay, reprocessing just rewrites the same document in the index (idempotent by nature), and it was exactly the effect that contaminated the others the most: slowness or errors in Elasticsearch delayed or crashed the process that also handled the “paid” screen.

With the job split out, Elasticsearch backpressure only delays search. The job reads the oplog, transforms the document and writes to the index, and if the cluster slows down the source simply reads more slowly.

This stage also served as operational training. It’s where we learned that a stopped Flink job doesn’t warn anyone: the indexing job once sat almost two days in FAILED after a database maintenance. That’s where the job state and source lag alerts that now cover every job came from.

Before stage 2, we cleaned up the publisher: we removed the leftover Elasticsearch code and the streams with no effect at all (from about 20 down to 10), which eliminated more than a million reads per day.

Stage 2: domain events in Kafka

The central decision was to separate capture from effect. Capture is generic and needs state: read the oplog, keep the position, build the event, publish. The effect is domain code: render the receipt for the printer, translate messages, build the subscription payload.

Flink took capture. It publishes one <entity>.upserted topic per collection. The effects stayed in TypeScript, as Kafka consumers.

flowchart LR
  M[(MongoDB<br/>oplog)] -->|"1 CDC source per collection"| FE["Flink job<br/>events"]
  FE -->|"entity.upserted<br/>key = _id"| K[(Kafka)]
  K --> W["Node consumers<br/>general topics"]
  K --> WC["Node consumer<br/>charges, 6 partitions"]
  K --> O["other services<br/>in-platform notification, analytics..."]
  W --> R[Redis PubSub] & Q[MQTT] & B[cache queue] & C[Customer support]
  WC --> R & Q
  M -->|CDC| FES["Flink job<br/>indexing"] --> ES[(Elasticsearch)]

Porting the effects into Flink would have meant rewriting in Java the printer’s ESC/POS rendering, the internationalization and the real-time BR Code generation, plus giving the job MQTT credentials and access to the product’s Redis. And any job restart would stop the “paid” screen.

We also considered the usual alternatives:

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

// simplified version of the claim
const claimed = await redis.set(`event:${event.eventId}`, '1', 'EX', 3600, 'NX')
if (!claimed) return // redelivery, already processed

await handleChange(rebuildChange(event))

The cutover

The first design used time windows per effect: each side knew, through configuration, from what point an effect belonged to it. It would have worked, but it required parsers on both sides and several coordinated deployments. We swapped it for two flags:

• one in Flink, which starts emitting the topics the publisher used to emit as well;
• another shared by the publisher (which stops opening the streams) and the consumers (which start processing).

That implies a window where both sides perform the same effects. Before turning it on, we audited every external consumer of the topics: the platform notifications one deduplicates through a unique index, the analytics one uses fixed job ids and recomputes, another does an upsert. The known exception was the in-platform “charge paid” push notification, with no deduplication, which went out twice during the overlap.

Order matters. Turn the consumers on first and then turn the publisher off, accepting duplicates. In the reverse order, the gap between the two deploys would lose effects. In production, there were about 20 minutes of duplicated effects. Rollback is the reverse, with one caveat: before turning the consumers back on, the consumer groups need to be repositioned at the end of the topic, or they reprocess everything that was left behind.

Verification had three parts. In staging, we compared publisher and Flink payloads for real charges, on insert and update, and they were identical except for a few new fields. In production, we compared, in 10 minute windows, the publisher’s APM count, the changes read by Flink and the offset progress of the consumer groups, and all three matched within about 1%. Finally, we watched the commands in Redis: the claim SET NX calls showed up at the moment of the switch, and the PubSub PUBLISH calls went up during the overlap and returned to their previous level afterwards.

From the start of the analysis to the removal of the publisher’s code and infrastructure took four days. The removal took about 10 thousand lines out of the monorepo.

Before and after

In staging, a consumer was down for 25 minutes because of a deploy mistake. When it came back, it consumed the backlog with no loss and not a single handler failure. Under the old model that would have been impossible.

What got worse

Lessons

1. The read position is part of the data. Saving the resume token before the effect, or with a TTL, turns any failure into silent loss.
2. EventEmitter is not a queue. An async listener without await is unbounded concurrency, and load shows up as latency, then OOM, then loss.
3. Separating capture from effect let each part run on the runtime that makes sense for it, and Kafka in the middle decouples the two.
4. Port to the other language only the minimum that has to be in the event, and protect it with fixtures generated from the original code.
5. A deterministic eventId plus consumer-side deduplication is what makes at-least-once tolerable.
6. A deliberate overlap is simpler than time windows, as long as each consumer’s idempotency was audited first. The only consumer that didn’t deduplicate was exactly the one the customer saw.
7. Measure per-collection throughput against what you expect. The stall that lost orders didn’t produce a single error.
8. Delay isn’t loss, but it needs an owner. The new path ended the losses and brought new delays, which are now tuning work on oplog reading and on the consumer.