Kafka InvalidStateStoreException: State Store May Have Migrated

1. The Error

You exposed a Kafka Streams state store through an interactive query (IQ) endpoint, hit it with curl, and got:

org.apache.kafka.streams.errors.InvalidStateStoreException: The state store, word-counts, may have migrated to another instance.
	at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:75)
	at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1929)
	at com.example.CountsController.getCount(CountsController.java:31)

Or the startup-time variant:

org.apache.kafka.streams.errors.StreamsNotStartedException: KafkaStreams is not running. State is CREATED.

Or, mid-rebalance:

org.apache.kafka.streams.errors.StreamsRebalancingException: KafkaStreams is currently rebalancing and the requested state store, word-counts, is not available for querying.

All three are InvalidStateStoreException — since Kafka 2.5 (KIP-216) the generic message was split into typed subclasses: StreamsNotStartedException, StreamsRebalancingException, StateStoreMigratedException, StateStoreNotAvailableException, UnknownStateStoreException. Older posts and older clients show only the generic "may have migrated" wording for every one of these situations, which is exactly why this error confuses people: five different problems share one exception hierarchy.

Typical setup: kafka-streams 3.x (anything 2.5+ behaves as described), Java 17/21, Spring Boot with spring-kafka's StreamsBuilderFactoryBean or a plain KafkaStreams app, local Docker broker or managed Kafka. This is a client-side exception — the broker is fine. Your query arrived at a moment (or an instance) where the store isn't locally queryable.

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

Single-broker KRaft cluster:

# docker-compose.yml
services:
  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://:19092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:19092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

docker compose up -d
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:19092 --create --topic words --partitions 4

Minimal Streams app (kafka-streams 3.8.0) with a materialized count and a query method:

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

StreamsBuilder builder = new StreamsBuilder();
builder.stream("words", Consumed.with(Serdes.String(), Serdes.String()))
       .groupBy((k, v) -> v)
       .count(Materialized.as("word-counts"));

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

// Query IMMEDIATELY — this is the reproduction
ReadOnlyKeyValueStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType("word-counts",
        QueryableStoreTypes.keyValueStore()));

Run it. The store() call executes a few milliseconds after start(), while the app is still in REBALANCING — you get the exception every time. That's reproduction #1: querying before the app reaches RUNNING.

Reproduction #2 — the multi-instance variant: start two copies of the app (same application.id, different state.dir). The 4 partitions of the changelog split 2/2 between instances. Query instance A for a key whose task lives on instance B — with the pre-KIP-535 default, A throws "may have migrated" for keys it doesn't host. Then kill -9 instance B: while its tasks migrate to A, every query on A fails with StreamsRebalancingException until state restoration from the changelog completes — seconds to minutes depending on store size.

Environment-specific triggers: only under instance churn (deploys, pod evictions, autoscaling), only on multi-instance deployments, and reliably in Kubernetes readiness probes that hit the IQ endpoint before the app is RUNNING — which makes the pod never become ready, which stalls the rollout, which keeps the group rebalancing. A genuinely self-sustaining failure.

3. Why It Happens — Surface Level

KafkaStreams#store() only hands you a store wrapper when (a) the app instance is in a queryable state and (b) at least one active (or, if you allow it, standby) copy of that store exists on this instance. State stores are not cluster-wide services. Each store is sharded per task, tasks are assigned to instances by the consumer group protocol, and every rebalance can move them. If the store shard you want is elsewhere — or is being rebuilt right now — the call throws instead of silently returning stale or missing data.

4. Why It Happens — Under the Hood

A materialized KTable is physically N RocksDB instances (one per input partition / task) on local disk under state.dir, each backed by a compacted changelog topic (<application.id>-word-counts-changelog). The task assignment decides which instances host which shards; the changelog is what makes a shard rebuildable anywhere.

The query path: KafkaStreams.store() → QueryableStoreProvider.getStore() → per-thread StreamThreadStateStoreProvider.stores(). That provider checks two gates. First, the KafkaStreams state machine: CREATED → REBALANCING → RUNNING (with PENDING_SHUTDOWN/NOT_RUNNING/ERROR off to the side). Queries are only served in RUNNING (and, for stale-tolerant queries since KIP-535, also REBALANCING). Second, task state: a store is only exposed once its task is fully restored — during restoration the store is receiving changelog replay and is not consistent.

