This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new bd156e900 [flink] Emit event time lag should not increase when no new
records are read (#4281)
bd156e900 is described below
commit bd156e900d786d61b4332ef07277de642efbbd11
Author: tsreaper <[email protected]>
AuthorDate: Thu Oct 3 23:14:50 2024 +0800
[flink] Emit event time lag should not increase when no new records are
read (#4281)
---
.../paimon/flink/source/operator/ReadOperator.java | 19 ++++++-------------
.../flink/source/operator/OperatorSourceTest.java | 12 +++++++++++-
2 files changed, 17 insertions(+), 14 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
index 97c20ed79..80c85f7cd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
@@ -54,6 +54,9 @@ public class ReadOperator extends
AbstractStreamOperator<RowData>
private transient IOManager ioManager;
private transient FileStoreSourceReaderMetrics sourceReaderMetrics;
+ // we create our own gauge for currentEmitEventTimeLag, because this
operator is not a FLIP-27
+ // source and Flink can't automatically calculate this metric
+ private transient long emitEventTimeLag =
FileStoreSourceReaderMetrics.UNDEFINED;
private transient Counter numRecordsIn;
public ReadOperator(ReadBuilder readBuilder) {
@@ -65,19 +68,7 @@ public class ReadOperator extends
AbstractStreamOperator<RowData>
super.open();
this.sourceReaderMetrics = new
FileStoreSourceReaderMetrics(getMetricGroup());
- // we create our own gauge for currentEmitEventTimeLag, because this
operator is not a
- // FLIP-27 source and Flink can't automatically calculate this metric
- getMetricGroup()
- .gauge(
- MetricNames.CURRENT_EMIT_EVENT_TIME_LAG,
- () -> {
- long eventTime =
sourceReaderMetrics.getLatestFileCreationTime();
- if (eventTime ==
FileStoreSourceReaderMetrics.UNDEFINED) {
- return FileStoreSourceReaderMetrics.UNDEFINED;
- } else {
- return System.currentTimeMillis() - eventTime;
- }
- });
+ getMetricGroup().gauge(MetricNames.CURRENT_EMIT_EVENT_TIME_LAG, () ->
emitEventTimeLag);
this.numRecordsIn =
InternalSourceReaderMetricGroup.wrap(getMetricGroup())
.getIOMetricGroup()
@@ -108,6 +99,8 @@ public class ReadOperator extends
AbstractStreamOperator<RowData>
try (CloseableIterator<InternalRow> iterator =
read.createReader(split).toCloseableIterator()) {
while (iterator.hasNext()) {
+ emitEventTimeLag = System.currentTimeMillis() - eventTime;
+
// each Split is already counted as one input record,
// so we don't need to count the first record
if (firstRecord) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
index 0f6d66815..61a03a29a 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
@@ -203,6 +203,7 @@ public class OperatorSourceTest {
readerOperatorMetricGroup,
"currentEmitEventTimeLag")
.getValue())
.isEqualTo(-1L);
+
harness.processElement(new StreamRecord<>(splits.get(0)));
assertThat(
(Long)
@@ -211,13 +212,22 @@ public class OperatorSourceTest {
"currentFetchEventTimeLag")
.getValue())
.isGreaterThan(0);
+ long emitEventTimeLag =
+ (Long)
+ TestingMetricUtils.getGauge(
+ readerOperatorMetricGroup,
"currentEmitEventTimeLag")
+ .getValue();
+ assertThat(emitEventTimeLag).isGreaterThan(0);
+
+ // wait for a while and read metrics again, metrics should not change
+ Thread.sleep(100);
assertThat(
(Long)
TestingMetricUtils.getGauge(
readerOperatorMetricGroup,
"currentEmitEventTimeLag")
.getValue())
- .isGreaterThan(0);
+ .isEqualTo(emitEventTimeLag);
}
private <T> T testReadSplit(
