sunm2n commented on PR #23390: URL: https://github.com/apache/kafka/pull/23390#issuecomment-5577929996
Thanks for the review. All four inline comments are addressed in the new commit. I chose GlobalStreamThread because InvalidOffsetException already uses closeStateConsumer(stateConsumer, true). This delegates to GlobalStateUpdateTask.close(true), which closes registered stores before deleting the global state directory. The existing deletion in GlobalStateManagerImpl.initialize() happens before closing the stores. If we extend that catch to handle TaskCorruptedException, we would also need to preserve the close-before-delete ordering: an earlier global store may already be open when a later store fails to initialize. Reusing the thread-level path preserves that ordering. Regarding the EOS question, I noticed that RocksDBStore.openDB() passes !eosEnabled to cfAccessor.open(). Under ALOS, this bypasses the invalid-state check that produces the TaskCorruptedException in this case. This explains the specific RocksDB path, but does not establish why the broader ProcessorStateException catch in GlobalStateManagerImpl.initialize() is EOS-gated. I'm happy to move the handling into the state manager if that is preferred, while preserving the close-before-delete ordering -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