Why rebalances make this so visible: when an instance dies, its tasks are reassigned and the new owner must replay the changelog into fresh RocksDB instances before those shards are queryable. With no standbys, a 20 GB store means the shard is dark for the whole replay. KIP-441 (Kafka 2.6) improved this — the assignor now keeps tasks on instances with warm state and moves them only after a warm-up copy catches up — but a hard-killed instance with no standby replica still means full restoration on a cold node.

The multi-instance case isn't a failure at all — it's routing. Streams partitions by key: key → partition → task → instance. Instance A fundamentally cannot answer for keys hashed to instance B's tasks. The API tells you where to go — streams.queryMetadataForKey("word-counts", key, Serdes.String().serializer()) returns the HostInfo (from each instance's application.server config) of the active owner — but you own the RPC hop between instances. There is no built-in scatter-gather.

One more trap: num.standby.replicas defaults to 0. No standbys means no warm failover copies and (even with enableStaleStores()) nothing else to query during restoration.

5. The Fix

Three fixes, matched to the three reproductions.

Fix 1 — never query before RUNNING. Gate your endpoint on the state machine instead of calling store() blind:

- KafkaStreams streams = new KafkaStreams(builder.build(), props);
- streams.start();
- ReadOnlyKeyValueStore<String, Long> store = streams.store(...); // throws
+ KafkaStreams streams = new KafkaStreams(builder.build(), props);
+ CountDownLatch running = new CountDownLatch(1);
+ streams.setStateListener((newState, old) -> {
+     if (newState == KafkaStreams.State.RUNNING) running.countDown();
+ });
+ streams.start();
+ running.await();   // or expose isRunning() to your readiness probe

In Spring Boot, wire the same listener through the factory bean and use it for the readiness probe — do not point the probe at the IQ endpoint:

@Bean
StreamsBuilderFactoryBeanConfigurer stateListener(ApplicationAvailability unused,
        ApplicationEventPublisher publisher) {
    return fb -> fb.setStateListener((newState, old) -> {
        AvailabilityChangeEvent.publish(publisher, this,
            newState == KafkaStreams.State.RUNNING
                ? ReadinessState.ACCEPTING_TRAFFIC
                : ReadinessState.REFUSING_TRAFFIC);
    });
}

Fix 2 — tolerate rebalances instead of erroring. Since KIP-535 (Kafka 2.5), enableStaleStores() lets a query be served from a standby or a restoring active, during REBALANCING:

  streams.store(
      StoreQueryParameters.fromNameAndType("word-counts",
-         QueryableStoreTypes.keyValueStore()))
+         QueryableStoreTypes.keyValueStore())
+         .enableStaleStores())

Pair it with standbys, or there's nothing stale to read:

num.standby.replicas=1
acceptable.recovery.lag=10000

Use this when slightly stale reads are acceptable (dashboards, caches, feature lookups). Don't use it when the read must reflect all committed input (billing, dedup decisions) — there, catch StreamsRebalancingException and return HTTP 503 with Retry-After, and let the caller retry.

Fix 3 — route multi-instance queries to the right host. Set application.server on each instance and forward:

application.server=${POD_IP}:8080

KeyQueryMetadata meta = streams.queryMetadataForKey(
    "word-counts", key, Serdes.String().serializer());

if (meta.equals(KeyQueryMetadata.NOT_AVAILABLE)) {
    throw new StoreUnavailableException();          // 503, retry
}
if (isLocal(meta.activeHost())) {
    return localLookup(key);
}
return httpClient.get("http://" + meta.activeHost().host()
        + ":" + meta.activeHost().port() + "/counts/" + key);

Which fix when: a single-instance POC needs only Fix 1. A production multi-instance deployment needs all three — readiness gating, standbys with stale-tolerant reads for availability, and key routing for correctness.

6. Best Practices & The Better Design

