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

Code before:
            } else {
                setDeadline = true;
            }
        }
        result.put(getName() + "-restore-consumer", restoreConsumerInstanceIdFuture);

        if (producerInstanceIdFuture.isDone()) {
            if (producerInstanceIdFuture.isCompletedExceptionally()) {
                producerInstanceIdFuture = new KafkaFutureImpl<>();
                setDeadline = true;
            }
        } else {
            setDeadline = true;
        }
        result.put(getName() + "-producer", producerInstanceIdFuture);

        if (setDeadline) {
            fetchDeadlineClientInstanceId = time.milliseconds() + timeout.toMillis();
        }

        return result;
    }

    public void closeConsumer(final long timeoutMs, final boolean leaveGroup) {
        final CloseOptions.GroupMembershipOperation operation = leaveGroup ? LEAVE_GROUP : REMAIN_IN_GROUP;
        final CloseOptions closeOptions = CloseOptions.timeout(Duration.ofMillis(timeoutMs <= 0 ? 0 : timeoutMs))
            .withGroupMembershipOperation(operation);
        mainConsumer.close(closeOptions);
    }

    // the following are for testing only
    void setNow(final long now) {
        this.now = now;
    }

    TaskManager taskManager() {
        return taskManager;
    }

    int currentNumIterations() {
        return numIterations;
    }

    ConsumerRebalanceListener rebalanceListener() {
        return rebalanceListener;
    }

    Consumer<byte[], byte[]> mainConsumer() {
        return mainConsumer;
    }


Review comment (suggestion):
 it's definitely a bit questionable that we don't pass the timeout in to consumer#close in the current code (interestingly we also don't for producer#close but do use the timeout for admin#close) But, I don't want to change that in this PR, on the off chance there was an actual reason for it 

I know we had some issues with the producer timeout on close in the past but I don't remember the specifics and I'm not personally aware of any similar issues for the consumer's close. It's worth following up on, but separately from this. Let's just use the default for now (ie don't set the timeout at all)

Code after:
        }

        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("");
    }

    /**
     * Produces a string representation containing useful information about a StreamThread, starting with the given indent.
     * This is useful in debugging scenarios.
     *
     * @return A string representation of the StreamThread instance.
     */