Kafka Streams TaskCorruptedException: Fix and Prevent It

TaskCorruptedException is the one Streams error that scares people, because the recovery path literally deletes your local state directory. It is not a bug and, most of the time, it is not data loss — it is Kafka Streams choosing consistency over your RocksDB cache. But when it fires in a hot loop, it will pin CPU on changelog restore, stall processing for minutes, and shred your p99. This is the fifth Streams-category article in the series, and it is mechanically distinct from the four you have already seen: it is not a lifecycle/IQ problem ([InvalidStateStoreException](kafka-invalidstatestoreexception-state-store-migrated.md)), not a file-descriptor problem ([RocksDB "Too many open files"](kafka-streams-rocksdb-too-many-open-files-fix.md)), not a state-directory lock ([LockException`](kafka-streams-lockexception-failed-to-lock-state-directory-fix.md)), and not a SerDe mismatch (serializer not compatible). This one is about the consistency contract between your local state and the changelog topic.

1. The Error

The stack trace you pasted into Google:

org.apache.kafka.streams.errors.TaskCorruptedException: Tasks [0_2, 0_1] are corrupted and hence needs to be re-initialized
    at org.apache.kafka.streams.processor.internals.StreamTask.commitNeeded(StreamTask.java:...)
    at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:...)
    at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:...)
    at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:...)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnceWithoutProcessingThreads(StreamThread.java:...)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:...)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:...)

Just as often it arrives wrapped around an offset problem on restore:

org.apache.kafka.streams.errors.TaskCorruptedException: Tasks [0_0] are corrupted and hence needs to be re-initialized
Caused by: org.apache.kafka.clients.consumer.OffsetOutOfRangeException:
    Fetch position FetchPosition{offset=48213, ...} is out of range for partition my-app-my-store-changelog-0

Followed, in the logs, by the line that makes people panic:

Deleting obsolete state directory 0_0 for task 0_0 as cleanup delay of 600000 ms has passed
Wiping state stores for corrupted task 0_0
Reinitializing StreamTasks: [0_0]

This is Kafka Streams (kafka-streams 2.6 through 4.0), typically with processing.guarantee=exactly_once_v2. It shows up in Docker/Testcontainers POCs, on Kubernetes during rolling restarts, and in production during broker elections. The broker version is largely irrelevant — this is a client-side consistency mechanism inside the Streams runtime.

2. How to Reproduce It (step-by-step)

The cleanest reproduction is EOS + a broker-side transaction timeout during commit. Bring up a single-broker KRaft cluster:

# docker-compose.yml
services:
  kafka:
    image: apache/kafka:3.9.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:29093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      # single broker: internal topics must be RF=1 or the app never starts
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

A minimal stateful topology with EOS and an aggressively short transaction timeout:

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "corrupt-demo");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);
// Force the transaction to time out before the commit round-trip finishes.
props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 100);

StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic")
       .groupByKey()
       .count(Materialized.as("counts-store"))   // RocksDB + changelog
       .toStream()
       .to("output-topic");

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Drive traffic through it:

docker exec -it $(docker ps -qf name=kafka) \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic input-topic \
  --property "parse.key=true" --property "key.separator=:"
# type: a:1  b:2  a:3 ...

With transaction.timeout.ms=100, the EOS commit path cannot finish the two-phase commit inside the window. The broker aborts the transaction, the producer's commitTransaction() surfaces a TimeoutException, and Streams converts it into TaskCorruptedException for the affected tasks. You will see the wipe-and-restore cycle repeat on every commit interval.

The second, more common production trigger — changelog OffsetOutOfRange — reproduces by aggressively compacting/truncating the changelog (or deleting and recreating the store's changelog topic) while a checkpoint file on disk still references an offset that no longer exists. On the next restore, the fetch position is out of range and the task is marked corrupted.

3. Why It Happens — Surface Level

TaskCorruptedException means Kafka Streams detected that a task's local state can no longer be trusted to match its changelog, so it throws the state away and rebuilds it from the changelog topic — the durable source of truth.

Two immediate causes dominate. First, under EOS a transaction that wrote to the changelog times out or is aborted during commit: the local RocksDB store already applied the records, but the changelog commit did not land atomically, so local and remote diverge. Since Kafka Streams 2.7 (KAFKA-9274), the runtime deliberately throws TaskCorruptedException instead of bubbling up the raw TimeoutException, precisely so it can wipe and re-restore rather than continue on inconsistent state. Second, on restore the consumer's position is out of range for the changelog partition — the changelog was compacted, truncated, or recreated, and the checkpointed offset points at data the broker no longer has.

4. Why It Happens — Under the Hood

The state store and the changelog form a write-ahead-log pair. Every update to a RocksDB store is also produced to a compacted ${application.id}-${storeName}-changelog topic. A checkpoint file on disk records the changelog offset up to which local state is known-good. On restart, Streams reads the checkpoint and replays the changelog from that offset forward to warm the store.

Under exactly_once_v2, that produce-to-changelog is part of the same Kafka transaction as your output records and consumer offset commits. EOS-v2 commits all tasks of a StreamThread together through a single transactional producer per thread (that is the "v2" optimization over the old producer-per-task model). The commit is two-phase: AddOffsetsToTxn/AddPartitionsToTxn, then EndTxn. If EndTxn does not complete within transaction.timeout.ms (bounded on the broker by transaction.max.timeout.ms), the transaction coordinator aborts it. Now you have a hard inconsistency: RocksDB mutated in memory/on disk, but the changelog records that would have made those mutations durable were rolled back. There is no safe way to "un-apply" the RocksDB writes cheaply, so Streams marks the task corrupted.

Recovery is deterministic. TaskManager closes the corrupted task as dirty — under EOS that means wipeStateStores() deletes the store's on-disk directory — and revives the task into the CREATED state. The next poll cycle drives it through restoration: the changelog is replayed from the last durable committed offset to the log end, rebuilding RocksDB from the source of truth. Because the changelog is the transactional record and the local store was the disposable cache, throwing away the cache is always correct. It is just expensive.

The OffsetOutOfRangeException variant is the same contract from the other side. If the checkpoint says "resume changelog restore at offset 48213" but retention/compaction moved the log start offset past that (or a topic recreate reset it), the fetch position is invalid. Streams cannot trust a partial restore against a moved log, so it corrupts the task and restores from a valid position. In modern Streams (2.6+), restoration runs on a dedicated StateUpdater/restore thread (KAFKA-10199), and a subtlety fixed in KAFKA-14299 was that a TaskCorruptedException thrown during initialization could leave a task un-owned by both the StreamThread and the StateUpdater — patched so the revive path is consistent.

A third trigger worth knowing: a ProducerFencedException/InvalidProducerEpochException during the EOS commit (a zombie instance after a rebalance) also routes through TaskCorruptedException for the fenced tasks. If you see this together with rebalances, treat it as a churn problem, not a broker problem.

5. The Fix

Match the fix to the trigger. The single most common self-inflicted cause is a transaction timeout that is too tight for real commit latency.

Before — a POC config (or a copied blog snippet) that starves the commit:

// BEFORE — transaction times out under normal broker latency / GC pauses
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 100);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);   // commit storm

After — give the transaction room, and stop committing so aggressively:

// AFTER — transaction budget comfortably exceeds real commit latency
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
// EOS default commit interval is 100ms in older versions; raise it deliberately.
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30_000);
// transaction.timeout.ms must be >= commit.interval.ms + max processing time per commit.
props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 60_000);
// Give producer retries/acks time before a commit-time timeout can even occur.
props.put(StreamsConfig.producerPrefix(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG), 120_000);

The invariant: transaction.timeout.ms must exceed commit.interval.ms plus the worst-case time to process and flush a commit, and it must be less than or equal to the broker's transaction.max.timeout.ms (default 900000 ms / 15 min). If your commit timed out because the broker was genuinely slow — under-replicated __transaction_state, an ISR shrink, or a leader election — the real fix is on the broker: keep internal topics at RF=3 with min.insync.replicas=2, and stop commits from racing elections.

For the OffsetOutOfRange variant, do not treat wipe-and-restore as the fix if it loops. Find who is truncating the changelog: an over-aggressive cleanup.policy, a topic recreated by an automation job, or a manual kafka-topics.sh --delete on a *-changelog topic. Changelog topics are cleanup.policy=compact by default and should never be deleted out from under a running app. If the checkpoint is stale after a legitimate topic reset, a one-time KafkaStreams.cleanUp() before start() clears local state so restore begins from a valid position — but only run it deliberately, never on every boot.

If the corruption is driven by fencing during rebalances, the fix is the standard churn toolkit: static membership and cooperative rebalancing so healthy instances stop getting fenced (see the rebalance storm and static membership articles).

6. Best Practices & The Better Design

The better design accepts that under classic EOS, TaskCorruptedException is a correct-but-costly recovery, and minimizes both its frequency and its blast radius.

Keep state stores small enough that a restore is cheap, and add standby replicas so a wipe does not mean a cold rebuild:

// A warm standby means recovery restores from a near-current replica, not from zero.
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 60_000);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30_000);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
// Bound restore cost / smooth rebalances.
props.put(StreamsConfig.MAX_WARMUP_REPLICAS_CONFIG, 2);

The fundamentally better answer, and the direction Kafka is moving, is transactional state stores (KIP-892), which land as the default in Kafka Streams 4.x. With transactional stores, local RocksDB writes are staged and made durable in lockstep with the changelog commit, so a commit timeout no longer leaves local and remote diverged. The runtime can roll back the local staged writes instead of wiping the entire store — eliminating the expensive re-restore for the timeout case entirely. If you are on 4.0+, prefer the transactional-store path over hand-tuning transaction timeouts:

// Kafka Streams 4.x: transactional state stores make the timeout-driven wipe unnecessary.
// (Default under EOS in 4.0+. Verify against your exact minor version.)
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

If you do not actually need exactly-once — a lot of "we turned on EOS because it sounded safer" pipelines do not — running at_least_once with idempotent downstream writes sidesteps the entire transactional-commit corruption class. Reserve EOS for genuine read-process-write atomicity requirements.

7. How to Prevent It Long-Term

Alert on the recovery, not just the crash. TaskCorruptedException is logged and handled internally, so it will not always kill the app — it will silently burn restore bandwidth. Watch the Streams metrics task-created-rate and restore-total/restore-records-rate climbing without deployments, and the consumer restore-consumer lag. A steady trickle of corruptions with flat traffic is the fingerprint of a too-tight transaction timeout or a churn loop.

On the broker side, keep __transaction_state and __consumer_offsets at RF=3 / min.insync.replicas=2, and alert on UnderReplicatedPartitions and LeaderElectionRateAndTimeMs — slow transaction commits driven by ISR trouble are a leading cause of commit-time timeouts. Never let automation delete or repartition *-changelog or *-repartition topics; treat them as app-owned internal topics and exclude them from topic-cleanup jobs. In CI, run a Testcontainers test that processes records through the EOS topology, forces a restart mid-stream, and asserts the counts are correct after recovery — that catches both the timeout tuning and any changelog-retention misconfiguration before production. Finally, set an uncaught-exception handler so an escalated corruption replaces the thread instead of quietly degrading:

streams.setUncaughtExceptionHandler(ex ->
    StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD);

8. Key Takeaways / Learnings

• TaskCorruptedException is Kafka Streams choosing consistency over your cache: it wipes local state (under EOS) and re-restores from the changelog, the durable source of truth. It is usually not data loss.
• The two dominant triggers are an EOS transaction timeout/abort during commit (KAFKA-9274 turns the TimeoutException into corruption on purpose) and a changelog OffsetOutOfRange on restore.
• Fix the timeout variant with the invariant commit.interval.ms + processing time ≤ transaction.timeout.ms ≤ transaction.max.timeout.ms; fix the offset variant by stopping whatever truncates/recreates the changelog.
• Add num.standby.replicas=1 so recovery restores from a warm replica, and keep internal topics at RF=3 / min.insync.replicas=2 so commits do not race elections.
• On Kafka Streams 4.x, transactional state stores (KIP-892) make the timeout-driven wipe obsolete — prefer them over hand-tuning; and if you do not truly need EOS, at_least_once avoids this class entirely.