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

Code before:
    @Test
    public void shouldStartAndCleanlyShutdownStateUpdaterThread() {

        final String stateUpdaterName = "test-state-updater";
        final Metrics metrics = new Metrics(time);
        final StreamsConfig config = new StreamsConfig(configProps());
        final DefaultStateUpdater defaultStateUpdater =
            new DefaultStateUpdater(stateUpdaterName, metrics, config, null, changeLogReader, topologyMetadata, time);

        final TaskManager taskManager = new TaskManager(
            time,
            changeLogReader,
            ProcessId.randomProcessId(),
            "logPrefix",
            activeTaskCreator,
            standbyTaskCreator,
            null,
            topologyMetadata,
            adminClient,
            stateDirectory,
            defaultStateUpdater,
            null
        );

        taskManager.init();
        assertTrue(defaultStateUpdater.isRunning());

        defaultStateUpdater.shutdown(Duration.ofMinutes(1));
        assertFalse(defaultStateUpdater.isRunning());
    }

    private Properties configProps() {
        return mkObjectProperties(mkMap(
            mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, "appId"),
            mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2171")
        ));
    }

    private static KafkaFutureImpl<DeletedRecords> completedFuture() {
        final KafkaFutureImpl<DeletedRecords> futureDeletedRecords = new KafkaFutureImpl<>();
        futureDeletedRecords.complete(null);
        return futureDeletedRecords;
    }

    private void makeTaskFolders(final String... names) throws Exception {
        final ArrayList<TaskDirectory> taskFolders = new ArrayList<>(names.length);
        for (int i = 0; i < names.length; i++) {
            final String name = names[i];
            final Path path = testFolder.resolve(name).toAbsolutePath();
            if (Files.notExists(path)) {
                Files.createDirectories(path);

Review comment (suggestion):
You do not need to check for `isRunning()` here. `isRunning()` is an implementation detail of the `DefaultStateUpdater`. It suffices to verify that `stateUpdater.start()` is called on the mock. Something simple like the following:

```java
    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void shouldStartStateUpdaterOnInit(final boolean stateUpdaterEnabled) {
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, stateUpdaterEnabled);
        taskManager.init();
        if (stateUpdaterEnabled) {
            verify(stateUpdater).start();
        } else {
            verify(stateUpdater, never()).start();
        }
    }
```
The shutdown is tested elsewhere. You do not need to test it again here.

Code after:
    public void shouldStartStateUpdaterOnInit(final boolean stateUpdaterEnabled) {
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, stateUpdaterEnabled);
        taskManager.init();
        if (stateUpdaterEnabled) {
            verify(stateUpdater).start();
        } else {
            verify(stateUpdater, never()).start();
        }
    }

    private static KafkaFutureImpl<DeletedRecords> completedFuture() {
        final KafkaFutureImpl<DeletedRecords> futureDeletedRecords = new KafkaFutureImpl<>();
        futureDeletedRecords.complete(null);
        return futureDeletedRecords;
    }

    private void makeTaskFolders(final String... names) throws Exception {
        final ArrayList<TaskDirectory> taskFolders = new ArrayList<>(names.length);
        for (int i = 0; i < names.length; i++) {
            final String name = names[i];
            final Path path = testFolder.resolve(name).toAbsolutePath();
            if (Files.notExists(path)) {
                Files.createDirectories(path);
            }
            taskFolders.add(new TaskDirectory(path.toFile(), null));
        }
        when(stateDirectory.listNonEmptyTaskDirectories()).thenReturn(taskFolders);
    }

    private void writeCheckpointFile(final TaskId task, final Map<TopicPartition, Long> offsets) throws Exception {
        final File checkpointFile = getCheckpointFile(task);
        final Path checkpointFilePath = checkpointFile.toPath();
        Files.createFile(checkpointFilePath);
        new OffsetCheckpoint(checkpointFile).write(offsets);
        lenient().when(stateDirectory.checkpointFileFor(task)).thenReturn(checkpointFile);
        expectDirectoryNotEmpty(task);
    }

    private File getCheckpointFile(final TaskId task) {
        return new File(new File(testFolder.toAbsolutePath().toString(), task.toString()), StateManagerUtil.CHECKPOINT_FILE_NAME);
    }

    private static ConsumerRecord<byte[], byte[]> getConsumerRecord(final TopicPartition topicPartition, final long offset) {
        return new ConsumerRecord<>(topicPartition.topic(), topicPartition.partition(), offset, null, null);
    }

    private static class StateMachineTask extends AbstractTask implements Task {
        private final boolean active;

        // TODO: KAFKA-12569 clean up usage of these flags and use the new commitCompleted flag where appropriate
        private boolean commitNeeded = false;