[
https://issues.apache.org/jira/browse/KAFKA-21034?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
이선민 reassigned KAFKA-21034:
---------------------------
Assignee: 이선민
> Wipe global store on start up, when corrupted
> ---------------------------------------------
>
> Key: KAFKA-21034
> URL: https://issues.apache.org/jira/browse/KAFKA-21034
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Uladzislau Blok
> Assignee: 이선민
> Priority: Minor
>
> *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
(v8.20.10#820010)
