You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2024/06/18 12:16:43 UTC
(flink) branch release-1.18 updated (72fac4c8ebd -> b60318ba1a7)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
from 72fac4c8ebd [FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish (#24757)
new fb5f516ce34 [FLINK-34252][table] Fix lastRecordTime tracking in WatermarkAssignerOperator to prevent erroneous idle WatermarkStatus
new b60318ba1a7 fixup! [FLINK-34252][table] Optimize WatermarkAssignerOperator to avoid calling System.currentTimeMillis per every record
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../wmassigners/WatermarkAssignerOperator.java | 92 ++++++++++++++++++----
.../wmassigners/WatermarkAssignerOperatorTest.java | 74 ++++++++++++++++-
.../WatermarkAssignerOperatorTestBase.java | 8 ++
3 files changed, 154 insertions(+), 20 deletions(-)
(flink) 02/02: fixup! [FLINK-34252][table] Optimize WatermarkAssignerOperator to avoid calling System.currentTimeMillis per every record
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b60318ba1a79239a74b1d7d61b447fc0c60e8429
Author: Piotr Nowojski <pi...@gmail.com>
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 25e3626b68f..0adac654998 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.util.FunctionUtils;
import org.apache.flink.configuration.Configuration;
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 bf373eb563e..8080e4080d8 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(
(flink) 01/02: [FLINK-34252][table] Fix lastRecordTime tracking in WatermarkAssignerOperator to prevent erroneous idle WatermarkStatus
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fb5f516ce34f5ea0eb688f2399f05e9c043fe407
Author: David Christle <da...@discordapp.com>
AuthorDate: Sun Jan 28 00:04:09 2024 -0800
[FLINK-34252][table] Fix lastRecordTime tracking in WatermarkAssignerOperator to prevent erroneous idle WatermarkStatus
---
.../wmassigners/WatermarkAssignerOperator.java | 9 +++++---
.../wmassigners/WatermarkAssignerOperatorTest.java | 25 ++++++++++++++++++++++
.../WatermarkAssignerOperatorTestBase.java | 8 +++++++
3 files changed, 39 insertions(+), 3 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 90559a5ce26..25e3626b68f 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
@@ -100,11 +100,14 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
@Override
public void processElement(StreamRecord<RowData> element) throws Exception {
- if (idleTimeout > 0 && currentStatus.equals(WatermarkStatus.IDLE)) {
- // mark the channel active
- emitWatermarkStatus(WatermarkStatus.ACTIVE);
+ if (idleTimeout > 0) {
+ if (currentStatus.equals(WatermarkStatus.IDLE)) {
+ // mark the channel active
+ emitWatermarkStatus(WatermarkStatus.ACTIVE);
+ }
lastRecordTime = getProcessingTimeService().getCurrentProcessingTime();
}
+
RowData row = element.getValue();
if (row.isNullAt(rowtimeFieldIndex)) {
throw new RuntimeException(
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 bb8e9aa3459..bf373eb563e 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
@@ -84,6 +84,31 @@ public class WatermarkAssignerOperatorTest extends WatermarkAssignerOperatorTest
assertThat(filterOutRecords(output)).isEqualTo(expectedOutput);
}
+ @Test
+ public void testIdleStateAvoidanceWithConsistentDataFlow() throws Exception {
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
+ createTestHarness(0, WATERMARK_GENERATOR, 1000);
+ testHarness.getExecutionConfig().setAutoWatermarkInterval(50);
+ testHarness.open();
+
+ ConcurrentLinkedQueue<Object> output = testHarness.getOutput();
+ List<Object> expectedOutput = new ArrayList<>();
+
+ // Process elements at intervals less than idleTimeout (1000ms)
+ for (long i = 1; i <= 10; i++) {
+ long timestamp = i * 900;
+ 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));
+ }
+
+ // Check if the status ever becomes IDLE (it shouldn't)
+ assertThat(extractWatermarkStatuses(output)).doesNotContain(WatermarkStatus.IDLE);
+ assertThat(filterOutRecords(output)).isEqualTo(expectedOutput);
+ }
+
@Test
public void testWatermarkAssignerOperator() throws Exception {
OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTestBase.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTestBase.java
index 1aeba2e6c86..6f64d2c6b24 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTestBase.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTestBase.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.table.data.RowData;
import java.util.ArrayList;
@@ -60,6 +61,13 @@ public abstract class WatermarkAssignerOperatorTestBase {
return watermarks;
}
+ protected List<WatermarkStatus> extractWatermarkStatuses(Collection<Object> collection) {
+ return collection.stream()
+ .filter(obj -> obj instanceof WatermarkStatus)
+ .map(obj -> (WatermarkStatus) obj)
+ .collect(Collectors.toList());
+ }
+
protected List<Object> filterOutRecords(Collection<Object> collection) {
return collection.stream()
.filter(obj -> !(obj instanceof StreamElement && ((StreamElement) obj).isRecord()))