public void start() {
        if (stateUpdaterThread == null) {
            if (!restoredActiveTasks.isEmpty() || !exceptionsAndFailedTasks.isEmpty()) {
                throw new IllegalStateException("State updater started with non-empty output queues. "
                    + BUG_ERROR_MESSAGE);
            }
            stateUpdaterThread = new StateUpdaterThread(name, metrics, changelogReader);
            stateUpdaterThread.start();
            shutdownGate = new CountDownLatch(1);

            // initialize the last commit as of now to prevent first commit happens immediately
            this.lastCommitMs = time.milliseconds();
        }
    }
@Test
    public void shouldNotResumeNonExistingTasks() throws Exception {
        stateUpdater.start();

        verifyPausedTasks();
        verifyRestoredActiveTasks();
        verifyUpdatingTasks();
        verifyExceptionsAndFailedTasks();
    }