A Flink job's keyed state grows without bound across millions of keys. How do you bound it?

## Why the state grows Keyed state is created lazily, the first time a record for a key touches a state handle, and Flink has no notion of a key being "finished". If your key space is unbounded — session ids, order ids, device ids that churn — then every key that ever appeared keeps its entry forever. On `HashMapStateBackend` this shows up as heap pressure and eventually OOM; on `EmbeddedRocksDBStateBackend` it shows up as growing local disk use, growing checkpoint size and slowly degrading access latency as the LSM tree deepens. The diagnosis is usually straightforward: checkpoint size climbing monotonically with no plateau, while throughput is flat. What matters is the fix. ## Mechanism one: state TTL A time-to-live can be assigned to keyed state of any type. You build a `StateTtlConfig` and enable it on the descriptor: ```java StateTtlConfig ttl = StateTtlConfig .newBuilder(Duration.ofHours(24)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); descriptor.enableTimeToLive(ttl); ``` Four properties decide whether it behaves the way you expect. **It is processing-time only.** TTL is measured against wall clock, not against watermarks. A job replaying two years of history in an hour will not expire anything, because only an hour of processing time has passed. If your retention rule is genuinely in event time, TTL is the wrong tool. **The update type sets what refreshes the clock.** `OnCreateAndWrite` (default) refreshes on creation and writes only; `OnReadAndWrite` also refreshes on reads, turning the TTL into an idle-timeout rather than an age limit. **Visibility is separate from cleanup.** `NeverReturnExpired` (default) makes expired state behave as if it does not exist even while the bytes remain — the right choice when data must become unreadable after a retention deadline. `ReturnExpiredIfNotCleanedUp` will hand back an expired value that has not been collected yet. **Collections expire per entry.** All state collection types support per-entry TTLs: list elements and map entries expire independently. That is a strong argument for `MapState` over a `ValueState` holding a map. ## How the cleanup actually happens This is where people get burned. By default, expired values are removed *on read* — `ValueState#value` drops them — and collected in the background if the configured backend supports it. Background cleanup can be turned off with `disableCleanupInBackground()`. The two backends do background cleanup differently: - **Heap backend: incremental cleanup.** The backend keeps a lazy global iterator over the state's entries and advances it on state access and, optionally, on record processing. `cleanupIncrementally(10, true)` checks 10 entries per trigger and additionally triggers per record. The default background cleanup for the heap backend checks 5 entries without cleanup per record processed. Note this is heap-only — setting it for RocksDB has no effect. It also adds latency to record processing, and with synchronous snapshotting the iterator keeps a copy of all keys. - **RocksDB: compaction filter.** A Flink-specific compaction filter runs during RocksDB's own asynchronous compactions and drops entries whose expiry timestamp has passed. `cleanupInRocksdbCompactFilter(queryTimeAfterNumEntries, periodicCompactionTime)` controls how often the filter refreshes its notion of "now" from Flink — the default queries the current timestamp every 1000 entries — and how old a file must be before it is picked for compaction anyway, defaulting to 30 days. Refreshing more often speeds cleanup but costs JNI calls into native code; periodic compaction speeds up removal of rarely-accessed entries at the cost of more compaction work. - **Full snapshot cleanup.** `cleanupFullSnapshot()` excludes expired entries when a full snapshot is taken, shrinking the snapshot. It does not clean local state, and it does not apply to RocksDB incremental checkpointing. The consequence to state out loud in an interview: **if no access happens to the state and no records are processed, expired state persists.** TTL is not a guarantee that disk shrinks by the deadline. It is a guarantee about visibility plus a best-effort reclaim. ## Mechanism two: explicit timers The deterministic tool is a `KeyedProcessFunction`. From `processElement` you register a timer on `ctx.timerService()` — `registerEventTimeTimer(t)` or `registerProcessingTimeTimer(t)` — and implement `onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out)`, where all state is again scoped to the key the timer was created for. Clearing every handle there physically retires the key. Event-time timers fire when the watermark advances to or beyond their timestamp, so this is the mechanism that respects event time and behaves identically on replay. The `TimerService` deduplicates timers per key and timestamp, so the standard "idle timeout" pattern — store the last-modified time in a `ValueState`, register a timer for `lastModified + gap`, and on fire check whether the stored timestamp still matches before clearing — is safe. Flink synchronises `onTimer()` and `processElement()` on the same key, so no concurrent-modification handling is needed. Under RocksDB, timers are themselves stored in RocksDB by default (`state.backend.rocksdb.timer-service.factory`, default `rocksdb`), which scales but is not free — a timer per key is state too. ## Costs and caveats to raise TTL is not free storage-wise: the backend stores a last-modification timestamp alongside each user value. RocksDB adds 8 bytes per stored value, list entry or map entry; the heap backend adds an object reference plus a primitive long. The TTL configuration is not part of checkpoints or savepoints — it is how the currently running job treats the data — and it is not recommended to restore state while lengthening a short TTL, which can produce data errors. Since Flink 2.2.0, switching TTL on or off for existing state migrates cleanly rather than throwing a `StateMigrationException` on restore. ## What a strong answer sounds like Diagnose first — checkpoint size trend, key cardinality, which operator owns the growth. Then bound it: TTL for coarse retention and for the privacy requirement, explicit timers for the semantic "this session is over" rule, `MapState` rather than blobs so entries expire individually, and a hard look at whether the key itself should be coarser. Finish with the honest caveat that TTL is best-effort reclamation, not a scheduled delete.