> ## Content Index
> Fetch the complete content index at: https://www.ggorantala.dev/llms.txt
> Use this file to discover other available public pages before exploring further.

# Kafka InvalidStateStoreException: State Store May Have Migrated
- URL: https://www.ggorantala.dev/kafka-invalidstatestoreexception-state-store-migrated/
- Published: 2026-05-19T15:25:00.000Z
- Updated: 2026-08-29T08:31:22.000Z
- Description: Fixing InvalidStateStoreException: "the state store may have migrated to another instance" in Kafka Streams — root causes, retry logic, and the right IQ design.
- Author: Gopi Gorantala
- Tags: kafka-errors, invalidstatestoreexception, kafka-streams, interactive-queries, state-store, rocksdb, Java

## 1\. The Error

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

```text
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:

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

```

Or, mid-rebalance:

```text
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:

```yaml
# 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

```

```bash
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:

```java
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:

```diff
- 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:

```java
@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`:

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

```

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

```properties
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:

```properties
application.server=${POD_IP}:8080

```

```java
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:

```java
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