The problem created by isolation

In Part 1, we described why we separated historical snapshot data from the shared CDC mutation stream.

Each application consumes:

• a shared topic carrying live INSERT, UPDATE, and DELETE events
• a private topic carrying snapshot READ events for that application

A custom Flink source combines them into one logical stream.

This architecture isolates rebuild traffic, but it removes any assumption that records from the two inputs arrive in a useful order.

A live mutation can arrive before its corresponding snapshot row, or it can arrive after the snapshot row. The snapshot topic may process faster than the mutation topic, or vice versa. And a restart can change the timing all over again.

The merge must remain correct under each and every one of those schedules.

The invariant

Suppose a source table contains a row for key K.

The snapshot captures K at some earlier database position. While the snapshot is still running, the row may be updated or deleted. When both records reach the Flink job, the newer live mutation must win.

That gives us the invariant: For any key, a snapshot READ may be emitted only if no newer live mutation for that key has already been observed and then forgotten.

The last three words matter because remembering a mutation is easy, but remembering it for long enough is the actual correctness boundary.

Why this is not a conventional join

At first glance, the problem looks like a two-stream join:

• join snapshot records with live mutations by primary key
• choose the newest value

A conventional windowed join is not a good fit; we do not want to buffer both streams until an event-time window closes. Snapshot duration can be long, and the two sources do not provide a useful shared arrival order. We also do not need to retain complete records from both inputs.

For each key, we just need to answer one question: Have we already seen a live mutation that makes this snapshot row stale?

That can be represented with a small piece of keyed state.

The merge operator

The operator has three steps.

1. Partition by primary key

Both snapshot and mutation events are shuffled by the same primary key.

That guarantees all records for key K reach the same parallel operator instance, even though they originated from different Kafka topics.

2. Record live mutations

When an INSERT, UPDATE, or DELETE arrives from the shared mutation stream, the operator:

1. writes a marker for the key into TTL-backed keyed state
2. emits the mutation downstream

The marker does not need to contain the full entity value. Its purpose is to record that a newer live event has been observed for the key.

Conceptually:

3. Filter stale snapshot rows

When a snapshot READ arrives, the operator checks keyed state.

If a marker exists, a newer mutation has already superseded the historical row.

If no marker exists, the snapshot row fills in state that the application has not yet observed from the live stream.

As such, the algorithm deliberately does not depend on ordering between the Kafka topics.

Updates and deletes use the same rule

Updates are the easiest case to visualize.

Imagine the snapshot contains:

The live stream later contains:

If the mutation arrives first, the operator marks K5, emits value_at_t10, and drops the stale snapshot row upon arrival.

Deletes work the same way.

Suppose a row existed when the snapshot was taken but was deleted afterward. The delete mutation records the key and emits the deletion downstream. A later snapshot READ for that key is dropped, preventing the historical row from resurrecting an entity that no longer exists.

The marker represents precedence, not merely the presence of an update.

The role of TTL

Without expiration, mutation markers would grow forever. Every key ever updated would remain in state, even after the snapshot completed.

We therefore store the markers with a time-to-live.

The TTL bounds the extra state introduced by a rebuild, but it also creates the system's most important failure mode. Consider this sequence:

1. The snapshot captures K5 = value_at_t1.
2. A live update produces K5 = value_at_t10.
3. The operator records a marker for K5.
4. The marker expires.
5. The delayed snapshot READ for K5 = value_at_t1 arrives.
6. The operator no longer remembers the mutation and emits the stale value.

Unless another mutation for K5 arrives later, the reconstructed output remains incorrect.

Correctness, therefore, depends on the following operational guarantee:

The mutation-marker TTL must exceed the maximum time between the snapshot boundary and the processing of the final relevant snapshot row, with a safety margin.

In our platform, the default TTL is three days and can be dynamically extended for larger snapshots.

Reasoning about the state window

It is useful to model the operator's memory as a moving range of retained mutation positions. Let:

