[
https://issues.apache.org/jira/browse/KAFKA-21034?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18112003#comment-18112003
]
이선민 commented on KAFKA-21034:
-----------------------------
[~bloku] I would like to work on this, if nobody has started yet.
While reading the code I found that the wipe already exists for this case in
{{GlobalStateManagerImpl#initialize}}, but it only runs when EOS is enabled:
{code:java}
try {
maybeWrappedStore.init(globalProcessorContext, maybeWrappedStore);
} catch (final ProcessorStateException e) {
if (eosEnabled) {
log.warn("{}Detected unclean shutdown for global store {}. " +
"Wiping global state directory.", logPrefix, stateStore.name(),
e);
try {
Utils.delete(stateDirectory.globalStateDir().getAbsoluteFile());
} catch (final IOException ioe) {
e.addSuppressed(ioe);
}
}
throw e;
}
{code}
This is the {{ProcessorStateException}} path you mentioned. With the default
at-least-once config nothing is deleted, and since {{ProcessorStateException}}
is a
{{StreamsException}}, it ends up in {{closeStateConsumer(stateConsumer,
false)}} in
{{GlobalStreamThread}}, so the directory stays on disk.
The {{eosEnabled}} check was added in KAFKA-17411, so I am not sure if it is
EOS-specific on purpose. Would removing that check be the right fix here?
> 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)