The right-way IQ service wraps every store access in one helper that encodes the whole contract — state gate, stale tolerance, migration retry:

public class StoreQueryService {

    private final KafkaStreams streams;

    public <T> T withStore(String name, Function<ReadOnlyKeyValueStore<String, Long>, T> fn) {
        if (streams.state() != KafkaStreams.State.RUNNING
                && streams.state() != KafkaStreams.State.REBALANCING) {
            throw new ServiceUnavailableException("streams state: " + streams.state());
        }
        try {
            var store = streams.store(
                StoreQueryParameters.fromNameAndType(name,
                    QueryableStoreTypes.<String, Long>keyValueStore())
                .enableStaleStores());
            return fn.apply(store);
        } catch (InvalidStateStoreException e) {
            // store handle went stale mid-query (task migrated) — safe to retry
            throw new ServiceUnavailableException(e.getMessage());
        }
    }
}

Note the catch around the use of the store, not just its acquisition — a store handle can be invalidated by a migration between store() and get(), and RocksDB access then throws InvalidStateStoreException too. Treat every InvalidStateStoreException as retriable-with-backoff at the edge (503 + Retry-After), never as a 500.

Beyond the helper, the bigger design decisions:

Question deep IQ dependence. Interactive queries make each app instance a stateful database node with cache-like availability semantics. If your read path needs strong availability guarantees, the more robust pattern is to keep Streams as the processor and sink results to an external store (.toStream().to("word-counts-out") → Connect sink → Postgres/Redis), where reads don't care about rebalances. IQ shines for co-located, latency-critical lookups; it's a poor general-purpose query API.

Standbys are not optional in production. num.standby.replicas=1 turns "store dark for a full changelog replay" into "store served stale for seconds." Budget the disk: standbys double state footprint.

Static membership for rolling restarts. group.instance.id per pod plus a session.timeout.ms larger than your restart window means a bounced pod rejoins without triggering task migration at all — the store never leaves.

IQv2 (KIP-796, Kafka 3.2+). The newer KafkaStreams#query(StateQueryRequest) API returns per-partition results with positions instead of throwing on partial availability, and lets you query specific partitions. For new IQ code it's the better foundation; the failure modes in this article still apply, but the API surfaces them as per-partition results rather than a single exception.

7. How to Prevent It Long-Term

Monitor the state machine, not just the endpoint: alert on KafkaStreams.state() != RUNNING for more than your restoration SLO, and on the kafka.streams metrics restore-latency-max / restore-records-rate (restoration listeners since KIP-869 give per-store progress). Track consumer-group rebalance rate for the <application.id> group — a rebalance storm here is an IQ outage.

Convention-wise: every Streams service in the org ships with num.standby.replicas>=1, group.instance.id set, readiness bound to the state listener (never the IQ endpoint), and an IQ edge that maps InvalidStateStoreException to 503. Codify it in your platform starter/library so teams can't forget.

Test it before production does: a Testcontainers integration test that starts two app instances, kills one, and asserts the IQ endpoint returns 503-then-recovers (rather than 500 or wrong data) catches every regression in this class. Chaos-test restoration time against your real store size — the changelog replay rate, not your code, sets your failover SLO. And capacity-plan state.dir disk for actives plus standbys plus one warm-up copy (KIP-441 moves them around during rebalances).

8. Key Takeaways / Learnings

• InvalidStateStoreException is five situations in one hierarchy: not started, rebalancing, migrated, closed, or wrong store name. Since Kafka 2.5 (KIP-216) the subclass tells you which — read it before fixing.
• State stores are per-task local shards, not a cluster service. An instance can only answer for keys it hosts; use queryMetadataForKey() + application.server to route, because Streams won't do the RPC for you.
• Never query before RUNNING, and never point a readiness probe at the IQ endpoint — that combination self-sustains rebalance outages.
• num.standby.replicas=1 + enableStaleStores() is the difference between seconds of stale reads and minutes of downtime during failover.
• Treat every InvalidStateStoreException as retriable at the edge (503 + Retry-After). If reads need database-grade availability, sink to an external store instead of leaning on IQ