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

Code before:
        return records;
    }

    public void handleStreamsRebalanceData() {
        if (streamsRebalanceData.isPresent()) {
            boolean hasMissingSourceTopics = false;
            String missingTopicsDetail = null;
            
            for (final StreamsGroupHeartbeatResponseData.Status status : streamsRebalanceData.get().statuses()) {
                if (status.statusCode() == StreamsGroupHeartbeatResponse.Status.SHUTDOWN_APPLICATION.code()) {
                    shutdownErrorHook.run();
                } else if (status.statusCode() == StreamsGroupHeartbeatResponse.Status.MISSING_SOURCE_TOPICS.code()) {
                    hasMissingSourceTopics = true;
                    missingTopicsDetail = status.statusDetail();
                } else if (status.statusCode() == StreamsGroupHeartbeatResponse.Status.INCORRECTLY_PARTITIONED_TOPICS.code()) {
                    final String errorMsg = status.statusDetail();
                    log.error(errorMsg);
                    throw new TopologyException(errorMsg);
                }
            }
            
            if (hasMissingSourceTopics) {
                handleMissingSourceTopicsWithTimeout(missingTopicsDetail);
            } else {
                // Reset timeout tracking when no missing source topics are reported
                firstMissingSourceTopicTime = -1L;
            }

            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) {
        final long currentTime = time.milliseconds();
        
        // Start timeout tracking on first encounter with missing topics
        if (firstMissingSourceTopicTime == -1L) {
            firstMissingSourceTopicTime = currentTime;

Review comment (suggestion):
I think if you use org.apache.kafka.common.utils.Timer and call reset here, you don't need the inline comment.

Code after:
        return records;
    }

    public void handleStreamsRebalanceData() {
        if (streamsRebalanceData.isPresent()) {
            boolean hasMissingSourceTopics = false;
            String missingTopicsDetail = null;
            
            for (final StreamsGroupHeartbeatResponseData.Status status : streamsRebalanceData.get().statuses()) {
                if (status.statusCode() == StreamsGroupHeartbeatResponse.Status.SHUTDOWN_APPLICATION.code()) {
                    shutdownErrorHook.run();
                } else if (status.statusCode() == StreamsGroupHeartbeatResponse.Status.MISSING_SOURCE_TOPICS.code()) {
                    hasMissingSourceTopics = true;
                    missingTopicsDetail = status.statusDetail();
                } else if (status.statusCode() == StreamsGroupHeartbeatResponse.Status.INCORRECTLY_PARTITIONED_TOPICS.code()) {
                    final String errorMsg = status.statusDetail();
                    log.error(errorMsg);
                    throw new TopologyException(errorMsg);
                }
            }
            
            if (hasMissingSourceTopics) {
                handleMissingSourceTopicsWithTimeout(missingTopicsDetail);
            } else {
                // Reset timeout tracking when no missing source topics are reported
                topicsReadyTimer = null;
            }

            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) {
        // Start timeout tracking on first encounter with missing topics
        if (topicsReadyTimer == null) {
            topicsReadyTimer = time.timer(maxPollTimeMs);
            log.info("Missing source topics detected: {}. Will wait up to {}ms before failing.", 
                missingTopicsDetail, maxPollTimeMs);