[
https://issues.apache.org/jira/browse/KAFKA-21034?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133888#comment-18133888
]
이선민 edited comment on KAFKA-21034 at 10/29/26, 12:49 AM:
---------------------------------------------------------
[~zoro30102000] Thanks for the detailed report. This matches the failure path
addressed by my PR: under EOS, RocksDBStore#openDB wraps the invalid-state
failure in TaskCorruptedException, which is not caught by the
ProcessorStateException handler in GlobalStateManagerImpl#initialize, so the
existing wipe is skipped for this path.
My open PR, https://github.com/apache/kafka/pull/23390, handles this in
GlobalStreamThread#initialize alongside InvalidOffsetException, closing the
state consumer with wipeStateStore=true so the global store can be rebuilt on
the next start. Review is ongoing, with discussion about where the cleanup
should live and whether the existing cleanup paths should be unified.
Please share the reproduction; I'd like to validate the PR against it. If
you're able to test the patch in your setup as well, that would be helpful.
I'll also raise the request for backports to 4.3.x and 4.4.x with the
committers.
was (Author: JIRAUSER314548):
[~zoro30102000] Thanks for the detailed report. This matches the failure path
addressed by my PR: under EOS, RocksDBStore#openDB wraps the invalid-state
failure in TaskCorruptedException, which is not caught by the
ProcessorStateException handler in GlobalStateManagerImpl#initialize, so the
existing wipe is skipped for this path.
My open PR, https://github.com/apache/kafka/pull/23390, handles this in
GlobalStreamThread#initialize alongside InvalidOffsetException, closing the
state consumer with wipeStateStore=true so the global store can be rebuilt on
the next start. Review is ongoing, with discussion about where the cleanup
should live and whether the existing cleanup paths should be unified.
> Wipe global store on start up, when corrupted
> ---------------------------------------------
>
> Key: KAFKA-21034
> URL: https://issues.apache.org/jira/browse/KAFKA-21034
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.3.0, 4.3.1, 4.4.0, 4.3.2
> Reporter: Uladzislau Blok
> Assignee: 이선민
> Priority: Major
>
> *Problem statement:*
> Currently, Kafka Streams provides an automated recovery mechanism for
> standard state stores: when a {{StreamThread}} starts up, it can wipe a
> corrupted local store and rebuild it from its changelog.
> However, this self-healing behavior is not supported for global state stores.
> Implementing dynamic runtime recovery for global stores would require complex
> thread synchronization—namely, pausing all stream threads until the global
> store has completely rebuilt. Streams simply lets the application fail fast
> and crash
> Because Streams does not wipe a corrupted global store before crashing, the
> corrupted data remains on disk. On subsequent restarts, the application
> encounters the same corrupted state and crashes again. In environments with
> automated restart policies (such as Kubernetes StatefulSets), this leads to
> an unrecoverable {{{}CrashLoopBackOff{}}}, requiring manual intervention to
> exec into the pod and delete the local directory.
> Instead of attempting complex runtime thread synchronization, we can handle
> this purely at the startup/initialization phase:
> # Catch the store corruption exception when the global state store
> initializes.
> # Automatically wipe the corrupted global store directory on disk.
> # Terminate or fail the current startup sequence cleanly.
> On the next restart the application will start with an empty store directory
> and restore the global store from scratch, breaking the crash loop without
> manual intervention.
>
> *Mechanics:*
> We already have similar mechanism:
> {{GlobalStreamThread#initialize}} -> {{stateConsumer#initialize}} -> throws
> InvalidOffsetException -> {{{}closeStateConsumer(stateConsumer, true){}}},
> where true means wipe state store
> In other cases:
> StreamsException -> {{closeStateConsumer(stateConsumer, false)}}
> Throwable -> {{closeStateConsumer(stateConsumer, false)}}
>
> *Potential solution:*
> Catch exceptions of other types (exact list to be analyzed). I saw it for
> TaskCorruptedException(ProcessorStateException) ({{{}RocksDBStore#openDB ;
> AbstractColumnFamilyAccessor#open{}}})
>
> *Open questions:*
> * Do we need KIP for this change?
> * Exact exception (or list of exceptions to trigger wiping)
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
