[jira] [Created] (KAFKA-20808) StreamThread dies with fatal InvalidStateStoreException after TaskCorruptedException

["Matthias J. Sax (Jira)" <ji...@apache.org>]

Matthias J. Sax created KAFKA-20808:
---------------------------------------

             Summary: StreamThread dies with fatal InvalidStateStoreException after TaskCorruptedException
                 Key: KAFKA-20808
                 URL: https://issues.apache.org/jira/browse/KAFKA-20808
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 4.3.0
            Reporter: Matthias J. Sax


First KafkaStreams hits an error opening a store (this code was added via KIP-1035, and triggers only via EOS)
{code:java}
Caused by: org.apache.kafka.streams.errors.ProcessorStateException: State store example-window-store-v1.1700000000000 didn't find a valid state, since under EOS it has the risk of getting uncommitted data in stores	... 37 common frames omitted
    Caused by: org.apache.kafka.streams.errors.ProcessorStateException: Invalid state during store open. Expected state to be either empty or closed
	at org.apache.kafka.streams.state.internals.AbstractColumnFamilyAccessor.open(AbstractColumnFamilyAccessor.java:93)
	at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:263){code}
The error is converted into a TaskCorruptedException and KafkaStreams try to close the task dirty, and re-initialize it, but fails with
{code:java}
org.apache.kafka.streams.errors.InvalidStateStoreException: Store example-window-store-v1.1700000000000 is currently closed
	at org.apache.kafka.streams.state.internals.RocksDBStore.validateStoreOpen(RocksDBStore.java:472)
	at org.apache.kafka.streams.state.internals.RocksDBStore.committedOffset(RocksDBStore.java:745)
	at org.apache.kafka.streams.state.internals.AbstractSegments.committedOffset(AbstractSegments.java:196)
	at org.apache.kafka.streams.state.internals.AbstractRocksDBSegmentedBytesStore.committedOffset(AbstractRocksDBSegmentedBytesStore.java:329)
	at org.apache.kafka.streams.state.internals.WrappedStateStore.committedOffset(WrappedStateStore.java:153)
	at org.apache.kafka.streams.state.internals.WrappedStateStore.committedOffset(WrappedStateStore.java:153)
	at org.apache.kafka.streams.state.internals.WrappedStateStore.committedOffset(WrappedStateStore.java:153)
	at org.apache.kafka.streams.processor.internals.ProcessorStateManager.initializeStoreOffsets(ProcessorStateManager.java:308)
	at org.apache.kafka.streams.processor.internals.StateManagerUtil.registerStateStores(StateManagerUtil.java:150)
	at org.apache.kafka.streams.processor.internals.StandbyTask.initializeIfNeeded(StandbyTask.java:112)
	at org.apache.kafka.streams.processor.internals.TaskManager.addTaskToStateUpdater(TaskManager.java:949)
	at org.apache.kafka.streams.processor.internals.TaskManager.addTasksToStateUpdater(TaskManager.java:934)
	at org.apache.kafka.streams.processor.internals.TaskManager.checkStateUpdater(TaskManager.java:848)
	at org.apache.kafka.streams.processor.internals.StreamThread.checkStateUpdater(StreamThread.java:1409)
	at org.apache.kafka.streams.processor.internals.StreamThread.runOnceWithoutProcessingThreads(StreamThread.java:1235)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:956)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:916)"} {code}
 

Root cause:

When a segmented store fails to open a segment during openExisting under EOS, the task is closed dirty and revived as intended, but the retry dies fatally instead of recovering. Root cause is a stale, unopened segment left in the in-memory AbstractSegments.segments map that the dirty-close never clears.

Mechanism
1. AbstractSegments.getOrCreateSegment inserts the segment into the map (:95) before openSegmentDB (:99) — long-standing (KAFKA-3522).
2. Under EOS the open now throws TaskCorruptedException via the OPEN/closed status-key check (AbstractColumnFamilyAccessor.open:107 → RocksDBStore.openDB:279). This fires inside AbstractRocksDBSegmentedBytesStore.init at segments.openExisting (:313) — before stateStoreContext.register(...) (:319).
3. Because the store was never registered, ProcessorStateManager.close() (closes only registered stores, :666) skips it, so AbstractSegments.close()/segments.clear() (:205/:209) never runs. The unopened segment (open==false) survives closeDirty.
4. On revive+retry, openExisting short-circuits on segments.containsKey (:90-91) and returns the stale segment without reopening; init completes and registers the store.
5. ProcessorStateManager.initializeStoreOffsets then calls committedOffset (:345), which iterates segments (AbstractSegments:196) and calls validateStoreOpen on the stale one → fatal InvalidStateStoreException (RocksDBStore:894→:478) → caught as generic StreamsException (StreamThread:1039, not the TaskCorruptedException branch at :1004) → SHUTDOWN_CLIENT.

While the code in AbstractSegments.getOrCreateSegment is old, it could not trip before, because no TaskCorruptedException was throw on this code path. This only change with KIP-1035.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)