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

Code before:
                null,
                new StreamsMetricsImpl(metrics, CLIENT_ID, PROCESS_ID.toString(), mockTime),
                new TopologyMetadata(internalTopologyBuilder, config),
                PROCESS_ID,
                CLIENT_ID,
                new LogContext(""),
                null,
                new AtomicLong(Long.MAX_VALUE),
                new LinkedList<>(),
                shutdownErrorHook,
                HANDLER,
                null,
                Optional.of(streamsRebalanceData),
                streamsMetadataState
        ).updateThreadMetadata(adminClientId(CLIENT_ID));

        thread.setState(State.STARTING);
        thread.runOnceWithoutProcessingThreads();

        // Set missing source topics status
        streamsRebalanceData.setStatuses(List.of(
                new StreamsGroupHeartbeatResponseData.Status()
                        .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_SOURCE_TOPICS.code())
                        .setStatusDetail("Missing source topics")
        ));
        
        // First call should not throw exception (within timeout)
        thread.runOnceWithoutProcessingThreads();
        
        // Advance time but not beyond timeout
        mockTime.sleep(150000); // Half of max.poll.interval.ms
        
        // Should still not throw exception
        thread.runOnceWithoutProcessingThreads();
        
        // Clear the missing source topics (simulate recovery)
        streamsRebalanceData.setStatuses(List.of());
        
        // Should complete without exception (recovery successful)
        assertDoesNotThrow(() -> thread.runOnceWithoutProcessingThreads());
        
        // Set missing topics again - should reset the timeout
        streamsRebalanceData.setStatuses(List.of(
                new StreamsGroupHeartbeatResponseData.Status()
                        .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_SOURCE_TOPICS.code())
                        .setStatusDetail("Different missing topics")
        ));
        
        // Should not throw immediately (timeout reset)
        assertDoesNotThrow(() -> thread.runOnceWithoutProcessingThreads());
    }

Review comment (suggestion):
Suggestion: advance time less than 5 min and check if 2nd call throws exception and also check the log message (if easy) and then next step advancing time beyond 5 min as you did!

Code after:
                new LogContext(""),
                null,
                new AtomicLong(Long.MAX_VALUE),
                new LinkedList<>(),
                shutdownErrorHook,
                HANDLER,
                null,
                Optional.of(streamsRebalanceData),
                streamsMetadataState
        ).updateThreadMetadata(adminClientId(CLIENT_ID));

        thread.setState(State.STARTING);
        thread.runOnceWithProcessingThreads();

        streamsRebalanceData.setStatuses(List.of(
                new StreamsGroupHeartbeatResponseData.Status()
                        .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_SOURCE_TOPICS.code())
                        .setStatusDetail("Missing source topics")
        ));
        
        // First call should not throw exception (within timeout)
        thread.runOnceWithProcessingThreads();
        
        // Advance time beyond max.poll.interval.ms (default is 300000ms) to trigger timeout
        mockTime.sleep(300001);
        
        final MissingSourceTopicException exception = assertThrows(MissingSourceTopicException.class, () -> thread.runOnceWithProcessingThreads());
        assertTrue(exception.getMessage().contains("Missing source topics"));
        assertTrue(exception.getMessage().contains("Timeout exceeded"));
    }

    @Test
    public void testStreamsProtocolMissingSourceTopicRecovery() {
        final ConsumerGroupMetadata consumerGroupMetadata = Mockito.mock(ConsumerGroupMetadata.class);
        when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
        when(mainConsumer.poll(Mockito.any(Duration.class))).thenReturn(new ConsumerRecords<>(Map.of(), Map.of()));
        when(mainConsumer.groupMetadata()).thenReturn(consumerGroupMetadata);
        final StreamsRebalanceData streamsRebalanceData = new StreamsRebalanceData(
                UUID.randomUUID(),
                Optional.empty(),
                Map.of(),
                Map.of()
        );

        final Properties props = configProps(false, false, false);
        final Runnable shutdownErrorHook = mock(Runnable.class);
        final StreamsConfig config = new StreamsConfig(props);
        final StreamsMetadataState streamsMetadataState = new StreamsMetadataState(
                new TopologyMetadata(internalTopologyBuilder, config),
                StreamsMetadataState.UNKNOWN_HOST,
                new LogContext(String.format("stream-client [%s] ", CLIENT_ID))