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

[GitBox <gi...@apache.org>]

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



##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
##########
@@ -104,26 +104,27 @@ static void closeStateManager(final Logger log,
             if (stateDirectory.lock(id)) {
                 try {
                     stateMgr.close();
-
-                    if (wipeStateStore) {
-                        log.debug("Wiping state stores for {} task {}", taskType, id);
-                        // we can just delete the whole dir of the task, including the state store images and the checkpoint files,
-                        // and then we write an empty checkpoint file indicating that the previous close is graceful and we just
-                        // need to re-bootstrap the restoration from the beginning
-                        Utils.delete(stateMgr.baseDir());
-                    }
                 } catch (final ProcessorStateException e) {
                     firstException.compareAndSet(null, e);
                 } finally {
-                    stateDirectory.unlock(id);
+                    try {
+                        if (wipeStateStore) {
+                            log.debug("Wiping state stores for {} task {}", taskType, id);
+                            // we can just delete the whole dir of the task, including the state store images and the checkpoint files,
+                            // and then we write an empty checkpoint file indicating that the previous close is graceful and we just
+                            // need to re-bootstrap the restoration from the beginning
+                            Utils.delete(stateMgr.baseDir());
+                        }
+                    } finally {
+                        stateDirectory.unlock(id);
+                    }

Review comment:
       I take it this block can also throw an exception? We shouldn't throw exceptions inside a finally block because it's not defined when the exception will be thrown, or in the case where the first try block threw, which exception is ultimately thrown is also undefined.
   
   To make this simpler to grapple with, we added org.apache.kafka.streams.state.internals.ExceptionUtils#executeAll

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
##########
@@ -222,8 +222,16 @@ void initializeStoreOffsetsFromCheckpoint(final boolean storeDirIsEmpty) {
             log.trace("Loaded offsets from the checkpoint file: {}", loadedCheckpoints);
 
             for (final StateStoreMetadata store : stores.values()) {
+                if (store.corrupted) {
+                    log.error("Tried to initialize store offsets for corrupted store {}", store);
+                    throw new IllegalStateException("Should not initialize offsets for a corrupted task");
+                }
+
                 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()) {

Review comment:
       I think we should also remove the changelogPartition from loadedCheckpoints, if it exists. Otherwise, we'll spuriously warn in L267.




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