[BP-2.2][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

@@ -780,12 +780,16 @@ private void checkSplitWatermarkAlignment() {
         Collection<String> splitsToResume = new ArrayList<>();
         splitCurrentWatermarks.forEach(
                 (splitId, splitWatermark) -> {
-                    if (currentlyIdleSplits.contains(splitId)) {
-                        return;
-                    }
                     if (splitWatermark > 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

@@ -547,6 +547,56 @@ void testAlignmentCheckIsDeferredForIdleSplits() throws Exception {
                 0L, operator.getSplitMetricGroup(split0.splitId()).getAccumulatedPausedTime());
     }
 
+    @Test
+    void testPausedIdleSplitsCanBeResumedByAlignmentCheck() throws Exception {
+        final long idleTimeout = 100;
+        final MockSourceReader sourceReader =
+                new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, true, true);
+        final TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+        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(4);
+        split0.addRecord(5);
+        split0.addRecord(6);
+        split0.addRecord(7);
+        split0.addRecord(8);
+        operator.handleOperatorEvent(
+                new AddSplitEvent<>(Arrays.asList(split0), new MockSourceSplitSerializer()));
+        final CollectingDataOutput<Integer> actualOutput = new CollectingDataOutput<>();
+
+        for (int i = 0; i < 3; i++) {
+            operator.emitNext(actualOutput);
+            processingTimeService.advance(idleTimeout - 1);
+        }
+        assertOutput(actualOutput, Arrays.asList(4, 5, 6));
+
+        // Alignment check fires and pauses the split
+        operator.handleOperatorEvent(new WatermarkAlignmentEvent(allowedWatermark4));
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isPaused()).isTrue();
+        assertThat(sourceReader.getPausedSplits()).containsExactly("0");
+        assertOutput(actualOutput, Arrays.asList(4, 5, 6));
+
+        // Normally idlenessTimer can't elapse while the split is paused
+        // So calling it manually to simulate a race condition
+        operator.updateCurrentSplitIdle(split0.splitId(), true);
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isIdle()).isTrue();
+
+        // Watermark advances and resumes the split (Though it is still considered idle)
+        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); // 7
+        assertThat(operator.getSplitMetricGroup(split0.splitId()).isActive()).isTrue();
+        assertOutput(actualOutput, Arrays.asList(4, 5, 6, 7));
+    }
+
     private void assertOutput(
             CollectingDataOutput<Integer> actualOutput, List<Integer> expectedOutput) {
         assertThat(