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..36fda4eee681d3 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,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 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 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()); }