[
https://issues.apache.org/jira/browse/KAFKA-21034?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Uladzislau Blok updated KAFKA-21034:
------------------------------------
Description:
*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)
> 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
> 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)
