Key Group Ceiling

A team sizes a stateful job at parallelism 24 and it runs for a year. Peak forecasts arrive, 200 task slots are provisioned, and the restart fails in seconds with a message nobody has seen before. The hardware is there and the job will not use it.

Keyed state is not assigned to subtasks. It is assigned to key groups, and each subtask owns a contiguous range of them. The key-group count is the real unit of parallelism, and it was fixed the first time the job wrote a snapshot.

Why it matters

This is one of the few capacity decisions in a streaming system that cannot be undone with a restart. Scaling a stateless service is a number in a deployment file; scaling a keyed stateful job past its ceiling means rewriting its state offline. The decision is made implicitly, by a default, at the moment of the first checkpoint, which is usually a Tuesday afternoon when someone was testing.

Flink derives a default from the job's initial parallelism, with a floor of 128 and an upper bound of 32768. A job started small therefore typically carries 128, and 128 is a plausible-looking number that nobody questions until peak.

Implementation patterns

• Set it explicitly in the job definition, at four to eight times the largest parallelism anyone can foresee, as a power of two. 1024 is a reasonable default for a job that might one day run at 128 subtasks.
• Keep running parallelisms as divisors of it. Key groups spread evenly only when parallelism divides the count. At 128 groups and 100 subtasks, 28 subtasks own two groups and 72 own one, so the busiest does 2× the work of the idlest and sets the job's throughput while the dashboard shows an average.
• Record it in the runbook beside the checkpoint interval, because it is invisible in every metric and only appears in savepoint metadata.
• To change it, rewrite the savepoint with the State Processor API: an offline batch job that reads the keyed state and writes a new savepoint with the new count. The original is untouched, so the step is reversible.
• Check the source partition count first. A job reading 24 partitions gains nothing from the 25^th subtask, so the ceiling may not be the binding constraint at all.

Industry example

The mechanism is documented in Apache Flink's own state and rescaling documentation, which has carried the key-group model since the 1.x line, and the State Processor API exists in the project specifically because changing the value otherwise means discarding state. Teams running multi-terabyte keyed state in production hit it on the first serious capacity expansion, which is typically the first real peak after a product launch.

Failure scenarios

• The restart that will not start, discovered during the change window for the peak it was meant to serve.
• Hidden skew: a parallelism that does not divide the key-group count gives some subtasks twice the state and twice the work, so checkpoint duration and backpressure concentrate on a few tasks while aggregate throughput looks fine.
• An emergency rewrite: 4 TB of state read and rewritten at a few hundred megabytes per second is hours of work, performed under time pressure, on the only copy anyone is thinking about.
• Silent loss when someone resolves it by starting the job without state, which quietly resets every aggregate.

Trade-offs

A high key-group count costs a small amount of metadata per group and a little overhead in state assignment; a low one costs an entire migration later. The asymmetry is extreme, which makes this one of the rare cases where over-provisioning on day one is unambiguously correct. The counter-pressure is that a very high count with low parallelism creates many small state partitions, which adds files and metadata to every checkpoint, so 32768 for a job that will never exceed 32 subtasks is also wrong.

When not to use it

Not every job needs the consideration. A stateless job has no key groups and rescales freely, and so does a keyed job whose state has a time-to-live of minutes: throw it away and rebuild from the log at whatever value you like, which is faster than any savepoint rewrite. The ceiling only binds where state is large, long-lived and expensive to rebuild from the source.

Interview question

Q: A keyed job at parallelism 24 must reach 200 slots before Black Friday, holding 4 TB of state. The restart fails immediately. Sequence the change so every step is reversible, and tell me where the point of no return is.

What a strong answer covers: the key-group model and the default's floor of 128; reading the current value from savepoint metadata; choosing a power-of-two target and keeping parallelisms as divisors; rewriting the savepoint offline with the State Processor API; shadow-running against scratch sinks with a separate consumer group and a separate transactional producer id; the point of no return being the new job's first committed write past the old job's stop offset; and checking the source partition count before any of it.

Quick check

Quiz: A job runs at parallelism 24 with no explicit setting. What is its likely maximum parallelism and why? 128 - the floor of the framework's default derivation, fixed at the first snapshot.

Flashcard: Why does parallelism 100 against 128 key groups cost throughput? 28 subtasks own two groups and 72 own one, so the busiest carries 2× the load and sets the pace.