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

Code before:
     */
    private long advanceNowAndComputeLatency() {
        final long previous = now;
        now = time.milliseconds();

        return Math.max(now - previous, 0);
    }

    /**
     * Shutdown this stream thread.
     * <p>
     * Note that there is nothing to prevent this function from being called multiple times
     * (e.g., in testing), hence the state is set only the first time
     *
     * @param leaveGroup this flag will control whether the consumer will leave the group on close or not
     */
    public void shutdown(final boolean leaveGroup) {
        log.info("Informed to shut down");
        final State oldState = setState(State.PENDING_SHUTDOWN);
        if (leaveGroup) {
            requestLeaveGroupDuringShutdown();
        }
        if (oldState == State.CREATED) {
            // The thread may not have been started. Take responsibility for shutting down
            completeShutdown(true);
        }
    }

    private void completeShutdown(final boolean cleanRun) {
        // set the state to pending shutdown first as it may be called due to error;
        // its state may already be PENDING_SHUTDOWN so it will return false but we
        // intentionally do not check the returned flag
        setState(State.PENDING_SHUTDOWN);

        log.info("Shutting down {}", cleanRun ? "clean" : "unclean");

        mainConsumerInstanceIdFuture.complete(null);

        try {
            taskManager.shutdown(cleanRun);
        } catch (final Throwable e) {
            log.error("Failed to close task manager due to the following error:", e);
        }
        try {
            topologyMetadata.unregisterThread(threadMetadata.threadName());
        } catch (final Throwable e) {
            log.error("Failed to unregister thread due to the following error:", e);
        }
        try {
            changelogReader.clear();
        } catch (final Throwable e) {

Review comment (suggestion):
We could leverage the `leaveGroupRequested` and don't need a new argument.

Code after:
            if (cause == null) {
                throw new StreamsException(format);
            } else {
                throw new StreamsException(format, cause);
            }
        }
    }

    private void addToResetList(
        final TopicPartition partition,
        final Set<TopicPartition> partitions,
        final String logMessage,
        final Set<String> loggedTopics
    ) {
        final String topic = partition.topic();
        if (loggedTopics.add(topic)) {
            log.info(logMessage, topic);
        }
        partitions.add(partition);
    }

    private void addToResetList(
        final TopicPartition partition,
        final Map<TopicPartition, Duration> durationForPartitions,
        final Duration durationTime,
        final String logMessage,
        final String durationString,
        final Set<String> loggedTopics
    ) {
        final String topic = partition.topic();
        if (loggedTopics.add(topic)) {
            log.info(logMessage, topic, durationString);
        }
        durationForPartitions.put(partition, durationTime);
    }

    // This method is added for usage in tests where mocking the underlying native call is not possible.
    public boolean isThreadAlive() {
        return isAlive();
    }

    // Call method when a topology is resumed
    public void signalResume() {
        taskManager.signalResume();
    }

    /**
     * Try to commit all active tasks owned by this thread.
     *
     * @throws TaskMigratedException if committing offsets failed (non-EOS)
     *                               or if the task producer got fenced (EOS)