diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java index bc522d228b3f5..a379a43476494 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSourceSplitMetricGroup.java +++ b/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(); } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java index 05b279b66c17d..6e0331c3d83ac 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java @@ -780,12 +780,16 @@ private void checkSplitWatermarkAlignment() { Collection 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); } }); diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java index f184c3fac14fb..21c00e44b5d3b 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java +++ b/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 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 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 actualOutput, List expectedOutput) { assertThat(