An unbounded regular join grows state to 1.4 TB and checkpoints never finish
orders_enrich is a Flink SQL job on 8 task managers (64 GB each, 2 slots) joining orders (append-only, ~25k events/s, Confluent Cloud) with customers, a compacted changelog topic holding 200 million rows at roughly 1 KB each. The statement is a plain inner join on customer_id, projecting segment and region onto each order. No table.exec.state.ttl is set. The state backend is RocksDB without incremental checkpoints, and each checkpoint takes 11-14 minutes against a 10-minute interval, so checkpoints run back to back. State is 1.4 TB and grows roughly 80 GB per week. During each snapshot numRecordsInPerSecond drops from 25k to about 9k, checkpoint alignment takes 4-8 minutes, and lag on orders_raw never fully recovers. Nothing has failed yet, but the team wants both higher throughput and faster recovery, and is debating whether to raise parallelism, tune checkpoints, or change the query.
1. Diagnose the root cause. Why do the checkpoint configuration and the throughput dip follow from it, and why is the answer not simply "add more task managers"?
Read the worked answer
A regular, non-windowed JOIN in Flink SQL is backed by unbounded stateful join operators: both inputs are stored in keyed state keyed by the join key, and nothing is evicted unless table.exec.state.ttl is configured. The customers side alone is on the order of 200 GB of live rows in RocksDB, and the orders side grows without limit - that is the ~80 GB/week. Checkpoint duration is dominated by the synchronous snapshot phase, which must flush and upload all of that state, so as state grows the snapshot stops fitting between checkpoints, alignment stretches to 4-8 minutes because one channel is consistently last, and the throughput dip is the pipeline stalling behind snapshot barriers. Adding task managers increases slot count and distributes RocksDB instances, but each checkpoint still has to move roughly the same total bytes, and more subtasks means more files and more overhead. The query shape is the root cause; the checkpoint and throughput symptoms are downstream of it.
2. Propose a state-reducing change. Compare at least two alternatives and state the correctness consequence of the cheap one most teams reach for first.
Read the worked answer
The cheap fix is table.exec.state.ttl on the join, and its consequence is silent data loss of a specific kind: when a customer row expires, subsequent orders for that key no longer find a match, and an inner join drops them entirely rather than emitting stale enrichment. Cleanup is also lazy, so state can exceed the configured horizon until compaction runs, and the TTL must exceed the maximum out-of-orderness of orders or live matches disappear. A better fit is a temporal join (FOR SYSTEM_TIME AS OF o.proc_time) against the compacted customers changelog: Flink keeps only the latest version per key, which is bounded by the customer count and does not grow with order volume, and results are deterministic per order time. A lookup join with caching against a primary-key store is the third option: it moves state out of Flink but adds external QPS, cache staleness and a non-exactly-once read path. If the business only needs a bounded window of context, an interval join removes the history problem entirely. Choose the semantics first; the TTL is a mitigation, not a design.
3. In what order would you apply checkpoint and recovery tuning, and which of those changes would not have helped on their own?
Read the worked answer
Fix the state shape first, because every checkpoint metric is a function of state size. Then enable RocksDB incremental checkpoints (S3 supports them), which reduces the bytes uploaded per checkpoint after the first full one - but note it would not have helped at 1.4 TB on its own, since the initial full snapshot is unchanged and the intermittent growth still compounds. Next add execution.checkpointing.min-pause of a minute or two so the pipeline gets a stable interval between snapshots instead of running back to back. Enable local recovery to shorten restart time, accepting extra local disk. Consider unaligned checkpoints only if alignment remains high while the job is genuinely backpressured; the trade-off is larger checkpoints carrying in-flight data, and it does nothing if the sink itself cannot keep up. Raising parallelism would have made things marginally worse: same total state, more instances, more overhead per checkpoint. Measure success as checkpoint duration under the interval and alignment under a few seconds, not as restored slot count.