@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);
        }
    }
@Test
    public void shouldTransitToRunningAfterInitialization() {
        doNothing().when(stateManager).registerStateStores(any(), any());

        task = createStandbyTask();

        assertEquals(CREATED, task.state());

        task.initializeIfNeeded();

        assertEquals(RUNNING, task.state());

        // initialize should be idempotent
        task.initializeIfNeeded();

        assertEquals(RUNNING, task.state());
    }