Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java (Java)

Code before:
            consumer,
            changelogReader,
            null,
            taskManager,
            null,
            streamsMetrics,
            topologyMetadata,
            PROCESS_ID,
            CLIENT_ID,
            new LogContext(""),
            new AtomicInteger(),
            new AtomicLong(Long.MAX_VALUE),
            new LinkedList<>(),
            null,
            HANDLER,
            null,
            Optional.empty(),
            null
        );
    }

    private void maybeRunStateUpdater(final boolean stateUpdaterEnabled) {
        if (stateUpdaterEnabled) {
            thread.taskManager().init();
        }
    }
    
    private void runOnce(final boolean processingThreadsEnabled) {
        if (processingThreadsEnabled) {
            thread.runOnceWithProcessingThreads();
        } else {
            thread.runOnceWithoutProcessingThreads();
        }
    }

    private void runUntilTimeoutOrException(final Runnable action) {
        final long expectedEnd = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
        while (System.currentTimeMillis() < expectedEnd) {
            action.run();
            mockTime.sleep(10);
        }
    }

    private boolean runUntilTimeoutOrCondition(final Runnable action, final TestCondition testCondition) throws Exception {
        final long expectedEnd = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
        while (System.currentTimeMillis() < expectedEnd) {
            action.run();
            if (testCondition.conditionMet()) {
                return true;
            }
            mockTime.sleep(10);

Review comment (suggestion):
It should be enough to call `thread.taskManager().init()` instead of this method. Whether the state updater is enabled is already checked in the `init()`.

You also do not need to call it in each and every test method. Only in some the call is needed.

Code after:
            topologyMetadata,
            PROCESS_ID,
            CLIENT_ID,
            new LogContext(""),
            new AtomicInteger(),
            new AtomicLong(Long.MAX_VALUE),
            new LinkedList<>(),
            null,
            HANDLER,
            null,
            Optional.empty(),
            null
        );
    }
    
    private void runOnce(final boolean processingThreadsEnabled) {
        if (processingThreadsEnabled) {
            thread.runOnceWithProcessingThreads();
        } else {
            thread.runOnceWithoutProcessingThreads();
        }
    }

    private void runUntilTimeoutOrException(final Runnable action) {
        final long expectedEnd = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
        while (System.currentTimeMillis() < expectedEnd) {
            action.run();
            mockTime.sleep(10);
        }
    }

    private boolean runUntilTimeoutOrCondition(final Runnable action, final TestCondition testCondition) throws Exception {
        final long expectedEnd = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
        while (System.currentTimeMillis() < expectedEnd) {
            action.run();
            if (testCondition.conditionMet()) {
                return true;
            }
            mockTime.sleep(10);
        }
        return false;
    }

}