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

Code before:
    /**
     * Compute the latency based on the current marked timestamp, and update the marked timestamp
     * with the current system timestamp.
     *
     * @return latency
     */
    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) {

Review comment (suggestion):
`requestLeaveGroupDuringShutdown` is not used by other classes now, so we can remove `requestLeaveGroupDuringShutdown` by calling `leaveGroupRequested.set(true);`

Code after:
        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)
     */
    // visible for testing
    int maybeCommit() {
        final int committed;
        if (now - lastCommitMs > commitTimeMs) {
            if (log.isDebugEnabled()) {
                log.debug("Committing all active tasks {} and standby tasks {} since {}ms has elapsed (commit interval is {}ms)",
                          taskManager.activeRunningTaskIds(), taskManager.standbyTaskIds(), now - lastCommitMs, commitTimeMs);
            }

            committed = taskManager.commit(
                taskManager.allOwnedTasks()
                    .values()
                    .stream()
                    .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
                    .collect(Collectors.toSet())
            );

            if (committed == -1) {
                log.debug("Unable to commit as we are in the middle of a rebalance, will try again when it completes.");
            } else {
                now = time.milliseconds();
                lastCommitMs = now;
            }
        } else {
            committed = taskManager.maybeCommitActiveTasksPerUserRequested();
        }

        if ((now - lastPurgeMs) > purgeTimeMs) {
            // try to purge the committed records for repartition topics if possible