[FLINK-40093][runtime] Resume idle splits by alignment check if previously paused

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java

@@ -121,10 +121,9 @@ public void markPaused() {
     public void markIdle() {
         maybeMarkSplitStart();
         if (isPaused()) {
-            // If a split is marked idle, it has no records to emit.
-            // hence it shouldn't be considered paused anymore
             markNotPaused();
-            LOG.warn("[{}] Split marked idle while still paused", splitId);
+            // This is benign: idleness takes over paused state if they race
+            LOG.info("[{}] Split marked idle while still paused", splitId);
         }
         this.idleTimePerSecond.markStart();
     }

flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

@@ -848,12 +848,16 @@ private void checkSplitWatermarkAlignment() {
         Collection<String> splitsToResume = new ArrayList<>();
         sampledSplitWatermarks.forEach(
                 (splitId, splitWatermarks) -> {
-                    if (currentlyIdleSplits.contains(splitId)) {
-                        return;
-                    }
                     if (splitWatermarks.getOldestSample() > currentMaxDesiredWatermark) {
+                        // Skipping pause for idle splits so we won't clear their idleness
+                        if (currentlyIdleSplits.contains(splitId)) {
+                            LOG.info("[{}] Skipping pause for idle split", splitId);
+                            return;
+                        }
                         splitsToPause.add(splitId);
                     } else if (currentlyPausedSplits.contains(splitId)) {
+                        // Resuming possibly-idle splits without clearing their idleness state
+                        // (the next record to arrive will do it naturally)
                         splitsToResume.add(splitId);
                     }
                 });

flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java

@@ -569,6 +569,66 @@ void testAlignmentCheckIsDeferredForIdleSplits() throws Exception {
                 0L, operator.getSplitMetricGroup(split0.splitId()).getAccumulatedPausedTime());
     }
 
+    @Test
+    void testPausedIdleSplitsCanBeResumedByAlignmentCheck() throws Exception {
+        final long idleTimeout = 1000;
+        final long autoWatermarkIntervalMillis = 100;
+        final long marginMillis = autoWatermarkIntervalMillis / 2;
+        final MockSourceReader sourceReader =
+                new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, true, true);
+        final TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+        processingTimeService.setCurrentTime(0);
+        final SourceOperator<Integer, MockSourceSplit> operator =
+                createAndOpenSourceOperatorWithIdleness(
+                        sourceReader, processingTimeService, idleTimeout);
+
+        final MockSourceSplit split0 = new MockSourceSplit(0, 0, 10);
+        final int allowedWatermark4 = 4;
+        final int allowedWatermark7 = 7;
+        split0.addRecord(5);
+        split0.addRecord(6);
+        operator.handleOperatorEvent(
+                new AddSplitEvent<>(
+                        Collections.singletonList(split0), new MockSourceSplitSerializer()));
+        final CollectingDataOutput<Integer> actualOutput = new CollectingDataOutput<>();
+
+        operator.emitNext(actualOutput); // 5
+        sampleAllWatermarks(
+                processingTimeService); // fill the alignment ring buffer with watermark=5
+        final long timeElapsed =
+                WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue() * updateIntervalMillis;
+
+        // Advancing in small steps (TestProcessingTimeService limitation)
+        processingTimeService.advance(2 * autoWatermarkIntervalMillis - timeElapsed); // t=200
+        processingTimeService.advance(idleTimeout); // t=1200
+        processingTimeService.advance(marginMillis); // t=1250
+
+        // Firing alignment check at t=1250.
+        // Idleness clock will be now paused 250ms past the timeout (1000),
+        // which means it will fire on the next idleness periodic check
+        operator.handleOperatorEvent(new WatermarkAlignmentEvent(allowedWatermark4));
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isPaused()).isTrue();
+        assertThat(sourceReader.getPausedSplits()).containsExactly("0");
+        assertOutput(actualOutput, Arrays.asList(5));
+
+        // Triggerring the next idleness periodic check
+        // which marks the split idle while paused.
+        processingTimeService.advance(autoWatermarkIntervalMillis + marginMillis); // t=1400
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isTrue();
+        assertThat(sourceReader.getPausedSplits()).containsExactly("0");
+
+        // Watermark advances and resumes the split
+        // (Though it is still considered idle - this is what this commit is about)
+        operator.handleOperatorEvent(new WatermarkAlignmentEvent(allowedWatermark7));
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isTrue();
+        assertThat(sourceReader.getPausedSplits()).isEmpty();
+
+        // The split emits a record and breaks out of idleness
+        operator.emitNext(actualOutput); // 6
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isActive()).isTrue();
+        assertOutput(actualOutput, Arrays.asList(5, 6));
+    }
+
     private void sampleAllWatermarks(TestProcessingTimeService timeService) throws Exception {
         sampleWatermarks(timeService, WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue());
     }