[flink] Support pendingRecords metric for Fluss source.

fluss-client/src/main/java/org/apache/fluss/client/metrics/ScannerMetricGroup.java

@@ -24,6 +24,7 @@
 import org.apache.fluss.metrics.CharacterFilter;
 import org.apache.fluss.metrics.Counter;
 import org.apache.fluss.metrics.DescriptiveStatisticsHistogram;
+import org.apache.fluss.metrics.Gauge;
 import org.apache.fluss.metrics.Histogram;
 import org.apache.fluss.metrics.MeterView;
 import org.apache.fluss.metrics.MetricNames;
@@ -103,6 +104,14 @@ public Counter remoteFetchErrorCount() {
         return remoteFetchErrorCount;
     }
 
+    /**
+     * Registers the gauge for the number of log records that have not been fetched. It must be
+     * called at most once, otherwise the duplicated registration will be ignored with a warning.
+     */
+    public void registerRecordsLagGauge(Gauge<Long> recordsLagGauge) {
+        gauge(MetricNames.SCANNER_RECORDS_LAG, recordsLagGauge);
+    }
+
     public void recordPollStart(long pollStartMs) {
         this.pollStartMs = pollStartMs;
         this.timeMsBetweenPoll = lastPollMs != 0L ? pollStartMs - lastPollMs : 0L;

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogScanner.java

@@ -67,6 +67,7 @@ protected AbstractLogScanner(
         this.logScannerStatus = logScannerStatus;
         this.logFetcher = logFetcher;
         this.scannerMetricGroup = scannerMetricGroup;
+        this.scannerMetricGroup.registerRecordsLagGauge(logScannerStatus::recordsLag);
     }
 
     /**

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java

@@ -23,7 +23,7 @@
 @Internal
 class BucketScanStatus {
     private long offset; // last consumed position
-    private long highWatermark; // the high watermark from last fetch
+    private long highWatermark = -1L; // the high watermark from last fetch, -1 if never fetched
     // TODO add resetStrategy and nextAllowedRetryTimeMs.
 
     public BucketScanStatus() {
@@ -49,4 +49,18 @@ public void setOffset(Long offset) {
     public void setHighWatermark(Long highWatermark) {
         this.highWatermark = highWatermark;
     }
+
+    /**
+     * Returns the number of log records that have not been fetched for this bucket, or 0 if the lag
+     * is unknown, i.e. the offset is still a sentinel offset (like {@link
+     * LogScanner#EARLIEST_OFFSET}) not resolved by any fetch yet, or no high watermark has been
+     * returned by the server yet. The high watermark can also be staler than the offset, in which
+     * case the lag is 0 as well.
+     */
+    long recordsLag() {
+        if (offset < 0 || highWatermark < 0) {
+            return 0L;
+        }
+        return Math.max(highWatermark - offset, 0L);
+    }
 }

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerStatus.java

@@ -65,6 +65,14 @@ synchronized void updateOffset(TableBucket tableBucket, long offset) {
         bucketStatus(tableBucket).setOffset(offset);
     }
 
