fix

herefree · Dec 6, 2024 · cf45da4 · cf45da4
1 parent dc8823a
commit cf45da4
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 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
@@ -29,6 +29,7 @@ public class FileStoreSourceReaderMetrics {
     private long lastSplitUpdateTime = UNDEFINED;
 
     public static final long UNDEFINED = -1L;
+    public static final long ACTIVE = Long.MAX_VALUE;
 
     public FileStoreSourceReaderMetrics(MetricGroup sourceReaderMetricGroup) {
         sourceReaderMetricGroup.gauge(

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
@@ -58,7 +58,7 @@ public class ReadOperator extends AbstractStreamOperator<RowData>
     // is not a FLIP-27
     // source and Flink can't automatically calculate this metric
     private transient long emitEventTimeLag = FileStoreSourceReaderMetrics.UNDEFINED;
-    private transient long idleStartTime = FileStoreSourceReaderMetrics.UNDEFINED;
+    private transient long idleStartTime = FileStoreSourceReaderMetrics.ACTIVE;
     private transient Counter numRecordsIn;
 
     public ReadOperator(ReadBuilder readBuilder) {
@@ -99,7 +99,7 @@ public void processElement(StreamRecord<Split> record) throws Exception {
                         .orElse(FileStoreSourceReaderMetrics.UNDEFINED);
         sourceReaderMetrics.recordSnapshotUpdate(eventTime);
         // update idleStartTime when reading a new split
-        idleStartTime = FileStoreSourceReaderMetrics.UNDEFINED;
+        idleStartTime = FileStoreSourceReaderMetrics.ACTIVE;
 
         boolean firstRecord = true;
         try (CloseableIterator<InternalRow> iterator =
@@ -138,7 +138,7 @@ private void idlingStarted() {
     }
 
     private boolean isIdling() {
-        return idleStartTime != FileStoreSourceReaderMetrics.UNDEFINED;
+        return idleStartTime != FileStoreSourceReaderMetrics.ACTIVE;
     }
 
     private long getIdleTime() {