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

Code before:

            final Map<StreamsRebalanceData.HostInfo, StreamsRebalanceData.EndpointPartitions> partitionsByEndpoint =
                    streamsRebalanceData.get().partitionsByHost();
            final Map<HostInfo, Set<TopicPartition>> activeHostInfoMap = new HashMap<>();
            final Map<HostInfo, Set<TopicPartition>> standbyHostInfoMap = new HashMap<>();

            partitionsByEndpoint.forEach((hostInfo, endpointPartitions) -> {
                activeHostInfoMap.put(new HostInfo(hostInfo.host(), hostInfo.port()), new HashSet<>(endpointPartitions.activePartitions()));
                standbyHostInfoMap.put(new HostInfo(hostInfo.host(), hostInfo.port()), new HashSet<>(endpointPartitions.standbyPartitions()));
            });
            streamsMetadataState.onChange(
                    activeHostInfoMap,
                    standbyHostInfoMap,
                    getTopicPartitionInfo(activeHostInfoMap)
            );
        }
    }

    private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
        // Determine the timeout: use 2 * heartbeatIntervalMs
        final int heartbeatIntervalMs = streamsRebalanceData.get().getHeartbeatIntervalMs();
        final long timeoutMs = 2 * heartbeatIntervalMs;

        // Start timeout tracking on first encounter with missing topics
        if (topicsReadyTimer == null) {
            topicsReadyTimer = time.timer(timeoutMs);
            log.info("Missing source topics detected: {}. Will wait up to {}ms before failing.",
                missingTopicsDetail, timeoutMs);
        } else {
            topicsReadyTimer.update();
        }

        if (topicsReadyTimer.isExpired()) {
            final long elapsedTime = topicsReadyTimer.elapsedMs();
            final String errorMsg = String.format("Missing source topics: %s. Timeout exceeded after %dms.",
                missingTopicsDetail, elapsedTime);
            log.error(errorMsg);

            throw new MissingSourceTopicException(errorMsg);
        } else {
            log.debug("Missing source topics: {}. Elapsed time: {}ms, timeout in: {}ms",
                missingTopicsDetail, topicsReadyTimer.elapsedMs(), topicsReadyTimer.remainingMs());
        }
    }


    static Map<TopicPartition, PartitionInfo> getTopicPartitionInfo(final Map<HostInfo, Set<TopicPartition>> partitionsByHost) {
        final Map<TopicPartition, PartitionInfo> topicToPartitionInfo = new HashMap<>();
        for (final Set<TopicPartition> value : partitionsByHost.values()) {
            for (final TopicPartition topicPartition : value) {
                topicToPartitionInfo.put(

Review comment (bug):
Missing validation for `heartbeatIntervalMs` being -1 (its default value when not yet set by a heartbeat response). If no heartbeat has been received, this will result in `timeoutMs = -2`, causing incorrect timer behavior. Add a check to fall back to `maxPollTimeMs` when `heartbeatIntervalMs == -1`.

Code after:

            final Map<StreamsRebalanceData.HostInfo, StreamsRebalanceData.EndpointPartitions> partitionsByEndpoint =
                    streamsRebalanceData.get().partitionsByHost();
            final Map<HostInfo, Set<TopicPartition>> activeHostInfoMap = new HashMap<>();
            final Map<HostInfo, Set<TopicPartition>> standbyHostInfoMap = new HashMap<>();

            partitionsByEndpoint.forEach((hostInfo, endpointPartitions) -> {
                activeHostInfoMap.put(new HostInfo(hostInfo.host(), hostInfo.port()), new HashSet<>(endpointPartitions.activePartitions()));
                standbyHostInfoMap.put(new HostInfo(hostInfo.host(), hostInfo.port()), new HashSet<>(endpointPartitions.standbyPartitions()));
            });
            streamsMetadataState.onChange(
                    activeHostInfoMap,
                    standbyHostInfoMap,
                    getTopicPartitionInfo(activeHostInfoMap)
            );
        }
    }

    private void handleMissingSourceTopicsWithTimeout(final String missingTopicsDetail) {
        // Use 2 * heartbeatIntervalMs as the timeout ensures at least one heartbeat is sent before raising the exception
        final int heartbeatIntervalMs = streamsRebalanceData.get().heartbeatIntervalMs();
        final long timeoutMs = 2L * heartbeatIntervalMs;

        // Start timeout tracking on first encounter with missing topics
        if (topicsReadyTimer == null) {
            topicsReadyTimer = time.timer(timeoutMs);
            log.info("Missing source topics detected: {}. Will wait up to {}ms before failing.",
                missingTopicsDetail, timeoutMs);
        } else {
            topicsReadyTimer.update();
        }

        if (topicsReadyTimer.isExpired()) {
            final long elapsedTime = topicsReadyTimer.elapsedMs();
            final String errorMsg = String.format("Missing source topics: %s. Timeout exceeded after %dms.",
                missingTopicsDetail, elapsedTime);
            log.error(errorMsg);

            throw new MissingSourceTopicException(errorMsg);
        } else {
            log.debug("Missing source topics: {}. Elapsed time: {}ms, timeout in: {}ms",
                missingTopicsDetail, topicsReadyTimer.elapsedMs(), topicsReadyTimer.remainingMs());
        }
    }


    static Map<TopicPartition, PartitionInfo> getTopicPartitionInfo(final Map<HostInfo, Set<TopicPartition>> partitionsByHost) {
        final Map<TopicPartition, PartitionInfo> topicToPartitionInfo = new HashMap<>();
        for (final Set<TopicPartition> value : partitionsByHost.values()) {
            for (final TopicPartition topicPartition : value) {
                topicToPartitionInfo.put(