@Override
    public void initializeIfNeeded() {
        if (state() == State.CREATED) {
            recordCollector.initialize();

            StateManagerUtil.registerStateStores(log, logPrefix, topology, stateMgr, stateDirectory, processorContext);

            // without EOS the checkpoint file would not be deleted after loading, and
            // with EOS we would not checkpoint ever during running state anyways.
            // therefore we can initialize the snapshot as empty so that we would checkpoint right after loading
            offsetSnapshotSinceLastFlush = Collections.emptyMap();

            transitionTo(State.RESTORING);

            log.info("Initialized");
        }
    }
@throws TaskCorruptedException if the state cannot be reused (with EOS) and needs to be reset
@throws LockException    could happen when multi-threads within the single instance, could retry
@throws TimeoutException if initializing record collector timed out
@throws StreamsException fatal error, should close the thread
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java