Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java (Java)

Code before:
            timeoutMs = timeout.get();
            log.debug("Stopping Streams client with timeout = {}ms.", timeoutMs);
        } else {
            timeoutMs = Long.MAX_VALUE;
        }

        if (!setState(State.PENDING_SHUTDOWN)) {
            // Copy the state so that we can atomically check if we are shut down and act on it (log it)
            final State immutableStateCopy = state;
            if (immutableStateCopy.isShuttingDown()) {
                log.info("Skipping shutdown since Streams client is already in {}, waiting for a terminal state", immutableStateCopy);
                if (!waitOnStates(timeoutMs, State.ERROR, State.NOT_RUNNING)) {
                    log.warn("Streams client did transition to a terminal state (ERROR or NOT_RUNNING) within the {}ms timeout", timeoutMs);
                    return false;
                }
                log.info("Streams client stopped completely and transitioned to the terminal {} state", state);
                return true;
            }

            if (state.hasCompletedShutdown()) {
                log.info("Skipping shutdown since Streams client is already in the terminal {} state", state);
                return true;
            }

            throw new IllegalStateException("If transitioning to PENDING_SHUTDOWN fails, the state should be either in "
                + "PENDING_SHUTDOWN, PENDING_ERROR, ERROR, or NOT_RUNNING");
        }

        final Thread shutdownThread = shutdownHelper(false, timeoutMs, leaveGroup);

        shutdownThread.setDaemon(true);
        shutdownThread.start();

        if (waitOnStates(timeoutMs, State.NOT_RUNNING)) {
            log.info("Streams client stopped completely");
            return true;
        } else {
            log.info("Streams client cannot stop completely within the {}ms timeout", timeoutMs);
            return false;
        }
    }

    private void closeToError() {
        if (!setState(State.PENDING_ERROR)) {
            log.info("Skipping shutdown since we are already in {}", state());
        } else {
            final Thread shutdownThread = shutdownHelper(true, -1, false);

            shutdownThread.setDaemon(true);
            shutdownThread.start();
        }

Review comment (suggestion):
As previous discussion, we should only close the consumer in `StreamThread#completeShutdown`

Code after:

            stateDirectory.close();
            // passing in `timeoutMs` here is a quick fix to avoid blocking forever
            // cf https://issues.apache.org/jira/browse/KAFKA-7996
            if (timeoutMs >= 0) {
                adminClient.close(Duration.ofMillis(timeoutMs));
            } else {
                adminClient.close();
            }

            streamsMetrics.removeAllClientLevelSensorsAndMetrics();
            metrics.close();
            if (!error) {
                setState(State.NOT_RUNNING);
            } else {
                setState(State.ERROR);
            }
        }, clientId + "-CloseThread");
    }

    private boolean close(final Optional<Long> timeout, final boolean leaveGroup) {
        final long timeoutMs;
        if (timeout.isPresent()) {
            timeoutMs = timeout.get();
            log.debug("Stopping Streams client with timeout = {}ms.", timeoutMs);
        } else {
            timeoutMs = Long.MAX_VALUE;
        }

        if (!setState(State.PENDING_SHUTDOWN)) {
            // Copy the state so that we can atomically check if we are shut down and act on it (log it)
            final State immutableStateCopy = state;
            if (immutableStateCopy.isShuttingDown()) {
                log.info("Skipping shutdown since Streams client is already in {}, waiting for a terminal state", immutableStateCopy);
                if (!waitOnStates(timeoutMs, State.ERROR, State.NOT_RUNNING)) {
                    log.warn("Streams client did transition to a terminal state (ERROR or NOT_RUNNING) within the {}ms timeout", timeoutMs);
                    return false;
                }
                log.info("Streams client stopped completely and transitioned to the terminal {} state", state);
                return true;
            }

            if (state.hasCompletedShutdown()) {
                log.info("Skipping shutdown since Streams client is already in the terminal {} state", state);
                return true;
            }

            throw new IllegalStateException("If transitioning to PENDING_SHUTDOWN fails, the state should be either in "
                + "PENDING_SHUTDOWN, PENDING_ERROR, ERROR, or NOT_RUNNING");
        }