• ts be the write-ahead-log position represented by the snapshot
• tm-s be the oldest mutation position still represented in keyed state
• tm-e be the newest mutation position the operator has processed

The snapshot can fall in one of three places relative to that range.

Case 1: The snapshot is inside the retained window

The mutation stream has advanced beyond the snapshot boundary, and the operator still remembers mutations that occurred around that boundary.

Suppose the snapshot contains:

The operator has already seen K5_t10, a mutation after the snapshot's K5_t1, so it keeps K5_t10 and fills the keys absent from the keyed state (K3 and K4) from the snapshot. Every key resolves to its true latest value.

This is the state we want the system to remain in until the snapshot completes.

Case 2: The snapshot is ahead of the mutation stream

The change stream has not caught up to ts yet. The snapshot's K2_t9 sits ahead of tm-e, so it is emitted early as a future value (yellow) and re-delivered once the change stream advances to t9, at which point the output converges. Because downstream consumption is idempotent in our jobs, this is harmless.

This case is convergent, but it makes the sink requirement explicit: A pipeline that can observe the same logical value twice must use idempotent, upsert, deduplicating, or otherwise replay-safe output semantics.

A non-idempotent side effect cannot be treated as harmless.

Case 3: The snapshot falls behind the retained window

A mutation made between ts and tm-s has already aged out of the keyed state. The later update to K5 is no longer retained, so the operator does not know K5 changed and emits the snapshot's stale K5_t1 (red) instead of the correct K5_t10, and never corrects it.

This is why the TTL is not merely a performance setting. It is part of the correctness contract.

The platform must keep the system out of this case.

Explicit watermarks versus retained mutation state

This design was inspired by DBLog's watermark-based CDC framework. DBLog writes low- and high-watermark records to the source database and observes them in the transaction log, creating explicit boundaries around each snapshot chunk. It uses those boundaries to interleave selected rows with live log events and suppress stale snapshot values precisely.

Conceptually, DBLog's watermark interval corresponds to our safe tm-s <= ts <= tm-e case, but our boundaries are implicit: the operator derives them from the live mutations represented in TTL-backed keyed state. We size the TTL to outlive the snapshot, keeping the system out of the unsafe ts < tm-s case. We also tolerate ts > tm-e; the snapshot value may be emitted first, but later live mutations restore the latest value under our replay-safe sink semantics.

What the algorithm assumes

The merge is intentionally small, but its guarantees rely on several assumptions.

ts is a logical position, not an LSN

ts is a conceptual approximation of where a snapshot value falls in the source database's mutation history. It is not a PostgreSQL LSN carried by the snapshot event, nor is it a value compared by the Flink operator.

Federated snapshots deliberately avoid LSNs. They are PostgreSQL internals, may vary across versions, and would couple the design to whether the snapshot runs against a primary or secondary instance.

The three cases above describe where a snapshot READ effectively falls relative to the mutation window still retained in TTL-backed keyed state when it reaches the operator. The operator implements this model through marker presence and expiration, not by calculating ts or comparing database positions.

All records for a key reach the same operator instance

Snapshot and live records must use the same primary-key partitioning.

If a schema or keying change routes the two inputs differently, the marker and the snapshot row can end up on different operator instances, causing the algorithm to fail.

Mutation markers survive recovery

The keyed state is checkpointed with the job.

After a restart, the operator must restore both source positions and mutation markers consistently enough that it does not forget which keys have been superseded while replaying snapshot records.

The exact delivery guarantee may still be at-least-once, but the output must converge under the sink's replay semantics.

Downstream publication is replay-safe

Case 2 can produce duplicate values, and blue/green cutover can briefly produce duplicate output from two job versions.

This system is designed for materialized-state and upsert-style consumers. Pipelines with irreversible or non-idempotent side effects require additional deduplication or a transactional boundary.

Snapshot completion

The second input eventually becomes idle, but the control plane still needs an authoritative completion signal.

A robust readiness decision should distinguish at least three states:

