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

Code before:
        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        task = createStatelessTask(createConfig("100"));
        task.initializeIfNeeded();
        task.completeRestoration(noOpResetter -> { });
        task.suspend();

        assertThat("task is idling", task.timeCurrentIdlingStarted().isPresent());

        task.resume();

        assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
    }

    @Test
    public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        
        task = createStatelessTask(createConfig("100"));
        task.initializeIfNeeded();
        task.completeRestoration(noOpResetter -> { });

        task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));
        
        // Set lastNotReadyLogTime to simulate staleness
        final Field lastNotReadyLogTimeField = StreamTask.class.getDeclaredField("lastNotReadyLogTime");
        lastNotReadyLogTimeField.setAccessible(true);

        try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
             final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
            
            // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
            partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
            
            // Set lastNotReadyLogTime to 100 seconds ago
            final long initialTime = time.milliseconds();
            lastNotReadyLogTimeField.set(task, Optional.of(initialTime - 100_000L));
            
            // Advance time by approximately 5 seconds
            long newTime = time.milliseconds() + 5_000L;
            
            // Should not trigger logging after being stale for 105 seconds
            assertFalse(task.isProcessable(newTime));
            List<String> messages = streamTaskAppender.getMessages();
            assertEquals(0, messages.size(), "No log message should be logged before 20 seconds");

            // Advance time by approximately 20 seconds
            newTime = time.milliseconds() + 20_000L;

            // Should trigger logging after being stale for 120 seconds
            assertFalse(task.isProcessable(newTime));

Review comment (suggestion):
Wondering if we should better add a package-private method to `StreamTask` allowing us to just set it? Using reflections is always ugly.

Code after:
        task = createStatelessTask(createConfig("100"));
        task.initializeIfNeeded();
        task.completeRestoration(noOpResetter -> { });
        task.suspend();

        assertThat("task is idling", task.timeCurrentIdlingStarted().isPresent());

        task.resume();

        assertThat("task is not idling", task.timeCurrentIdlingStarted().isEmpty());
    }

    @Test
    public void shouldLogNotReadyWhenStaleAfterThreshold() throws Exception {
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        
        task = createStatelessTask(createConfig("100"));
        task.initializeIfNeeded();
        task.completeRestoration(noOpResetter -> { });

        task.addRecords(partition1, singleton(getConsumerRecordWithOffsetAsTimestamp(partition1, 0)));

        try (final LogCaptureAppender streamTaskAppender = LogCaptureAppender.createAndRegister(StreamTask.class);
             final LogCaptureAppender partitionGroupAppender = LogCaptureAppender.createAndRegister(PartitionGroup.class)) {
            
            // Enable TRACE logging for PartitionGroup to capture the "ready for processing" message
            partitionGroupAppender.setClassLogger(PartitionGroup.class, Level.TRACE);
            
            // Set lastNotReadyLogTime to 100 seconds ago
            final long initialTime = time.milliseconds();
            task.setLastNotReadyLogTime(initialTime - 100_000L);
            
            // Advance time by 19.999 seconds
            long newTime = time.milliseconds() + 19_999L;
            
            // Should not trigger logging after being stale for 119 seconds
            assertFalse(task.isProcessable(newTime));
            List<String> messages = streamTaskAppender.getMessages();
            assertEquals(0, messages.size(), "No log message should be logged before 120 seconds");

            // Advance time by 20 seconds
            newTime = time.milliseconds() + 20_000L;

            // Should trigger logging after being stale for 120 seconds
            assertFalse(task.isProcessable(newTime));
            
            // Validate INFO log from StreamTask about partition2 not being ready
            messages = streamTaskAppender.getMessages();
            final String expectedNotReadyMessage = "stream-thread [Test worker] task [0_0] Partition topic2-0 has fetched lag of -1\n\tWaiting to fetch data for topic2-0";
            final String expectedReadyMessage = "Partition topic1-0 has buffered data, ready for processing";