support flink sourceIdleTime metric

herefree · Nov 26, 2024 · cb9f852 · cb9f852
1 parent dc9849f
commit cb9f852
Show file tree

Hide file tree

Showing 4 changed files with 55 additions and 0 deletions.
diff --git a/...on/src/main/java/org/apache/paimon/flink/source/metrics/FileStoreSourceReaderMetrics.java b/...on/src/main/java/org/apache/paimon/flink/source/metrics/FileStoreSourceReaderMetrics.java
@@ -27,18 +27,22 @@ public class FileStoreSourceReaderMetrics {
 
     private long latestFileCreationTime = UNDEFINED;
     private long lastSplitUpdateTime = UNDEFINED;
+    private long idleStartTime = ACTIVE;
 
     public static final long UNDEFINED = -1L;
+    private static final long ACTIVE = Long.MAX_VALUE;
 
     public FileStoreSourceReaderMetrics(MetricGroup sourceReaderMetricGroup) {
         sourceReaderMetricGroup.gauge(
                 MetricNames.CURRENT_FETCH_EVENT_TIME_LAG, this::getFetchTimeLag);
+        sourceReaderMetricGroup.gauge(MetricNames.SOURCE_IDLE_TIME, this::getIdleTime);
     }
 
     /** Called when consumed snapshot changes. */
     public void recordSnapshotUpdate(long fileCreationTime) {
         this.latestFileCreationTime = fileCreationTime;
         lastSplitUpdateTime = System.currentTimeMillis();
+        idleStartTime = ACTIVE;
     }
 
     @VisibleForTesting
@@ -57,4 +61,19 @@ public long getLatestFileCreationTime() {
     long getLastSplitUpdateTime() {
         return lastSplitUpdateTime;
     }
+
+    public void idlingStarted() {
+        if (!isIdling()) {
+            idleStartTime = System.currentTimeMillis();
+        }
+    }
+
+    boolean isIdling() {
+        return idleStartTime != ACTIVE;
+    }
+
+    @VisibleForTesting
+    long getIdleTime() {
+        return isIdling() ? System.currentTimeMillis() - idleStartTime : 0;
+    }
 }
diff --git a/...imon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java b/...imon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
@@ -83,6 +83,7 @@ public void open() throws Exception {
         this.read = readBuilder.newRead().withIOManager(ioManager);
         this.reuseRow = new FlinkRowData(null);
         this.reuseRecord = new StreamRecord<>(reuseRow);
+        this.sourceReaderMetrics.idlingStarted();
     }
 
     @Override
@@ -113,6 +114,7 @@ public void processElement(StreamRecord<Split> record) throws Exception {
                 output.collect(reuseRecord);
             }
         }
+        sourceReaderMetrics.idlingStarted();
     }
 
     @Override

diff --git a/...rc/test/java/org/apache/paimon/flink/source/metrics/FileStoreSourceReaderMetricsTest.java b/...rc/test/java/org/apache/paimon/flink/source/metrics/FileStoreSourceReaderMetricsTest.java
@@ -52,4 +52,28 @@ public void testCurrentFetchLagUpdated() {
         assertThat(sourceReaderMetrics.getFetchTimeLag())
                 .isNotEqualTo(FileStoreSourceReaderMetrics.UNDEFINED);
     }
+
+    @Test
+    public void testSourceIdleTimeUpdated() throws InterruptedException {
+        MetricListener metricListener = new MetricListener();
+        final FileStoreSourceReaderMetrics sourceReaderMetrics =
+                new FileStoreSourceReaderMetrics(metricListener.getMetricGroup());
+
+        assertThat(sourceReaderMetrics.getIdleTime()).isEqualTo(0L);
+
+        // idle start
+        sourceReaderMetrics.idlingStarted();
+        Thread.sleep(10L);
+        assertThat(sourceReaderMetrics.getIdleTime()).isGreaterThan(9L);
+
+        //non-idle
+        sourceReaderMetrics.recordSnapshotUpdate(123);
+        Thread.sleep(10L);
+        assertThat(sourceReaderMetrics.getIdleTime()).isEqualTo(0L);
+
+        // idle start
+        sourceReaderMetrics.idlingStarted();
+        Thread.sleep(10L);
+        assertThat(sourceReaderMetrics.getIdleTime()).isGreaterThan(9L);
+    }
 }
diff --git a/...link-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java b/...link-common/src/test/java/org/apache/paimon/flink/source/operator/OperatorSourceTest.java
@@ -204,6 +204,11 @@ public void testReadOperatorMetricsRegisterAndUpdate() throws Exception {
                                 .getValue())
                 .isEqualTo(-1L);
 
+        Thread.sleep(300L);
+        assertThat((Long)TestingMetricUtils.getGauge(
+                        readerOperatorMetricGroup, "sourceIdleTime")
+                .getValue()).isGreaterThan(299L);
+
         harness.processElement(new StreamRecord<>(splits.get(0)));
         assertThat(
                         (Long)
@@ -228,6 +233,11 @@ public void testReadOperatorMetricsRegisterAndUpdate() throws Exception {
                                                 "currentEmitEventTimeLag")
                                         .getValue())
                 .isEqualTo(emitEventTimeLag);
+
+        assertThat((Long)TestingMetricUtils.getGauge(
+                        readerOperatorMetricGroup, "sourceIdleTime")
+                .getValue()).isGreaterThan(99L).isLessThan(300L);
+
     }
 
     private <T> T testReadSplit(