This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit d93e92c665af50f92ff3f25b68ca3943ad2699fa Author: Piotr Nowojski <[email protected]> AuthorDate: Fri Jun 14 15:12:26 2024 +0200 fixup! [FLINK-34252][table] Optimize WatermarkAssignerOperator to avoid calling System.currentTimeMillis per every record --- .../wmassigners/WatermarkAssignerOperator.java | 97 +++++++++++++++++----- .../wmassigners/WatermarkAssignerOperatorTest.java | 69 ++++++++++++--- 2 files changed, 132 insertions(+), 34 deletions(-) diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java index 9a7a5067973..5fd7c8a22e1 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java @@ -18,6 +18,7 @@ package org.apache.flink.table.runtime.operators.wmassigners; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.functions.DefaultOpenContext; import org.apache.flink.api.common.functions.util.FunctionUtils; import org.apache.flink.streaming.api.operators.AbstractStreamOperator; @@ -30,6 +31,7 @@ import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus; import org.apache.flink.table.data.RowData; import org.apache.flink.table.runtime.generated.WatermarkGenerator; +import static org.apache.flink.util.Preconditions.checkArgument; import static org.apache.flink.util.Preconditions.checkNotNull; /** @@ -52,12 +54,22 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData> private transient long watermarkInterval; + private transient long timerInterval; + private transient long currentWatermark; - private transient long lastRecordTime; + // Last time watermark have been (periodically) emitted + private transient long lastWatermarkPeriodicEmitTime; + + // Last time idleness status has been checked + private transient long timeSinceLastIdleCheck; private transient WatermarkStatus currentStatus = WatermarkStatus.ACTIVE; + private transient long processedElements; + + private transient long lastIdleCheckProcessedElements = -1; + /** * Create a watermark assigner operator. * @@ -87,11 +99,14 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData> // watermark and timestamp should start from 0 this.currentWatermark = 0; this.watermarkInterval = getExecutionConfig().getAutoWatermarkInterval(); - this.lastRecordTime = getProcessingTimeService().getCurrentProcessingTime(); + long now = getProcessingTimeService().getCurrentProcessingTime(); + this.lastWatermarkPeriodicEmitTime = now; + this.timeSinceLastIdleCheck = now; - if (watermarkInterval > 0) { - long now = getProcessingTimeService().getCurrentProcessingTime(); - getProcessingTimeService().registerTimer(now + watermarkInterval, this); + if (watermarkInterval > 0 || idleTimeout > 0) { + this.timerInterval = + calculateProcessingTimeTimerInterval(watermarkInterval, idleTimeout); + getProcessingTimeService().registerTimer(now + timerInterval, this); } FunctionUtils.setFunctionRuntimeContext(watermarkGenerator, getRuntimeContext()); @@ -100,12 +115,11 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData> @Override public void processElement(StreamRecord<RowData> element) throws Exception { - if (idleTimeout > 0) { - if (currentStatus.equals(WatermarkStatus.IDLE)) { - // mark the channel active - emitWatermarkStatus(WatermarkStatus.ACTIVE); - } - lastRecordTime = getProcessingTimeService().getCurrentProcessingTime(); + processedElements++; + + if (isIdlenessEnabled() && currentStatus.equals(WatermarkStatus.IDLE)) { + // mark the channel active + emitWatermarkStatus(WatermarkStatus.ACTIVE); } RowData row = element.getValue(); @@ -139,19 +153,28 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData> @Override public void onProcessingTime(long timestamp) throws Exception { - advanceWatermark(); + // timestamp and now can be off in case TM is heavily overloaded. + long now = getProcessingTimeService().getCurrentProcessingTime(); - if (idleTimeout > 0 && currentStatus.equals(WatermarkStatus.ACTIVE)) { - final long currentTime = getProcessingTimeService().getCurrentProcessingTime(); - if (currentTime - lastRecordTime > idleTimeout) { - // mark the channel as idle to ignore watermarks from this channel - emitWatermarkStatus(WatermarkStatus.IDLE); - } + if (watermarkInterval > 0 && lastWatermarkPeriodicEmitTime + watermarkInterval <= now) { + lastWatermarkPeriodicEmitTime = now; + advanceWatermark(); + } + + if (processedElements != lastIdleCheckProcessedElements) { + timeSinceLastIdleCheck = now; + lastIdleCheckProcessedElements = processedElements; + } + + if (isIdlenessEnabled() + && currentStatus.equals(WatermarkStatus.ACTIVE) + && timeSinceLastIdleCheck + idleTimeout <= now) { + // mark the channel as idle to ignore watermarks from this channel + emitWatermarkStatus(WatermarkStatus.IDLE); } // register next timer - long now = getProcessingTimeService().getCurrentProcessingTime(); - getProcessingTimeService().registerTimer(now + watermarkInterval, this); + getProcessingTimeService().registerTimer(now + timerInterval, this); } /** @@ -163,7 +186,7 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData> // if we receive a Long.MAX_VALUE watermark we forward it since it is used // to signal the end of input and to not block watermark progress downstream if (mark.getTimestamp() == Long.MAX_VALUE && currentWatermark != Long.MAX_VALUE) { - if (idleTimeout > 0 && currentStatus.equals(WatermarkStatus.IDLE)) { + if (isIdlenessEnabled() && currentStatus.equals(WatermarkStatus.IDLE)) { // mark the channel active emitWatermarkStatus(WatermarkStatus.ACTIVE); } @@ -193,4 +216,36 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData> FunctionUtils.closeFunction(watermarkGenerator); super.close(); } + + private boolean isIdlenessEnabled() { + return idleTimeout > 0; + } + + @VisibleForTesting + static long calculateProcessingTimeTimerInterval(long watermarkInterval, long idleTimeout) { + checkArgument(watermarkInterval > 0 || idleTimeout > 0); + if (watermarkInterval <= 0) { + return idleTimeout; + } + if (idleTimeout <= 0) { + return watermarkInterval; + } + + long smallerInterval = Math.min(watermarkInterval, idleTimeout); + long largerInterval = Math.max(watermarkInterval, idleTimeout); + + // If one of the intervals is 5x smaller, just pick the smaller one. The firing interval + // for the smaller one this way will be perfectly accurate, while for the larger one it will + // be good enough™. For example one timer is every 2s the other every 11s, the 2nd timer + // will be effectively checked every 12s, which is an acceptable accuracy. + long timerInterval; + if (smallerInterval * 5 < largerInterval) { + timerInterval = smallerInterval; + } else { + // Otherwise, just pick an interval 5x smaller than the smaller interval. Again accuracy + // will be good enough™. + timerInterval = smallerInterval / 5; + } + return Math.max(timerInterval, 1); + } } diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java index e5deacb3999..9cd558629bc 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java @@ -37,6 +37,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; +import static org.apache.flink.table.runtime.operators.wmassigners.WatermarkAssignerOperator.calculateProcessingTimeTimerInterval; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; @@ -46,6 +47,24 @@ public class WatermarkAssignerOperatorTest extends WatermarkAssignerOperatorTest private static final WatermarkGenerator WATERMARK_GENERATOR = new BoundedOutOfOrderWatermarkGenerator(0, 1); + @Test + public void testCalculateProcessingTimeTimerInterval() { + assertThat(calculateProcessingTimeTimerInterval(5, 0)).isEqualTo(5); + assertThat(calculateProcessingTimeTimerInterval(5, -1)).isEqualTo(5); + + assertThat(calculateProcessingTimeTimerInterval(0, 5)).isEqualTo(5); + assertThat(calculateProcessingTimeTimerInterval(-1, 5)).isEqualTo(5); + + assertThat(calculateProcessingTimeTimerInterval(5, 42)).isEqualTo(5); + assertThat(calculateProcessingTimeTimerInterval(42, 5)).isEqualTo(5); + + assertThat(calculateProcessingTimeTimerInterval(2, 4)).isEqualTo(1); + assertThat(calculateProcessingTimeTimerInterval(4, 2)).isEqualTo(1); + + assertThat(calculateProcessingTimeTimerInterval(100, 110)).isEqualTo(20); + assertThat(calculateProcessingTimeTimerInterval(110, 100)).isEqualTo(20); + } + @Test public void testWatermarkAssignerWithIdleSource() throws Exception { // with timeout 1000 ms @@ -68,7 +87,10 @@ public class WatermarkAssignerOperatorTest extends WatermarkAssignerOperatorTest expectedOutput.add(new Watermark(3)); assertThat(filterOutRecords(output)).isEqualTo(expectedOutput); - testHarness.setProcessingTime(1001); + stepProcessingTime(testHarness, 52, 1050, 50); + assertThat(filterOutRecords(output)).isEqualTo(expectedOutput); + + stepProcessingTime(testHarness, 1051, 1100, 50); expectedOutput.add(WatermarkStatus.IDLE); assertThat(filterOutRecords(output)).isEqualTo(expectedOutput); @@ -79,34 +101,55 @@ public class WatermarkAssignerOperatorTest extends WatermarkAssignerOperatorTest testHarness.processElement(new StreamRecord<>(GenericRowData.of(7L))); testHarness.processElement(new StreamRecord<>(GenericRowData.of(8L))); - testHarness.setProcessingTime(1060); + assertThat(filterOutRecords(output)).isEqualTo(expectedOutput); + + stepProcessingTime(testHarness, 1101, 1200, 50); expectedOutput.add(new Watermark(7)); assertThat(filterOutRecords(output)).isEqualTo(expectedOutput); } @Test - public void testIdleStateAvoidanceWithConsistentDataFlow() throws Exception { + public void testWatermarkIntervalSmallerThanIdleTimeout() throws Exception { + testIdleTimeout(1000, 50); + } + + @Test + public void testIdleTimeoutSmallerThanWatermarkInterval() throws Exception { + testIdleTimeout(50, 1000); + } + + private void testIdleTimeout(long idleTimeout, long watermarkInterval) throws Exception { + long step = Math.min(idleTimeout, watermarkInterval); OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = - createTestHarness(0, WATERMARK_GENERATOR, 1000); - testHarness.getExecutionConfig().setAutoWatermarkInterval(50); + createTestHarness(0, WATERMARK_GENERATOR, idleTimeout); + testHarness.getExecutionConfig().setAutoWatermarkInterval(watermarkInterval); testHarness.open(); ConcurrentLinkedQueue<Object> output = testHarness.getOutput(); - List<Object> expectedOutput = new ArrayList<>(); + long timeBetweenRecords = (long) (idleTimeout * 0.9); // Process elements at intervals less than idleTimeout (1000ms) for (long i = 1; i <= 10; i++) { - long timestamp = i * 900; + long timestamp = i * timeBetweenRecords; testHarness.processElement(new StreamRecord<>(GenericRowData.of(timestamp), timestamp)); - testHarness.setProcessingTime(timestamp); - - // Expect a watermark for each element, lagging by 1 - expectedOutput.add(new Watermark(timestamp - 1)); + stepProcessingTime(testHarness, timestamp, timestamp + timeBetweenRecords - 1, step); } // Check if the status ever becomes IDLE (it shouldn't) assertThat(extractWatermarkStatuses(output)).doesNotContain(WatermarkStatus.IDLE); - assertThat(filterOutRecords(output)).isEqualTo(expectedOutput); + } + + private void stepProcessingTime( + OneInputStreamOperatorTestHarness<?, ?> testHarness, + long fromInclusive, + long toInclusive, + long step) + throws Exception { + for (long time = fromInclusive; time < toInclusive; time += step) { + // incrementally fire processing time timers + testHarness.setProcessingTime(time); + } + testHarness.setProcessingTime(toInclusive); } @Test @@ -238,11 +281,11 @@ public class WatermarkAssignerOperatorTest extends WatermarkAssignerOperatorTest expected.add(Watermark.MAX_WATERMARK); // num_watermark + num_records - assertThat(testHarness.getOutput()).hasSize(expected.size() + 11); List<Watermark> results = extractWatermarks(testHarness.getOutput()); assertThat(results).isEqualTo(expected); assertThat(MyWatermarkGenerator.openCalled).isTrue(); assertThat(MyWatermarkGenerator.closeCalled).isTrue(); + assertThat(testHarness.getOutput()).hasSize(expected.size() + 11); } private static OneInputStreamOperatorTestHarness<RowData, RowData> createTestHarness(