+    synchronized long recordsLag() {
+        long recordsLag = 0L;
+        for (BucketScanStatus bucketScanStatus : bucketStatusMap.bucketStatusMap().values()) {
+            recordsLag += bucketScanStatus.recordsLag();
+        }
+        return recordsLag;
+    }
+
     synchronized void assignScanBuckets(Map<TableBucket, Long> scanBucketAndOffsets) {
         for (Map.Entry<TableBucket, Long> entry : scanBucketAndOffsets.entrySet()) {
             TableBucket scanBucket = entry.getKey();

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java

@@ -236,11 +236,13 @@ void testShouldContinueConsumeSameCompletedFetchAcrossPolls() throws Exception {
         ScanRecords firstPoll = collector.collectFetch(logFetchBuffer);
         assertThat(firstPoll.records(tb).size()).isEqualTo(2);
         assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(2L);
+        assertThat(logScannerStatus.recordsLag()).isEqualTo(8L);
         assertThat(completedFetch.isConsumed()).isFalse();
 
         ScanRecords secondPoll = collector.collectFetch(logFetchBuffer);
         assertThat(secondPoll.records(tb).size()).isEqualTo(2);
         assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(4L);
+        assertThat(logScannerStatus.recordsLag()).isEqualTo(6L);
         assertThat(completedFetch.isConsumed()).isFalse();
     }
 

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -331,6 +331,7 @@ public class MetricNames {
     public static final String SCANNER_FETCH_LATENCY_MS = "fetchLatencyMs";
     public static final String SCANNER_FETCH_RATE = "fetchRequestsPerSecond";
     public static final String SCANNER_BYTES_PER_REQUEST = "bytesPerRequest";
+    public static final String SCANNER_RECORDS_LAG = "recordsLag";
     public static final String SCANNER_REMOTE_FETCH_BYTES_RATE = "remoteFetchBytesPerSecond";
     public static final String SCANNER_REMOTE_FETCH_RATE = "remoteFetchRequestsPerSecond";
     public static final String SCANNER_REMOTE_FETCH_ERROR_RATE = "remoteFetchErrorPerSecond";

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java

@@ -18,21 +18,28 @@
 package org.apache.fluss.flink.source.metrics;
 
 import org.apache.fluss.flink.source.reader.FlinkSourceReader;
+import org.apache.fluss.metrics.Gauge;
 
 import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
 import org.apache.flink.runtime.metrics.MetricNames;
 
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
 /** A collection class for handling metrics in {@link FlinkSourceReader} of Fluss. */
 public class FlinkSourceReaderMetrics {
 
     public static final long UNINITIALIZED = -1;
 
     // Source reader metric group
     private final SourceReaderMetricGroup sourceReaderMetricGroup;
+    private final Set<Gauge<Long>> recordsLagMetrics = ConcurrentHashMap.newKeySet();
 
     // For currentFetchEventTimeLag metric
     private volatile long currentFetchEventTimeLag = UNINITIALIZED;
 
+    private boolean pendingRecordsGaugeRegistered;
+
     public FlinkSourceReaderMetrics(SourceReaderMetricGroup sourceReaderMetricGroup) {
         this.sourceReaderMetricGroup = sourceReaderMetricGroup;
     }
@@ -50,7 +57,32 @@ public void reportRecordEventTime(long lag) {
         currentFetchEventTimeLag = lag;
     }
 
+    /**
+     * Adds a scanner's records-lag metric to the metrics tracked by the standard Flink
+     * pendingRecords gauge. The Flink gauge itself is registered only once.
+     */
+    public synchronized void maybeAddRecordsLagMetric(Gauge<Long> recordsLagMetric) {
+        recordsLagMetrics.add(recordsLagMetric);
+        if (!pendingRecordsGaugeRegistered) {
+            sourceReaderMetricGroup.setPendingRecordsGauge(this::getPendingRecords);
+            pendingRecordsGaugeRegistered = true;
+        }
+    }
+
+    /** Removes a scanner's records-lag metric from the tracked metrics. */
+    public void removeRecordsLagMetric(Gauge<Long> recordsLagMetric) {
+        recordsLagMetrics.remove(recordsLagMetric);
+    }
+
     public SourceReaderMetricGroup getSourceReaderMetricGroup() {
         return sourceReaderMetricGroup;
     }
+
+    private long getPendingRecords() {
+        long pendingRecords = 0L;
+        for (Gauge<Long> recordsLagMetric : recordsLagMetrics) {
+            pendingRecords += recordsLagMetric.getValue();
+        }
+        return pendingRecords;
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java

@@ -40,6 +40,8 @@
 import org.apache.fluss.lake.source.LakeSplit;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.metrics.Gauge;
+import org.apache.fluss.metrics.MetricNames;
 import org.apache.fluss.predicate.Predicate;
 import org.apache.fluss.types.RowType;
 import org.apache.fluss.utils.CloseableIterator;
@@ -95,6 +97,7 @@ public class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
     @Nullable private final int[] projectedFields;
 
     private final FlinkSourceReaderMetrics flinkSourceReaderMetrics;
+    private final Gauge<Long> recordsLagMetric;
 
     @Nullable private BoundedSplitReader currentBoundedSplitReader;
     @Nullable private SourceSplitBase currentBoundedSplit;
@@ -127,7 +130,9 @@ public FlinkSourceSplitReader(
             @Nullable LakeSource<LakeSplit> lakeSource,
             FlinkSourceReaderMetrics flinkSourceReaderMetrics) {
         this.flinkMetricRegistry =
-                new FlinkMetricRegistry(flinkSourceReaderMetrics.getSourceReaderMetricGroup());
+                new FlinkMetricRegistry(
+                        flinkSourceReaderMetrics.getSourceReaderMetricGroup(),
+                        Collections.singleton(MetricNames.SCANNER_RECORDS_LAG));
         this.connection = ConnectionFactory.createConnection(flussConf, flinkMetricRegistry);
         this.table = connection.getTable(tablePath);
         this.tableId = table.getTableInfo().getTableId();
@@ -146,6 +151,16 @@ public FlinkSourceSplitReader(
         this.stoppingOffsets = new HashMap<>();
         this.emptyLogSplits = new HashSet<>();
         this.lakeSource = lakeSource;
+
+        @SuppressWarnings("unchecked")
+        Gauge<Long> recordsLagMetric =
+                (Gauge<Long>)
+                        checkNotNull(
+                                flinkMetricRegistry.getFlussMetric(MetricNames.SCANNER_RECORDS_LAG),
+                                "The gauge %s should have been registered by the log scanner.",
+                                MetricNames.SCANNER_RECORDS_LAG);
+        this.recordsLagMetric = recordsLagMetric;
+        flinkSourceReaderMetrics.maybeAddRecordsLagMetric(recordsLagMetric);
     }
 
     @Override
@@ -630,6 +645,7 @@ public void wakeUp() {
 
     @Override
     public void close() throws Exception {
+        flinkSourceReaderMetrics.removeRecordsLagMetric(recordsLagMetric);
         if (currentBoundedSplitReader != null) {
             currentBoundedSplitReader.close();
         }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/fetcher/FlinkSourceFetcherManager.java

@@ -77,7 +77,7 @@ public FlinkSourceFetcherManager(
     public void removePartitions(
             Map<Long, String> removedPartitions,
             Consumer<Set<TableBucket>> unsubscribeTableBucketsCallback) {
-        SplitFetcher<RecordAndPos, SourceSplitBase> splitFetcher = fetchers.get(0);
+        SplitFetcher<RecordAndPos, SourceSplitBase> splitFetcher = getRunningFetcher();
         if (splitFetcher != null) {
             // The fetcher thread is still running. This should be the majority of the cases.
             enqueuePartitionsRemovedTask(

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetricsTest.java

@@ -24,6 +24,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -49,4 +50,39 @@ void testCurrentFetchEventTimeLag() {
         flinkSourceReaderMetrics.reportRecordEventTime(18213L);
         assertThat((long) currentFetchEventTimeLag.get().getValue()).isEqualTo(18213L);
     }
+
+    @Test
+    void testPendingRecordsGaugeTracksActiveScanners() {
+        MetricListener metricListener = new MetricListener();
+        FlinkSourceReaderMetrics flinkSourceReaderMetrics =
+                new FlinkSourceReaderMetrics(
+                        InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
+
+        // the metric is not registered until a log scanner provides its records lag
+        assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS)).isEmpty();
+
+        AtomicLong firstRecordsLag = new AtomicLong(10L);
+        org.apache.fluss.metrics.Gauge<Long> firstRecordsLagMetric = firstRecordsLag::get;
+        flinkSourceReaderMetrics.maybeAddRecordsLagMetric(firstRecordsLagMetric);
+        Optional<Gauge<Long>> pendingRecords = metricListener.getGauge(MetricNames.PENDING_RECORDS);
+        assertThat(pendingRecords).isPresent();
+        assertThat((long) pendingRecords.get().getValue()).isEqualTo(10L);
+
+        AtomicLong secondRecordsLag = new AtomicLong(7L);
+        org.apache.fluss.metrics.Gauge<Long> secondRecordsLagMetric = secondRecordsLag::get;
+        flinkSourceReaderMetrics.maybeAddRecordsLagMetric(secondRecordsLagMetric);
+        assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS).get())
+                .isSameAs(pendingRecords.get());
+        assertThat((long) pendingRecords.get().getValue()).isEqualTo(17L);
+
+        firstRecordsLag.set(3L);
+        assertThat((long) pendingRecords.get().getValue()).isEqualTo(10L);
+
+        flinkSourceReaderMetrics.removeRecordsLagMetric(firstRecordsLagMetric);
+        secondRecordsLag.set(4L);
+        assertThat((long) pendingRecords.get().getValue()).isEqualTo(4L);
+
+        flinkSourceReaderMetrics.removeRecordsLagMetric(secondRecordsLagMetric);
+        assertThat((long) pendingRecords.get().getValue()).isEqualTo(0L);
+    }
 }