KAFKA-21034: Wipe the global state directory when the store is corrupted

[sunm2n]

GlobalStreamThread#initialize already wipes the global state directory when stateConsumer.initialize() throws InvalidOffsetException, but a corrupted state store is not handled the same way.

RocksDBStore#openDB wraps the ProcessorStateException into a TaskCorruptedException, which is a sibling of ProcessorStateException rather than a subclass. It therefore falls through to the generic catch (StreamsException) in GlobalStreamThread#initialize, which calls closeStateConsumer(stateConsumer, false), and the corrupted directory is left on disk. Every restart hits the same state, which becomes a crash loop under an automatic restart policy such as a Kubernetes StatefulSet.

This adds a TaskCorruptedException branch that closes the state consumer with wipeStateStore=true, so the next start rebuilds the global store from the changelog. GlobalStateUpdateTask#close already logs "Deleting global task directory after detecting corruption" on that path, so the deletion itself was already implemented — it was just never reached for this exception.

Does this need a KIP?
I do not think so, but I would like this confirmed. The wipe on the InvalidOffsetException path was introduced in KAFKA-10306 (2020) and the wipeStateStore flag in KAFKA-8897 (2021), both without a KIP. This change makes a corrupted store follow the same existing path rather than introducing new behaviour, and it adds no new config or public API. If that reasoning is wrong, please let me know and I will open a KIP instead.

Verification

• ./gradlew streams:test --tests GlobalStreamThreadTest — 18 tests, 0 failures
• ./gradlew streams:checkstyleMain streams:checkstyleTest streams:spotlessCheck — passed
• The new test fails without the fix: the assertion reports that the global directory still exists.