[
https://issues.apache.org/jira/browse/KAFKA-21034?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133474#comment-18133474
]
Eswarar Siva commented on KAFKA-21034:
--------------------------------------
We ran into this on 4.3.2 and reproduced it, in case the details help with the
priority.
Setup: {{processing.guarantee}} {{exactly_once_v2}}, one persistent (RocksDB,
timestamped) global store, and a {{state.dir}} that is kept across restarts (a
Kubernetes StatefulSet with a volume).
What we see: once the global store has flushed a memtable after it was opened,
the "open" status in its offsets column family is on disk. A SIGKILL from then
on (an OOM kill, or a kill after the shutdown grace period) leaves that status
at 1, and every later start fails with {{TaskCorruptedException}}: "Tasks
\[-1_-1\] are corrupted ... State store ... didn't find a valid state, since
under EOS ... Invalid state during store open". The global thread dies and the
"Detected unclean shutdown for global store" wipe in
{{GlobalStateManagerImpl#initialize}} never runs, as described above. With
{{at_least_once}} the same kill restarts fine, and so does a kill after a clean
restart before the store's first flush. So for an {{exactly_once_v2}}
application with a persistent global store, a hard kill after that first flush
ends in a crash loop until the global folder is deleted by hand.
As far as I can tell this came in with 4.3.0 (4.2 has no status key), and 4.3.2
and 4.4.0 both have it, so maybe those could go in as affected versions. A
backport of the fix to 4.3.x and 4.4.x would help us a lot, if that is possible.
Happy to share the small reproduction if it is useful.
> 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
(v10.3.26#10030026)
