[jira] [Assigned] (KAFKA-21034) Wipe global store on start up, when corrupted

["이선민 (Jira)" <ji...@apache.org>]

     [ 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)