The Savepoint That Grew Until It Killed Itself: Reproducing a Kafka Connector Migration OOM

How a one-line connector swap silently poisoned a savepoint, why raising heap memory only bought time, and what it took to cut the stale state out.

The incident

A Ververica Platform customer had several production Flink pipelines stuck in a restart loop, each dying the same way: java.lang.OutOfMemoryError: Java heap space, every time, during savepoint restore. Not steady-state processing — restore. The job would fail before it processed a single record.

The stack trace pointed at one specific place:

java.lang.OutOfMemoryError: Java heap space
  at java.util.ArrayList.add(...)
  at org.apache.flink.runtime.state.PartitionableListState.add(PartitionableListState.java:95)
  at org.apache.flink.runtime.state.OperatorStateRestoreOperation.deserializeOperatorStateValues(...)
  at org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(...)

The first response was the obvious one: throw memory at it. TaskManager heap went from a normal setting up to 6 GB, then 10 GB. It helped — briefly — and then the same OOM came back within hours. Whatever was living in that savepoint was growing, and no amount of heap was going to out-run it forever.

A heap dump settled the question. It found the smoking gun: several million instances of KafkaTopicPartition sitting on the heap, and growing further with every restart attempt — by the second attempt, over 14 million instances, chewing through roughly 2.8 GB of heap on their own, before the job had done any work. And KafkaTopicPartition was the tell: that class belongs to the old FlinkKafkaConsumer connector. The jobs in question had all been migrated to the new KafkaSource months earlier. Nothing in the current code should have referenced KafkaTopicPartition at all.

Root cause: a UID that never changed

The migration from FlinkKafkaConsumer to KafkaSource had been done correctly in almost every respect — except one. The connector-upgrade documentation Flink publishes for this exact move tells you to commit offsets on the old consumer, take a savepoint, and start the new source from those committed offsets. What it does not spell out loudly enough is this: if the new source operator keeps the same operator UID as the old consumer, Flink will keep restoring the old consumer's state into it forever — because Flink binds state to UID, not to connector class.

FlinkKafkaConsumer stores its partition offsets in a state called topic-partition-offset-states, holding one Tuple2<KafkaTopicPartition, Long> per partition. KafkaSource has no idea what that state is and never touches it. But it doesn't need to touch it to be hurt by it — Flink's generic operator-state machinery restores it into memory on every restart regardless of whether any code claims it, and then dutifully re-serializes it into the next checkpoint or savepoint. Nothing was pruning it. It was just riding along, unclaimed and immortal.

That alone would explain a leak that stays flat — restore N objects, write back N objects, repeat. What actually happened is worse, because of how that particular state is stored.

Why it multiplies instead of just leaking

Operator state in Flink comes in two redistribution flavors:

• Even-split (getListState) — on restore, Flink deals the saved list out round-robin across subtasks. The total item count is conserved.
• Union (getUnionListState) — on restore, Flink concatenates every subtask's saved list into one combined list and hands a full copy of the whole thing to every subtask. This is how Kafka connectors work, because any subtask needs to be able to see the complete partition-to-offset map.

A live FlinkKafkaConsumer gets away with union state because it immediately throws away everything in that combined list except the partitions it's actually assigned, then snapshots only its own slice. Across all subtasks, the savepoint still holds each partition exactly once. Union mode is safe only because the operator prunes its copy every single cycle.

KafkaSource never claims that state, so nothing prunes it. Which means every restart replays the union mechanics with no pruning step at the end:

1. On restore: union redistribution concatenates all subtasks' saved lists and gives each subtask the full, un-pruned combined list.
2. On snapshot: each subtask writes its entire local list straight back out.

With parallelism P, an operator state of size X becomes P × X after a single restore/snapshot cycle — and it compounds. Starting from a real seed of 2,000 Kafka partitions at parallelism 4:

That table is the entire incident. Every "successful" restart made the next restart's OOM more certain. Raising TaskManager memory bought a few more cycles and nothing else — it never addressed the leak, it just moved the finish line further down a track that only goes one direction. By the time recovery finally stalled out completely, the affected savepoint had grown from a reasonable size to roughly 5 GB, almost entirely KafkaTopicPartition objects nobody's code ever asked for.

Reproducing it on purpose

