You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ho...@apache.org on 2026/09/10 11:36:37 UTC
(fluss) branch main updated: [flink] Support pendingRecords metric for Fluss source. (#4250)
This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new dc49862ace [flink] Support pendingRecords metric for Fluss source. (#4250)
dc49862ace is described below
commit dc49862ace830273a74c32e1bf69b80f05e9035b
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Thu Sep 10 19:36:32 2026 +0800
[flink] Support pendingRecords metric for Fluss source. (#4250)
---
.../fluss/client/metrics/ScannerMetricGroup.java | 9 ++
.../table/scanner/log/AbstractLogScanner.java | 1 +
.../client/table/scanner/log/BucketScanStatus.java | 16 +-
.../client/table/scanner/log/LogScannerStatus.java | 8 +
.../table/scanner/log/LogFetchCollectorTest.java | 2 +
.../java/org/apache/fluss/metrics/MetricNames.java | 1 +
.../source/metrics/FlinkSourceReaderMetrics.java | 32 ++++
.../source/reader/FlinkSourceSplitReader.java | 18 ++-
.../reader/fetcher/FlinkSourceFetcherManager.java | 2 +-
.../metrics/FlinkSourceReaderMetricsTest.java | 36 +++++
.../flink/source/reader/FlinkSourceReaderTest.java | 167 ++++++++++++++++++++-
.../source/reader/FlinkSourceSplitReaderTest.java | 144 ++++++++++++++++++
.../maintenance/observability/monitor-metrics.md | 6 +
13 files changed, 434 insertions(+), 8 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metrics/ScannerMetricGroup.java b/fluss-client/src/main/java/org/apache/fluss/client/metrics/ScannerMetricGroup.java
index bae44d00ed..279db462c2 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/metrics/ScannerMetricGroup.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/metrics/ScannerMetricGroup.java
@@ -24,6 +24,7 @@ import org.apache.fluss.metadata.TablePath;
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 class ScannerMetricGroup extends AbstractMetricGroup {
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;
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogScanner.java
index 8e16b35ef7..60ac2acbb1 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogScanner.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogScanner.java
@@ -67,6 +67,7 @@ abstract class AbstractLogScanner<R> implements AutoCloseable {
this.logScannerStatus = logScannerStatus;
this.logFetcher = logFetcher;
this.scannerMetricGroup = scannerMetricGroup;
+ this.scannerMetricGroup.registerRecordsLagGauge(logScannerStatus::recordsLag);
}
/**
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java
index 738e04a49e..c99ae78387 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/BucketScanStatus.java
@@ -23,7 +23,7 @@ import org.apache.fluss.annotation.Internal;
@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 @@ class BucketScanStatus {
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);
+ }
}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerStatus.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerStatus.java
index c3ea600cda..8ae1391a9a 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerStatus.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerStatus.java
@@ -65,6 +65,14 @@ public class LogScannerStatus {
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();
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
index aafad8c0ee..064512a568 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
@@ -236,11 +236,13 @@ public class LogFetchCollectorTest {
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();
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
index 644b11fd7e..766e0e9abc 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
+++ b/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";
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java
index 400f1969fa..124635635e 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetrics.java
@@ -18,10 +18,14 @@
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 {
@@ -29,10 +33,13 @@ public class FlinkSourceReaderMetrics {
// 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 class FlinkSourceReaderMetrics {
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;
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
index 1fff429217..692775ede4 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
+++ b/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.LakeSource;
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 class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
@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 class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
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 class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
@Override
public void close() throws Exception {
+ flinkSourceReaderMetrics.removeRecordsLagMetric(recordsLagMetric);
if (currentBoundedSplitReader != null) {
currentBoundedSplitReader.close();
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/fetcher/FlinkSourceFetcherManager.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/fetcher/FlinkSourceFetcherManager.java
index 9b1c9ff0fd..fdbe0fd08e 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/fetcher/FlinkSourceFetcherManager.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/fetcher/FlinkSourceFetcherManager.java
@@ -77,7 +77,7 @@ public class 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(
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetricsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetricsTest.java
index f7bb7f2fcf..b40f5f0dba 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetricsTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/metrics/FlinkSourceReaderMetricsTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
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 @@ class FlinkSourceReaderMetricsTest {
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);
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java
index 91b685deac..7ae9a32c7b 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java
@@ -21,6 +21,7 @@ import org.apache.fluss.client.metadata.KvSnapshots;
import org.apache.fluss.client.table.Table;
import org.apache.fluss.client.table.writer.UpsertResult;
import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
import org.apache.fluss.flink.source.deserializer.DeserializerInitContextImpl;
@@ -56,6 +57,10 @@ import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.MetricNames;
+import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
import org.apache.flink.table.data.RowData;
import org.junit.jupiter.api.Test;
@@ -68,6 +73,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
@@ -409,6 +415,75 @@ class FlinkSourceReaderTest extends FlinkTestBase {
}
}
+ @Test
+ void testPendingRecordsMetricAfterFetcherRecreation() throws Exception {
+ TablePath tablePath =
+ TablePath.of(DEFAULT_DB, "test_pending_records_after_fetcher_recreation");
+ TableDescriptor tableDescriptor = DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR;
+ long tableId = createTable(tablePath, tableDescriptor);
+
+ Map<Long, String> partitionNameByIds =
+ waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath);
+ assertThat(partitionNameByIds).hasSizeGreaterThanOrEqualTo(2);
+ Iterator<Map.Entry<Long, String>> partitionIterator =
+ partitionNameByIds.entrySet().iterator();
+ Map.Entry<Long, String> firstPartition = partitionIterator.next();
+ Map.Entry<Long, String> secondPartition = partitionIterator.next();
+
+ appendRowsToPartition(tablePath, firstPartition.getValue(), 100);
+ appendRowsToPartition(tablePath, secondPartition.getValue(), 100);
+
+ Configuration sourceConf = new Configuration(clientConf);
+ sourceConf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 1);
+ sourceConf.setString(
+ ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES_FOR_BUCKET.key(), "1b");
+
+ MetricListener metricListener = new MetricListener();
+ TestingReaderContext readerContext =
+ new TestingReaderContext(
+ new org.apache.flink.configuration.Configuration(),
+ InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
+ FlinkSourceReaderMetrics sourceReaderMetrics =
+ new FlinkSourceReaderMetrics(readerContext.metricGroup());
+
+ try (TestingFlinkSourceReader<RowData> reader =
+ createReader(
+ sourceConf,
+ tablePath,
+ tableDescriptor.getSchema().getRowType(),
+ readerContext,
+ null,
+ sourceReaderMetrics)) {
+ assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS)).isEmpty();
+
+ reader.addSplits(createLogSplits(tableId, firstPartition));
+ Optional<Gauge<Long>> pendingRecords =
+ metricListener.getGauge(MetricNames.PENDING_RECORDS);
+ assertThat(pendingRecords).isPresent();
+ Gauge<Long> registeredPendingRecords = pendingRecords.get();
+ retry(
+ Duration.ofMinutes(1),
+ () -> assertThat((long) registeredPendingRecords.getValue()).isPositive());
+
+ TestingReaderOutput<RowData> output = new TestingReaderOutput<>();
+ removePartitionAndWaitForFetcherShutdown(reader, firstPartition, output);
+ assertThat((long) registeredPendingRecords.getValue()).isZero();
+
+ reader.addSplits(createLogSplits(tableId, secondPartition));
+ assertThat(reader.getNumAliveFetchers()).isOne();
+ assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS).get())
+ .isSameAs(registeredPendingRecords);
+ retry(
+ Duration.ofMinutes(1),
+ () -> assertThat((long) registeredPendingRecords.getValue()).isPositive());
+
+ // The recreated fetcher has a monotonically increasing ID. Removing its partition
+ // must target the currently running fetcher rather than assuming the original ID.
+ removePartitionAndWaitForFetcherShutdown(reader, secondPartition, output);
+ assertThat((long) registeredPendingRecords.getValue()).isZero();
+ }
+ }
+
private UpsertResult upsert(TablePath tablePath, InternalRow row) throws Exception {
try (Table table = conn.getTable(tablePath)) {
UpsertWriter writer = table.newUpsert().createWriter();
@@ -418,13 +493,68 @@ class FlinkSourceReaderTest extends FlinkTestBase {
}
}
- private FlinkSourceReader<RowData> createReader(
+ private void appendRowsToPartition(TablePath tablePath, String partitionName, int numRows)
+ throws Exception {
+ List<InternalRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ rows.add(row(i, partitionName));
+ }
+ writeRows(conn, tablePath, rows, true);
+ }
+
+ private List<SourceSplitBase> createLogSplits(long tableId, Map.Entry<Long, String> partition) {
+ List<SourceSplitBase> splits = new ArrayList<>(DEFAULT_BUCKET_NUM);
+ for (int bucket = 0; bucket < DEFAULT_BUCKET_NUM; bucket++) {
+ splits.add(
+ new LogSplit(
+ new TableBucket(tableId, partition.getKey(), bucket),
+ partition.getValue(),
+ 0L));
+ }
+ return splits;
+ }
+
+ private void removePartitionAndWaitForFetcherShutdown(
+ TestingFlinkSourceReader<RowData> reader,
+ Map.Entry<Long, String> partition,
+ TestingReaderOutput<RowData> output)
+ throws Exception {
+ reader.handleSourceEvents(
+ new PartitionsRemovedEvent(
+ Collections.singletonMap(partition.getKey(), partition.getValue())));
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ reader.pollNext(output);
+ assertThat(reader.getNumberOfCurrentlyAssignedSplits()).isZero();
+ assertThat(reader.getNumAliveFetchers()).isZero();
+ });
+ }
+
+ private TestingFlinkSourceReader<RowData> createReader(
Configuration flussConf,
TablePath tablePath,
RowType sourceOutputType,
SourceReaderContext context,
LakeSource<LakeSplit> lakeSource)
throws Exception {
+ return createReader(
+ flussConf,
+ tablePath,
+ sourceOutputType,
+ context,
+ lakeSource,
+ new FlinkSourceReaderMetrics(context.metricGroup()));
+ }
+
+ private TestingFlinkSourceReader<RowData> createReader(
+ Configuration flussConf,
+ TablePath tablePath,
+ RowType sourceOutputType,
+ SourceReaderContext context,
+ LakeSource<LakeSplit> lakeSource,
+ FlinkSourceReaderMetrics flinkSourceReaderMetrics)
+ throws Exception {
FutureCompletingBlockingQueue<RecordsWithSplitIds<RecordAndPos>> elementsQueue =
new FutureCompletingBlockingQueue<>();
@@ -436,19 +566,46 @@ class FlinkSourceReaderTest extends FlinkTestBase {
sourceOutputType));
FlinkRecordEmitter<RowData> recordEmitter = new FlinkRecordEmitter<>(deserializationSchema);
- return new FlinkSourceReader<>(
+ return new TestingFlinkSourceReader<>(
elementsQueue,
flussConf,
tablePath,
sourceOutputType,
context,
- null,
- null,
- new FlinkSourceReaderMetrics(context.metricGroup()),
+ flinkSourceReaderMetrics,
recordEmitter,
lakeSource);
}
+ private static final class TestingFlinkSourceReader<OUT> extends FlinkSourceReader<OUT> {
+
+ private TestingFlinkSourceReader(
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<RecordAndPos>> elementsQueue,
+ Configuration flussConfig,
+ TablePath tablePath,
+ RowType sourceOutputType,
+ SourceReaderContext context,
+ FlinkSourceReaderMetrics flinkSourceReaderMetrics,
+ FlinkRecordEmitter<OUT> recordEmitter,
+ LakeSource<LakeSplit> lakeSource) {
+ super(
+ elementsQueue,
+ flussConfig,
+ tablePath,
+ sourceOutputType,
+ context,
+ null,
+ null,
+ flinkSourceReaderMetrics,
+ recordEmitter,
+ lakeSource);
+ }
+
+ private int getNumAliveFetchers() {
+ return splitFetcherManager.getNumAliveFetchers();
+ }
+ }
+
private static final class TrackingLakeSource extends TestingLakeSource {
private final List<LogRecord> records;
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java
index 5c7cb313f2..139073a76f 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java
@@ -23,6 +23,8 @@ import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.client.table.writer.AppendWriter;
import org.apache.fluss.client.table.writer.UpsertWriter;
import org.apache.fluss.client.write.HashBucketAssigner;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
import org.apache.fluss.flink.source.metrics.FlinkSourceReaderMetrics;
import org.apache.fluss.flink.source.split.HybridSnapshotLogSplit;
@@ -43,12 +45,15 @@ import org.apache.fluss.types.RowType;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
+import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
import org.apache.flink.table.api.ValidationException;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -56,12 +61,14 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import static org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
import static org.apache.fluss.flink.source.testutils.RecordAndPosAssert.assertThatRecordAndPos;
import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -245,6 +252,133 @@ class FlinkSourceSplitReaderTest extends FlinkTestBase {
}
}
+ @Test
+ void testPendingRecordsMetric() throws Exception {
+ int recordsPerBucket = 5;
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .build();
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "test-pending-records-metric");
+ long tableId =
+ createTable(
+ tablePath,
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(DEFAULT_BUCKET_NUM)
+ .build());
+ // Keep each row in a separate log batch so each fetch leaves observable lag behind.
+ appendRowsInSeparateBatches(tablePath, recordsPerBucket * DEFAULT_BUCKET_NUM);
+
+ Configuration sourceConf = new Configuration(clientConf);
+ sourceConf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 1);
+ sourceConf.setString(
+ ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES_FOR_BUCKET.key(), "1b");
+ MetricListener metricListener = new MetricListener();
+ FlinkSourceReaderMetrics sourceReaderMetrics =
+ new FlinkSourceReaderMetrics(
+ InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup()));
+
+ FlinkSourceSplitReader splitReader =
+ new FlinkSourceSplitReader(
+ sourceConf,
+ tablePath,
+ schema.getRowType(),
+ null,
+ null,
+ null,
+ sourceReaderMetrics);
+ boolean splitReaderClosed = false;
+ try {
+ // the metric is registered when the split reader creates the log scanner, and reports
+ // 0 before anything is fetched
+ Optional<Gauge<Long>> pendingRecords =
+ metricListener.getGauge(MetricNames.PENDING_RECORDS);
+ assertThat(pendingRecords).isPresent();
+ assertThat((long) pendingRecords.get().getValue()).isEqualTo(0L);
+
+ LogSplit firstStreamingSplit = new LogSplit(new TableBucket(tableId, 0), null, 0L);
+ LogSplit secondStreamingSplit = new LogSplit(new TableBucket(tableId, 1), null, 0L);
+ LogSplit boundedSplit = new LogSplit(new TableBucket(tableId, 2), null, 0L, 1L);
+ List<SourceSplitBase> logSplits =
+ Arrays.asList(firstStreamingSplit, secondStreamingSplit, boundedSplit);
+ splitReader.handleSplitsChanges(new SplitsAddition<>(logSplits));
+ // Assigned buckets have no known high watermark until their first fetch completes.
+ assertThat((long) pendingRecords.get().getValue()).isEqualTo(0L);
+
+ Map<String, Integer> fetchedRowsBySplit = new HashMap<>();
+ Set<String> finishedSplits = new HashSet<>();
+ boolean aggregationVerified = false;
+ while (fetchedRowsBySplit.getOrDefault(firstStreamingSplit.splitId(), 0)
+ < recordsPerBucket
+ || fetchedRowsBySplit.getOrDefault(secondStreamingSplit.splitId(), 0)
+ < recordsPerBucket
+ || !finishedSplits.contains(boundedSplit.splitId())) {
+ RecordsWithSplitIds<RecordAndPos> records = splitReader.fetch();
+ int rowsInFetch = 0;
+ String splitId = records.nextSplit();
+ while (splitId != null) {
+ while (records.nextRecordFromSplit() != null) {
+ rowsInFetch++;
+ fetchedRowsBySplit.put(
+ splitId, fetchedRowsBySplit.getOrDefault(splitId, 0) + 1);
+ }
+ splitId = records.nextSplit();
+ }
+ finishedSplits.addAll(records.finishedSplits());
+ records.recycle();
+
+ if (rowsInFetch > 0) {
+ int firstStreamingFetched =
+ fetchedRowsBySplit.getOrDefault(firstStreamingSplit.splitId(), 0);
+ int secondStreamingFetched =
+ fetchedRowsBySplit.getOrDefault(secondStreamingSplit.splitId(), 0);
+
+ if (firstStreamingFetched > 0 && secondStreamingFetched > 0) {
+ assertThat(firstStreamingFetched).isLessThanOrEqualTo(recordsPerBucket);
+ assertThat(secondStreamingFetched).isLessThanOrEqualTo(recordsPerBucket);
+ assertThat((long) pendingRecords.get().getValue())
+ .isEqualTo(
+ 2L * recordsPerBucket
+ - firstStreamingFetched
+ - secondStreamingFetched);
+ if (firstStreamingFetched < recordsPerBucket
+ && secondStreamingFetched < recordsPerBucket) {
+ aggregationVerified = true;
+ }
+ }
+ }
+ }
+
+ assertThat(aggregationVerified).isTrue();
+ assertThat(fetchedRowsBySplit.get(boundedSplit.splitId())).isEqualTo(1);
+ assertThat(recordsPerBucket - fetchedRowsBySplit.get(boundedSplit.splitId()))
+ .isPositive();
+ // The bounded bucket was unassigned with records still unread, so only the two fully
+ // consumed streaming buckets contribute to the final value.
+ assertThat((long) pendingRecords.get().getValue()).isEqualTo(0L);
+
+ // Closing a scanner with unread records must remove its gauge from the aggregate.
+ appendRowsInSeparateBatches(tablePath, recordsPerBucket * DEFAULT_BUCKET_NUM);
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ RecordsWithSplitIds<RecordAndPos> records = splitReader.fetch();
+ records.recycle();
+ assertThat((long) pendingRecords.get().getValue()).isPositive();
+ });
+
+ splitReaderClosed = true;
+ splitReader.close();
+ assertThat((long) pendingRecords.get().getValue()).isZero();
+ } finally {
+ if (!splitReaderClosed) {
+ splitReader.close();
+ }
+ }
+ }
+
@Test
void testHandleMixSnapshotLogSplitChangesAndFetch() throws Exception {
TablePath tablePath = TablePath.of(DEFAULT_DB, "test-mix-snapshot-log-table");
@@ -531,6 +665,16 @@ class FlinkSourceSplitReaderTest extends FlinkTestBase {
return internalRows;
}
+ private void appendRowsInSeparateBatches(TablePath tablePath, int rows) throws Exception {
+ try (Table table = conn.getTable(tablePath)) {
+ AppendWriter appendWriter = table.newAppend().createWriter();
+ for (int i = 0; i < rows; i++) {
+ appendWriter.append(row(i, "v" + i));
+ appendWriter.flush();
+ }
+ }
+ }
+
private static String toLogSplitId(TableBucket tableBucket) {
return new LogSplit(tableBucket, null, 0L).splitId();
}
diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md
index 4d69d66484..9d8a0031cf 100644
--- a/website/docs/maintenance/observability/monitor-metrics.md
+++ b/website/docs/maintenance/observability/monitor-metrics.md
@@ -1258,6 +1258,12 @@ How to Use Flink Metrics, you can see [Flink Metrics](https://nightlies.apache.o
<td>Time difference between reading the data file and file creation.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>pendingRecords</td>
+ <td>Flink Source Operator</td>
+ <td>The number of log records that are available after the current source fetch offset. Only the streaming log part is counted, snapshot and lake records are excluded.</td>
+ <td>Gauge</td>
+ </tr>
</tbody>
</table>