Kafka Streams stuck rebalancing or restoring state for hours - changelogs, standby replicas and store names

#Symptom

After a rolling deploy, a pod reschedule or a broker incident, a Kafka Streams application stops processing. The state listener reports REBALANCING for minutes or hours, input-topic consumer lag climbs, and nothing is logged except restore progress or repeated group joins.

In almost every case one of two things is happening:

1. State restoration. Tasks were assigned to an instance without usable local state, so Streams is replaying <application.id>-<store>-changelog topics into RocksDB before it processes anything.
2. Rebalance churn. Instances keep leaving and joining the group, so restoration restarts over and over and never finishes.

Work out which one before changing configs.

#Check 1: is it restoring, and how much?

Streams does not expose restore progress through the consumer group (the restore consumer does not commit offsets). Register a StateRestoreListener and a state listener before start(), so this information is in your logs before the next incident:

KafkaStreams streams = new KafkaStreams(topology, props);

streams.setStateListener((newState, oldState) ->
    log.info("Streams state {} -> {}", oldState, newState));

streams.setGlobalStateRestoreListener(new StateRestoreListener() {
  public void onRestoreStart(TopicPartition tp, String store, long start, long end) {
    log.info("Restore start {} {} offsets {}..{} ({} records)", store, tp, start, end, end - start);
  }
  public void onBatchRestored(TopicPartition tp, String store, long batchEnd, long n) {
    log.debug("Restored {} {} up to {}", store, tp, batchEnd);
  }
  public void onRestoreEnd(TopicPartition tp, String store, long total) {
    log.info("Restore end {} {} total {}", store, tp, total);
  }
});
streams.start();

If you have no listener today, measure the changelog size from the broker side. The size of the changelog partitions is roughly what a cold instance has to replay:

kafka-topics.sh --bootstrap-server $BS --list | grep -- '-changelog$'

kafka-log-dirs.sh --bootstrap-server $BS --describe \
  --topic-list orders-app-agg-store-changelog

Tens of gigabytes per partition means a cold restore can take a long time. Restoration on one thread runs at a rate set by RocksDB write speed and the restore consumer's fetch size, not by your processing logic.

#Check 2: is the group churning?

Look for repeated JoinGroup/SyncGroup cycles, members timing out or threads dying:

kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group orders-app --members --verbose

kafka-consumer-groups.sh --bootstrap-server $BS \
  --describe --group orders-app --state

The group id is the application.id. If the member list keeps changing, look at the application logs for these causes:

A liveness probe that kills pods during restoration is the most common cause of restores that never finish: every kill throws away the partly restored store.

#Why the instance had no local state

Streams keeps state in state.dir with a .checkpoint file per task that records the changelog offset the local store matches. A fast restart only replays the gap. A full replay happens when:

• state.dir is ephemeral. The default sits under the JVM temp directory. On Kubernetes, an emptyDir or the container filesystem is lost on reschedule. Use a StatefulSet with a persistent volume per pod and an explicit state.dir.
• Exactly-once and an unclean shutdown. With processing.guarantee=exactly_once_v2, a crash leaves local stores considered dirty, and Streams wipes and rebuilds them from the changelog. Graceful shutdown (streams.close(Duration) on SIGTERM, with enough terminationGracePeriodSeconds) avoids this. Behaviour is changing with transactional state stores in newer releases; check your version's notes.
• Tasks moved to a different instance. Without standbys, the new owner starts cold.
• Store names changed. See below. This one also loses data, not just time.

#The store-name trap

If you don't name stores, Streams generates names from the topology position, for example KSTREAM-AGGREGATE-STATE-STORE-0000000003. Adding or removing an operator upstream shifts the number. The new topology then reads a new, empty changelog and repartition topic, and the old state is orphaned. The app starts "fine" with wrong aggregates.

Name every stateful operator and repartition explicitly:

builder.stream("orders", Consumed.with(Serdes.String(), orderSerde))
  .groupByKey(Grouped.as("orders-by-customer"))
  .aggregate(Totals::new, (k, v, agg) -> agg.add(v),
      Materialized.<String, Totals, KeyValueStore<Bytes, byte[]>>as("customer-totals")
          .withValueSerde(totalsSerde));

Before every deploy, diff topology.describe() output between the old and new build. Any changed store or topic name is a state migration, not a code change.

Warning

Changing a store name, the key, or the serialized value format of a store without a migration plan silently resets or corrupts that state. Plan it like a schema change: new app id or a reset, plus a reprocess from the source topics.

#Fix: make restores rare and short

#Standby replicas and warmups

num.standby.replicas=1
acceptable.recovery.lag=10000
max.warmup.replicas=2
probing.rebalance.interval.ms=600000

• num.standby.replicas=1 keeps a shadow copy of every store on another instance, so failover reuses a near-current store. It doubles local disk and changelog read traffic.
• The assignor treats a client within acceptable.recovery.lag records as caught up. Otherwise it keeps the active task where it is and builds warmup replicas in the background, then moves the task with a probing rebalance. This is why tasks move back some minutes after a scale-out. That is expected behaviour, not churn.

#Make the replay itself faster

• Compacted changelogs (cleanup.policy=compact) only replay the latest value per key, but only if the log cleaner keeps up. A changelog full of uncompacted segments replays every historical update. Check the cleaner and min.cleanable.dirty.ratio on large changelogs.
• Raise restore fetch sizes with the restore.consumer. prefix, for example restore.consumer.max.poll.records=5000.
• Spread restoration over more threads with num.stream.threads. Restoration is per task, so more threads (up to the number of tasks per instance) restore in parallel.
• Windowed stores use compact,delete. Keep window retention no longer than you actually query.

#Bound RocksDB memory

Unbounded RocksDB off-heap memory gets containers OOMKilled during restores, because bulk loading fills memtables fast. Implement RocksDBConfigSetter with a shared LRUCache and WriteBufferManager, register it with rocksdb.config.setter, and size the container for heap plus that bound.

#When you need a clean rebuild

If state is corrupt or a store's format changed, stop all instances, then:

kafka-streams-application-reset.sh --bootstrap-server $BS \
  --application-id orders-app --input-topics orders --to-earliest --dry-run
# review, then run again without --dry-run

The tool resets input offsets and deletes internal (changelog and repartition) topics. It does not touch local state, so call streams.cleanUp() before start() on each instance, or delete state.dir.

Caution

The reset deletes changelog topics. If the state can't be rebuilt from input topics (input retention shorter than the state's history), this is permanent data loss. Check input-topic retention first.

#Prevent it next time

• Persistent state.dir per instance, graceful shutdown on SIGTERM, liveness probes that tolerate REBALANCING.
• Name every store and repartition; diff topology.describe() in CI.
• num.standby.replicas=1 for any store that takes more than a few minutes to restore.
• Log restore start and end per store, and alert on time spent in REBALANCING.
• Recent 4.x releases add a broker-side Streams rebalance protocol (group.protocol=streams); check its maturity in your version before using it in production.