Reading a postmortem is one thing; watching the failure mode happen on your own laptop is another. We built a small Flink 1.16 DataStream project to reproduce this end to end against a real Kafka cluster (Redpanda), with RocksDB as the state backend — matching the original environment as closely as a homelab allows.

The repro has two stages:

• Stage A — seed the state. Run the legacy FlinkKafkaConsumer job against a topic with 2,000 partitions, let it checkpoint, then stop-with-savepoint. Inspecting that savepoint shows almost exactly 2,000 KafkaTopicPartition entries — the union list at its true, unamplified size.
• Stage B — the amplifier. Resume the migrated KafkaSource job — same source operator UID, --allowNonRestoredState — from that savepoint, then stop-with-savepoint again. Repeat. A custom inspector job (built on Flink's State Processor API) reads the union list out of each new savepoint via an accumulator, so we can watch the count multiply without ever shipping millions of objects back to a client and OOMing our own tooling in the process.

We ran the amplification loop against a TaskManager deliberately capped at a small heap (to reach the OOM in a handful of cycles instead of dozens), and after the fourth resume-and-resnapshot cycle, the savepoint had grown to 56 MB. On the next resume attempt, restore failed — reproducing the same signature seen in production, down to the class and line number:

2026-07-06 16:51:26,290 WARN  ... Source: kafka-source -> parse (2/4)#0 ... switched from
INITIALIZING to FAILED with failure cause: java.lang.OutOfMemoryError: Java heap space
    at java.base/java.util.ArrayList.add(Unknown Source)
    at org.apache.flink.runtime.state.PartitionableListState.add(PartitionableListState.java:95)
    at org.apache.flink.runtime.state.OperatorStateRestoreOperation.deserializeOperatorStateValues(OperatorStateRestoreOperation.java:217)
    at org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(OperatorStateRestoreOperation.java:188)
    at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createOperatorStateBackend(EmbeddedRocksDBStateBackend.java:525)

One detail worth calling out explicitly, because it trips people up: this is a heap OOM despite RocksDB being configured. Operator state — the kind Kafka connectors use for offsets — always lives on the JVM heap via DefaultOperatorStateBackend, never in RocksDB, regardless of what state backend you've configured for keyed state. Pointing a job at RocksDB does nothing to protect it from this failure mode — a RocksDB-backed job can still OOM on heap, on restore, before RocksDB even enters the picture.

The OOM took down the whole TaskManager JVM (exit code 239) — not just the failing task — matching a failure mode that can't be nursed back with more restart attempts, only with a different savepoint or a different fix.

The fix: surgery, not more memory

The real-world remediation, and the one we reproduced, does not try to make the leaked state smaller or slower-growing. It removes it entirely, using Flink's State Processor API — a batch-style API for reading and rewriting savepoints offline, outside of any running job.

The operation is conceptually one line:

SavepointWriter
    .fromExistingSavepoint(input, new EmbeddedRocksDBStateBackend(true))
    .removeOperator("kafka-source")
    .write(output);

removeOperator drops every state associated with that operator UID from the new savepoint it writes — critically, without ever materializing the giant union list in memory. The surgery job reads and writes savepoint metadata; it never has to hold millions of KafkaTopicPartition objects at once, so it doesn't inherit the OOM it's fixing. In our repro, the surgery ran in under a second and took the savepoint from 56 MB down to 276 KB. In production the numbers were predictably bigger — roughly 5 GB down to 3.5 GB, since that savepoint carried several other KafkaSource operators' worth of legitimate state alongside the one leaking. Post-fix heap dumps in both cases showed the same thing: zero KafkaTopicPartition instances.

The last piece is how you resume from the cleaned savepoint. Because the offending state carried Kafka offsets that could no longer be trusted (they'd been frozen at whatever point the legacy consumer last legitimately owned them), the safe choice is to resume with offsets=earliest and --allowNonRestoredState — the source finds no matching state, starts fresh from the earliest available offset, and the job comes up clean. (allowNonRestoredState is required here for an unrelated but easy-to-miss reason too: the KafkaSource itself now also has no split-enumerator state to restore into, since the savepoint predates it ever running.) In our repro this resumed job sat healthy for a full minute with checkpoints completing normally and zero further OOMs — the loop was broken.

A bug we didn't expect to hit

Here's where reproducing an incident earns its keep over just reading about it: we hit a second, unrelated problem that the original incident had also hit, in a form that's easy to gloss over.

The original support engagement notes that running the State Processor API surgery on Flink 1.15 threw:

IllegalStateException: Savepoint must contain at least one operator

— and that this was worked around by getting a patched jar and running the transform under Flink 1.17 instead. The assumption carried into this demo's own README was that Flink 1.16 was unaffected, on the reasoning that as long as the job retains other operators (in our case, a downstream counter and a sink), the savepoint being written is never actually empty, so the check should never fire.

It fired anyway. On Flink 1.16.2, calling .removeOperator(...).write(...) with no other change threw the exact same IllegalStateException, even though the job's other two operators were still present and their state was untouched. Decompiling the bundled State Processor API classes confirmed the message lives in SavepointWriter.writeOperatorStates, and empirically the trigger isn't "does the job have any operators left" — it's whether any new operator transformation was explicitly supplied to the writer via withOperator(...). A remove-only savepoint rewrite, with zero new transformations added, hits the same wall on 1.16 that the original ticket hit on 1.15. The fix that supposedly required jumping all the way to 1.17 turns out to still be needed on 1.16.

The workaround doesn't require a newer Flink version, though — it just requires giving the writer one throwaway operator to satisfy the check:

StateBootstrapTransformation<Integer> marker =
        OperatorTransformation.bootstrapWith(env.fromElements(1))
                .transform(new MarkerBootstrapFunction());   // writes 1 element to list state

SavepointWriter
        .fromExistingSavepoint(input, new EmbeddedRocksDBStateBackend(true))
        .removeOperator("kafka-source")
        .withOperator("surgery-marker", marker)
        .write(output);

The resumed job never sees this marker operator — it doesn't correspond to anything in the real job graph, so --allowNonRestoredState waves it away on resume, exactly like the KafkaSource's own missing state. It exists purely to give SavepointWriter something it's satisfied calling "at least one operator."

The practical takeaway: if you're doing this kind of surgery on Flink 1.15 or 1.16, don't assume retaining other operators in the job graph is enough to dodge the "at least one operator" bug. Test the exact write path you intend to use, or add a marker transformation defensively. It's a five-minute problem if you know to look for it, and a confusing dead end if you don't.

What we'd tell a team doing this migration today

1. Give the new source a new operator UID. This is the actual prevention. State is scoped to UID; if KafkaSource gets a UID FlinkKafkaConsumer never used, there's no old state for it to inherit, ever. The official upgrade guide covers commit-offsets → stop-with-savepoint → start-from-committed-offsets, but doesn't say the word "UID" loudly enough — it's easy to follow the steps precisely and still keep the old UID by default, which is exactly what happened here.
2. A flat-looking memory number can hide compounding growth. The failure here wasn't a slow leak you'd catch by watching a memory graph trend upward over weeks — checkpoints in between restarts looked fine. The growth only showed up at restart boundaries, multiplying by parallelism each time. If your monitoring only samples steady-state heap, this incident is invisible until it's fatal.
3. More memory is a stall, not a fix, for anything that scales with restart count rather than with data volume. If a job needs monotonically more memory to restart with no corresponding change in code or data, that's a structural question, not a sizing question.
4. RocksDB doesn't cover operator state. If your job uses any connector or transformation with operator (non-keyed) state — Kafka source/sink offset tracking is the common case — that state is always on-heap. A RocksDB state backend does not change this, and heap dumps are still the right diagnostic tool even on a job that's "using RocksDB."
5. The State Processor API can save you without a maintenance window. Stripping leaked or incompatible operator state from a savepoint offline, then resuming with allowNonRestoredState: true, is a legitimate production remediation — not just a testing tool. It's also the only tool here that inherently can't inherit the OOM it's fixing, since it never has to materialize the bad state in memory to remove it.
6. If you're on Flink 1.15 or 1.16 and doing remove-only savepoint surgery, add a marker operator. Don't assume other retained operators are enough to avoid the "Savepoint must contain at least one operator" exception — verified reproducible on 1.16.2, contrary to this project's own original assumption.

Where to look if you want to run this yourself

The full reproduction — Maven project, Docker-based Flink 1.16.2 cluster (native arm64 for Apple Silicon), and the driver scripts for every step from seeding the topic through the fix — lives alongside this write-up. README.md walks through the whole thing end to end; the mechanism explanation in this post is adapted from the deeper technical version there.