Aangbaeck opened a new pull request, #22731: URL: https://github.com/apache/kafka/pull/22731
## Summary [KAFKA-13009](https://issues.apache.org/jira/browse/KAFKA-13009) Adding a `GlobalKTable` to an application and then changing the topology (for example removing a subtopology) can crash Kafka Streams on the next restart with: ``` java.lang.IllegalStateException: Metrics recorder is re-initialised with different task: previous task is -1_-1 whereas current task is 0_1. This is a bug in Kafka Streams. ``` This was reported independently on 2.7.x (2021) and again on 4.1.1 (2026). The only known workaround is to delete internal changelog topics / local state. ## Root cause A global state store is a **single shared instance** managed by the global state manager (`GlobalStateManagerImpl`), initialised once under the global task id `-1_-1`. Subtopology ids are positional. When a subtopology is removed, the remaining node groups are renumbered, so a left-over on-disk task directory's subtopology id can now map onto the global store's node group. `InternalTopologyBuilder#buildProcessorNode` unconditionally put a referenced global store into the per-task local state-store map (`stateStoreMap`). So when `buildSubtopology(id)` was called for such a stale/renumbered task id, the returned `ProcessorTopology` contained the shared global-store instance as a **local** store. Opening it as a per-task store re-initialised the shared `RocksDBMetricsRecorder` under a second task id and threw. The startup-store pre-open optimisation (`StateDirectory#initializeStartupStores`, which scans on-disk task directories) makes this trivial to hit and is why the issue resurfaced: it opens the same shared global store once per left-over partition directory of the removed subtopology. The standby (`StandbyTaskCreator`) and active (`ActiveTaskCreator`) task paths share the same flaw — the standby path is the one in the original 2021 report. The global store also passes `hasStateWithChangelogs()`, because a global store registers its source topic as a pseudo-changelog (used for restore), so none of these paths skipped it. ## Fix Do not add global state stores to the per-task local state-store list in `InternalTopologyBuilder#buildProcessorNode`. Global stores are already exposed to the `ProcessorTopology` through the dedicated `globalStateStores` list (which is what `GlobalStateManagerImpl` consumes), so nothing that reads a task subtopology needs them in the local list. With the store no longer in the local list, a subtopology that maps onto the global node group reports no local state with changelogs, and the startup, standby and active task paths all correctly skip it. The global store continues to be initialised exactly once by the global state manager. This is a one-line change plus a comment; `hasPersistentStores` is unaffected (the persistence check still runs). ## Testing - New unit test `InternalTopologyBuilderTest#shouldNotAddGlobalStoreAsTaskLocalStateStore`: a global store appears only in `globalStateStores()`, never in a task subtopology's `stateStores()`. - New integration test `GlobalStoreTopologyChangeIntegrationTest#shouldNotCrashWhenSubtopologyRemovedWhileGlobalStoreRemains`: reproduces the reported scenario (two regular subtopologies + a global table, then the first subtopology removed, restarted with the same application id and state directory). It **fails without the fix** (`IllegalStateException: ... re-initialised with different task`) and **passes with it**. - The full `streams` unit test suite, the global-store integration tests, and the topology / task-creation / `StateDirectory` suites pass. `checkstyle` and `spotbugs` pass. ## Scope This fixes the crash reported in KAFKA-13009. It does not change the fact that subtopology ids are positional and can be remapped by an incompatible topology change — addressing that underlying instability is a separate, larger effort. Instead it enforces the invariant that a global store is never a per-task local store, which prevents the crash regardless of how a task's subtopology id happens to map after such a change. ## Compatibility No public API or behaviour change for valid topologies; this only removes an internal registration that a task could never use correctly. No KIP required. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
