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 bc522d228b3f5a..a379a434764940 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 f153617b2c4feb..c487aba5255e94 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 @@ -848,12 +848,16 @@ private void checkSplitWatermarkAlignment() { Collection 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); } }); 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 b14639a9fcc31f..a635064d949d9b 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 @@ -569,6 +569,58 @@ 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<>(); + + // Emit enough records to fill the sampler buffer + for (int i = 0; i < WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue(); i++) { + operator.emitNext(actualOutput); + processingTimeService.advance(idleTimeout - 1); + } + sampleAllWatermarks(processingTimeService); + 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 sampleAllWatermarks(TestProcessingTimeService timeService) throws Exception { sampleWatermarks(timeService, WATERMARK_ALIGNMENT_BUFFER_SIZE.defaultValue()); }