[
https://issues.apache.org/jira/browse/KAFKA-21034?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18112058#comment-18112058
]
Uladzislau Blok edited comment on KAFKA-21034 at 9/6/26 12:53 PM:
------------------------------------------------------------------
[~sunm2n]
Hey, sure thing. Take it if you want to
Reg. current functionality and {{{}GlobalStateManagerImpl#initialize{}}}. There
is a gap here.
The piece of code you have shared is triggered only for
{{{}ProcessorStateException{}}}, while {{RocksDBStore#openDB}} wraps
{{ProcessorStateException }}into {{{}TaskCorruptedException{}}}:
```
{code:java}
try {
openRocksDB(dbOptions, columnFamilyOptions);
dbAccessor = new DirectDBAccessor(db, fOptions, wOptions);
try {
final Position existingPositionOrEmpty =
cfAccessor.open(dbAccessor, !eosEnabled);
if (position == null) {
position = existingPositionOrEmpty;
} else {
// For segmented stores, the overall position is composed
of multiple underlying stores, so merge this store's position into it.
position.merge(existingPositionOrEmpty);
}
} catch (final ProcessorStateException e) {
final String message = "State store " + name + " didn't find a
valid state, since under EOS it has the risk of getting uncommitted data in
stores";
throw new TaskCorruptedException(Set.of(taskId), new
ProcessorStateException(message, e));
} catch (final StreamsException fatal) {
final String fatalMessage = "Fatal error while opening store "
+ name;
throw new ProcessorStateException(fatalMessage, fatal);
} catch (final RocksDBException fatal) {
final String fatalMessage = "Error opening store " + name;
throw new ProcessorStateException(fatalMessage, fatal);
}
} catch (final RuntimeException e) {
closeNativeResources();
throw e;
} {code}
So {{ProcessorStateException}} is the root cause, not wrapping exception.
Because of that we missed this try-catch
was (Author: JIRAUSER309258):
[~sunm2n]
Hey, sure thing. Take it if you want to
Reg. current functionality and `GlobalStateManagerImpl#initialize`. There is a
gap here.
The piece of code you have shared is triggered only for
`ProcessorStateException`, while `RocksDBStore#openDB` wraps
`ProcessorStateException` into `TaskCorruptedException`:
```
try {
openRocksDB(dbOptions, columnFamilyOptions);
dbAccessor = new DirectDBAccessor(db, fOptions, wOptions);
try {
final Position existingPositionOrEmpty =
cfAccessor.open(dbAccessor, !eosEnabled);
if (position == null) {
position = existingPositionOrEmpty;
} else {
// For segmented stores, the overall position is composed
of multiple underlying stores, so merge this store's position into it.
position.merge(existingPositionOrEmpty);
}
} catch (final ProcessorStateException e) {
final String message = "State store " + name + " didn't find a
valid state, since under EOS it has the risk of getting uncommitted data in
stores";
throw new TaskCorruptedException(Set.of(taskId), new
ProcessorStateException(message, e));
} catch (final StreamsException fatal) {
final String fatalMessage = "Fatal error while opening store "
+ name;
throw new ProcessorStateException(fatalMessage, fatal);
} catch (final RocksDBException fatal) {
final String fatalMessage = "Error opening store " + name;
throw new ProcessorStateException(fatalMessage, fatal);
}
} catch (final RuntimeException e) {
closeNativeResources();
throw e;
}
```
So `ProcessorStateException` is the root cause, not wrapping exception. Because
of that we missed this try-catch
> 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)
