[GitHub] [kafka] mjsax commented on a change in pull request #8996: KAFKA-10249: don't try to read un-checkpointed offsets of in-memory stores

[GitBox]

mjsax commented on a change in pull request #8996:
URL: https://github.com/apache/kafka/pull/8996#discussion_r451900686



##########
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:
       Just wondering about 
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L769-L802
 (can't comment below).
   
   For this case, if should hold that `store.offset() == 
changelogOffsetFromCheckpointedOffset(loadedCheckpoints.remove(store.changelogPartition))`
 ?
   
   Or maybe `>=`?
   
   Should we add a sanity check? (Not related to this PR itself actually. -- 
Just wondering.)




----------------------------------------------------------------
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:
us...@infra.apache.org