ableegoldman commented on a change in pull request #8996:
URL: https://github.com/apache/kafka/pull/8996#discussion_r451945495
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
##########
@@ -224,6 +224,9 @@ void initializeStoreOffsetsFromCheckpoint(final boolean
storeDirIsEmpty) {
for (final StateStoreMetadata store : stores.values()) {
if (store.changelogPartition == null) {
log.info("State store {} is not logged and hence would not
be restored", store.stateStore.name());
+ } else if (!store.stateStore.persistent()) {
+ log.info("Initializing to the starting offset for
changelog {} of in-memory state store {}",
+ store.changelogPartition,
store.stateStore.name());
Review comment:
No, because we close the state manager completely and clear all this
data before reviving a corrupted task.
I've had to re-convince myself of this several times already. Maybe we can
also add a check that none of them are marked corrupted while initializing
offsets
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
