void initializeStoreOffsetsFromCheckpoint(final boolean storeDirIsEmpty) {
        try {
            final Map<TopicPartition, Long> loadedCheckpoints = checkpointFile.read();

            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()) {
                    log.info("Initializing to the starting offset for changelog {} of in-memory state store {}",
                             store.changelogPartition, store.stateStore.name());
                } else if (store.offset() == null) {
                    if (loadedCheckpoints.containsKey(store.changelogPartition)) {
                        final Long offset = changelogOffsetFromCheckpointedOffset(loadedCheckpoints.remove(store.changelogPartition));
                        store.setOffset(offset);

                        log.info("State store {} initialized from checkpoint with offset {} at changelog {}",
                                  store.stateStore.name(), store.offset, store.changelogPartition);
                    } else {
                        // with EOS, if the previous run did not shutdown gracefully, we may lost the checkpoint file
                        // and hence we are uncertain that the current local state only contains committed data;
                        // in that case we need to treat it as a task-corrupted exception
                        if (eosEnabled && !storeDirIsEmpty) {
                            log.warn("State store {} did not find checkpoint offsets while stores are not empty, " +
                                "since under EOS it has the risk of getting uncommitted data in stores we have to " +
                                "treat it as a task corruption error and wipe out the local state of task {} " +
                                "before re-bootstrapping", store.stateStore.name(), taskId);

                            throw new TaskCorruptedException(Collections.singleton(taskId));
                        } else {
                            log.info("State store {} did not find checkpoint offset, hence would " +
                                "default to the starting offset at changelog {}",
                                store.stateStore.name(), store.changelogPartition);
                        }
                    }
                }  else {
                    loadedCheckpoints.remove(store.changelogPartition);
                    log.debug("Skipping re-initialization of offset from checkpoint for recycled store {}",
                              store.stateStore.name());
                }
            }

            if (!loadedCheckpoints.isEmpty()) {
                log.warn("Some loaded checkpoint offsets cannot find their corresponding state stores: {}", loadedCheckpoints);
            }

            if (eosEnabled) {
                checkpointFile.delete();
            }
        } catch (final TaskCorruptedException e) {
            throw e;
        } catch (final IOException | RuntimeException e) {
            // both IOException or runtime exception like number parsing can throw
            throw new ProcessorStateException(format("%sError loading and deleting checkpoint file when creating the state manager",
                logPrefix), e);
        }
    }
@throws ProcessorStateException if the task directory does not exist and could not be created
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java