@Override
    public void initializeIfNeeded() {
        if (state() == State.CREATED) {
            StateManagerUtil.registerStateStores(log, logPrefix, topology, stateMgr, stateDirectory, processorContext);

            // with and without EOS we would check for checkpointing at each commit during running,
            // and the file may be deleted in which case we should checkpoint immediately,
            // therefore we initialize the snapshot as empty
            offsetSnapshotSinceLastFlush = Collections.emptyMap();

            // no topology needs initialized, we can transit to RUNNING
            // right after registered the stores
            transitionTo(State.RESTORING);
            transitionTo(State.RUNNING);

            processorContext.initialize();

            log.info("Initialized");
        } else if (state() == State.RESTORING) {
            throw new IllegalStateException("Illegal state " + state() + " while initializing standby task " + id);
        }
    }
@throws TaskCorruptedException if the state cannot be reused (with EOS) and needs to be reset)
@throws StreamsException fatal error, should close the thread
streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java