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

Code before:
                task.getValue().inputPartitions(),
                task.getValue().committedOffsets(),
                task.getValue().highWaterMark(),
                task.getValue().timeCurrentIdlingStarted()
            ));
        }

        final String adminClientId = threadMetadata.adminClientId();
        threadMetadata = new ThreadMetadataImpl(
            getName(),
            state().name(),
            consumerClientId(getName()),
            restoreConsumerClientId(getName()),
            taskManager.producerClientIds(),
            adminClientId,
            activeTasksMetadata,
            standbyTasksMetadata
        );
    }

    /**
     * Getting the list of current active tasks of the thread;
     * Note that the returned list may be used by other thread than the StreamThread itself,
     * and hence need to be read-only
     */
    public Set<Task> readOnlyActiveTasks() {
        return readyOnlyAllTasks().stream()
            .filter(Task::isActive).collect(Collectors.toSet());
    }

    /**
     * Getting the list of all owned tasks of the thread, including both active and standby;
     * Note that the returned list may be used by other thread than the StreamThread itself,
     * and hence need to be read-only
     */
    public Set<Task> readyOnlyAllTasks() {
        return taskManager.readOnlyAllTasks();
    }

    /**
     * Produces a string representation containing useful information about a StreamThread.
     * This is useful in debugging scenarios.
     *
     * @return A string representation of the StreamThread instance.
     */
    @Override
    public String toString() {
        return toString("");
    }

    /**

Review comment (suggestion):
`Consumer# unsubscribe()` will make consumers leave the group. 
I believe this is a workaround, and now we could rely onthe  new close API

Code after:
            // 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) {
            log.error("Failed to close changelog reader due to the following error:", e);
        }
        try {
            final GroupMembershipOperation membershipOperation = leaveGroupRequested.get() ? LEAVE_GROUP : REMAIN_IN_GROUP;
            mainConsumer.close(CloseOptions.groupMembershipOperation(membershipOperation));
        } catch (final Throwable e) {
            log.error("Failed to close consumer due to the following error:", e);
        }
        try {
            // restore consumer isn't part of a consumer group so we use REMAIN_IN_GROUP to skip any leaveGroup checks
            restoreConsumer.close();
        } catch (final Throwable e) {
            log.error("Failed to close restore consumer due to the following error:", e);
        }
        streamsMetrics.removeAllThreadLevelSensors(getName());
        streamsMetrics.removeAllThreadLevelMetrics(getName());

        setState(State.DEAD);

        log.info("Shutdown complete");
    }

    /**