1. the snapshot connector has finished reading the source
2. the snapshot topic has been fully consumed
3. the Flink job has checkpointed after processing the final snapshot records and caught up to the live mutation stream

Only then is the reconstructed state ready for a blue/green promotion.

Benchmarking the invariant

We did not want correctness to depend only on a convincing argument.

We built an end-to-end benchmark that reconstructs a table through the composite source while concurrent clients continuously modify the source database. The key property is that the snapshot is triggered while writes are still in flight, so the snapshot READ stream and the live mutation stream overlap on the same keys — the exact condition the invariant governs.

The harness

The workload is driven by pgbench against a source Postgres table, public.hello(id, greeting), using this custom transaction:

Because id is drawn uniformly from a fixed key space of 10,000 keys, every transaction is either an INSERT for a new key or an UPDATE for an existing one. Mutations concentrate on the same keys the snapshot is capturing, maximizing the overlap between the two streams.

The e2e script sequences the run so the snapshot and mutation streams are forced to interleave:

1. Start the pgbench write workload against the source for 100 seconds.
2. Wait 10 seconds, then trigger a federated (incremental) snapshot while writes continue.
3. Let writes run for the remaining ~90 seconds.
4. Wait an additional 30 seconds for the snapshot and mutation streams to fully drain through the Flink job.
5. Compare the reconstructed sink table against the source-of-truth hello table with a set-difference query. Any surviving stale or resurrected row shows up as a differing row.

The run

After the streams drained, the set-difference query returned 0 rows: the state reconstructed through the composite source was identical to the source-of-truth table.

What this run does and does not cover

This workload is upsert-only, so it exercises the INSERT and UPDATE precedence paths (Case 1 and Case 2 above) under real concurrency, but it does not issue DELETEs and does not inject a job restart or a marker expiry. Those scenarios are covered separately in the operator's unit tests; extending the randomized e2e harness to include deletes, restarts, and forced marker expiry is the natural next step for publication-grade evidence.

Costs and tradeoffs

The keyed-state approach is deliberately simple, but it is not free.

During a backfill:

• each recently mutated key creates additional state
• checkpoints and savepoints grow
• RocksDB disk use can increase
• checkpoint duration may increase
• the job consumes from an additional Kafka topic
• a longer TTL raises the state bound

The state size is related to the number of distinct keys mutated during the TTL window, not the total size of the source table.

That is an attractive trade-off for our workloads: we retain a compact precedence marker for recently changed keys rather than buffering or versioning the entire snapshot.

The system also creates more Kafka topics and storage objects. We accepted that infrastructure cost in exchange for isolating rebuild traffic and allowing each pipeline to choose its own backfill capacity.

What we would make explicit in a reusable design

For another team implementing a similar system, we would treat the following as first-class platform controls rather than documentation alone:

• verify that snapshot and mutation events use identical keys
• require replay-safe sink semantics
• calculate a conservative snapshot-duration budget
• reject or warn on TTL configurations below that budget
• alert as elapsed snapshot time approaches the retention boundary
• expose snapshot-topic lag and live-topic lag separately
• track keyed-state size and checkpoint growth during backfills
• persist an authoritative snapshot-generation identifier
• make completion and promotion criteria machine-readable
• test deletes and restarts, not only updates

The merge algorithm is only a few lines of logic. Most of the engineering work is making its assumptions observable and enforceable.

The broader lesson

The central correctness problem is not that two streams can arrive out of order. Stream-processing systems already live with reordering.

The dangerous condition is forgetting precedence too early.

Once a live mutation has established that a historical row is stale, the system must retain that knowledge until the historical input can no longer produce the row.

In our design, keyed state carries that knowledge, and TTL bounds its lifetime.

That gives us a compact operator with a clear contract:

• live mutations always take precedence over historical reads
• the merge does not depend on cross-topic ordering
• temporary duplicates are handled by replay-safe sinks
• correctness fails if mutation memory expires before the snapshot is complete

A clear invariant made the architecture easier to reason about, benchmark, and operate.