The 30 second answer
In Flink SQL each regular join keeps both inputs in state, and a chain of joins stores the intermediate results again at every step, so state multiplies. Zalando's offer enrichment post describes this. They unioned all streams into one stream keyed by SKU and kept a single value state per SKU, updating only the changed fields and dropping older events. State fell from about two hundred forty to fifty-six gigabytes.
The problem the post describes. Zalando's product offer enrichment pipeline joins several streams to build what a shopper sees in the catalog: offer prices and stock from partners, sorting metadata it calls Boost, sponsored-product metadata and product data. It was first written with Flink's Table API and SQL. The post, about Flink 1.20, explains that each join operator is independent and keeps its own copy of the records it needs in RocksDB; the second join stores the output of the first again, and so on. The application reached roughly 235–245 GB of state.
Why that hurt. Hourly snapshots of that much state kept CPU at 100% for around 12 minutes, caused backpressure and lag, and often ended in crash-restart loops. Scaling required stopping with a snapshot, so every scaling operation took a similar time, and the team ran with extra parallelism as a safety margin, which cost money.
What the docs say. Flink's own documentation for regular SQL joins warns that they require keeping "both sides of the join input in Flink state forever", so state can grow without bound, and suggests a state TTL, noting that TTL may affect the correctness of the result. For an enrichment where an old product's price must never be forgotten, TTL is not a safe fix.
The fix. The team moved to the DataStream API, unioned all input streams into one stream of a common base event type, keyed it by SKU, and processed it in a single KeyedProcessFunction holding one ValueState per SKU. Each event updates only its own fields of that state object. There is no left or right side and no intermediate copy. Events older than the stored timestamp are dropped before any state write, and events without relevant changes can be skipped too.
The results reported. State went from 235 GB to 56 GB, snapshot duration from about 11 to 2.5 minutes, CPU from spiking at 100% to about 30% and stable, and AWS cost fell by 13%. The post explains why cost did not fall by 75%: managed Flink is billed by processing units that bundle CPU and memory, and the saving came mainly from removing the over-provisioning. It also notes that Flink 2.1 introduced an experimental multi-way join operator based on the same idea.
How to use this in an interview. The general lesson is to count state per operator, not per pipeline. When several streams share a key and you need only the latest value per key, one keyed operator with explicit state is often smaller and easier to reason about than a chain of declarative joins.
Alternate interview scenarios
"Events can arrive out of order. How does the keyed union stay correct?"
Each stream carries an event timestamp or version, and the state keeps the latest timestamp per field group. An event older than what is stored is dropped. That makes the operator idempotent under replays and robust to out-of-order arrival within a key, provided producers stamp events at the source rather than at ingestion.
"Why not just add a state TTL to the SQL joins?"
Because TTL forgets keys that have been quiet longer than the TTL, and a product whose price has not changed for weeks is still on sale. When it next changes, the join would produce an incomplete record. TTL suits joins where old data is genuinely irrelevant, such as session windows, not a "latest known state per SKU" view.
Sources (3)
- Zalando Engineering: Why We Ditched Flink Table API Joins: Cutting State by 75% with DataStream Unionsengineering.zalando.com
- Apache Flink 1.20: Joins in SQL (regular joins and state)nightlies.apache.org
- Apache Flink 1.20: Working with state (keyed state and ValueState)nightlies.apache.org