From cb9efa71dab0406577aa194408a24ef8881d646f Mon Sep 17 00:00:00 2001 From: Hongshun Wang Date: Tue, 8 Sep 2026 11:43:45 +0800 Subject: [PATCH 1/2] [flink] Support pendingRecords metric for Fluss source. Track log record lag using scanner high watermarks and current fetch offsets, and aggregate the lag across subscribed buckets. --- .../client/metrics/ScannerMetricGroup.java | 9 +++ .../table/scanner/log/AbstractLogScanner.java | 1 + .../table/scanner/log/BucketScanStatus.java | 16 ++++- .../table/scanner/log/LogScannerStatus.java | 8 +++ .../scanner/log/LogFetchCollectorTest.java | 23 +++++++ .../org/apache/fluss/metrics/MetricNames.java | 1 + .../metrics/FlinkSourceReaderMetrics.java | 15 +++++ .../source/reader/FlinkSourceSplitReader.java | 15 ++++- .../metrics/FlinkSourceReaderMetricsTest.java | 21 ++++++ .../reader/FlinkSourceSplitReaderTest.java | 67 +++++++++++++++++++ .../observability/monitor-metrics.md | 6 ++ 11 files changed, 180 insertions(+), 2 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 bae44d00ed5..279db462c2c 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.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 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 8e16b35ef75..60ac2acbb11 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 @@ protected AbstractLogScanner( 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 738e04a49e3..c99ae783875 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 @@ @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); + } } 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 c3ea600cda0..8ae1391a9a2 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 @@ 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 scanBucketAndOffsets) { for (Map.Entry 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 aafad8c0eeb..e00c07e8970 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,14 +236,37 @@ 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(); } + @Test + void testRecordsLagAggregation() { + TableBucket initializedBucket = new TableBucket(DATA1_TABLE_ID, 0); + TableBucket uninitializedBucket = new TableBucket(DATA1_TABLE_ID, 1); + Map scanBuckets = new HashMap<>(); + scanBuckets.put(initializedBucket, 2L); + scanBuckets.put(uninitializedBucket, LogScanner.EARLIEST_OFFSET); + + LogScannerStatus scannerStatus = new LogScannerStatus(); + scannerStatus.assignScanBuckets(scanBuckets); + scannerStatus.updateHighWatermark(initializedBucket, 10L); + assertThat(scannerStatus.recordsLag()).isEqualTo(8L); + + scannerStatus.updateHighWatermark(uninitializedBucket, 7L); + scannerStatus.updateOffset(uninitializedBucket, 3L); + assertThat(scannerStatus.recordsLag()).isEqualTo(12L); + + scannerStatus.unassignScanBuckets(Collections.singletonList(initializedBucket)); + assertThat(scannerStatus.recordsLag()).isEqualTo(4L); + } + @Test void testFilteredEmptyResponseAdvancesOffset() { Configuration conf = new Configuration(); 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 644b11fd7e7..766e0e9abc3 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 400f1969fa1..3d86c3eafb9 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 @@ -19,6 +19,7 @@ import org.apache.fluss.flink.source.reader.FlinkSourceReader; +import org.apache.flink.metrics.Gauge; import org.apache.flink.metrics.groups.SourceReaderMetricGroup; import org.apache.flink.runtime.metrics.MetricNames; @@ -33,6 +34,8 @@ public class FlinkSourceReaderMetrics { // For currentFetchEventTimeLag metric private volatile long currentFetchEventTimeLag = UNINITIALIZED; + private volatile boolean pendingRecordsGaugeRegistered; + public FlinkSourceReaderMetrics(SourceReaderMetricGroup sourceReaderMetricGroup) { this.sourceReaderMetricGroup = sourceReaderMetricGroup; } @@ -50,6 +53,18 @@ public void reportRecordEventTime(long lag) { currentFetchEventTimeLag = lag; } + /** + * Registers the gauge reporting the number of log records that have not been fetched yet, which + * backs the standard Flink pendingRecords metric. Only the first registration takes effect, so + * that re-created split readers won't register the metric again. + */ + public void registerPendingRecordsGauge(Gauge pendingRecordsGauge) { + if (!pendingRecordsGaugeRegistered) { + sourceReaderMetricGroup.setPendingRecordsGauge(pendingRecordsGauge); + pendingRecordsGaugeRegistered = true; + } + } + public SourceReaderMetricGroup getSourceReaderMetricGroup() { return sourceReaderMetricGroup; } 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 1fff4292175..e1f91692038 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.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; @@ -127,7 +129,9 @@ public FlinkSourceSplitReader( @Nullable LakeSource 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 +150,15 @@ public FlinkSourceSplitReader( this.stoppingOffsets = new HashMap<>(); this.emptyLogSplits = new HashSet<>(); this.lakeSource = lakeSource; + + @SuppressWarnings("unchecked") + Gauge recordsLagGauge = + (Gauge) + checkNotNull( + flinkMetricRegistry.getFlussMetric(MetricNames.SCANNER_RECORDS_LAG), + "The gauge %s should have been registered by the log scanner.", + MetricNames.SCANNER_RECORDS_LAG); + flinkSourceReaderMetrics.registerPendingRecordsGauge(recordsLagGauge::getValue); } @Override 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 f7bb7f2fcf9..78121dbd107 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.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,24 @@ void testCurrentFetchEventTimeLag() { flinkSourceReaderMetrics.reportRecordEventTime(18213L); assertThat((long) currentFetchEventTimeLag.get().getValue()).isEqualTo(18213L); } + + @Test + void testPendingRecords() { + 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 recordsLag = new AtomicLong(10L); + flinkSourceReaderMetrics.registerPendingRecordsGauge(recordsLag::get); + Optional> pendingRecords = metricListener.getGauge(MetricNames.PENDING_RECORDS); + assertThat(pendingRecords).isPresent(); + assertThat((long) pendingRecords.get().getValue()).isEqualTo(10L); + + recordsLag.set(3L); + assertThat((long) pendingRecords.get().getValue()).isEqualTo(3L); + } } 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 5c7cb313f28..394f2c27a96 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.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,7 +45,9 @@ 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; @@ -56,6 +60,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.OptionalLong; import java.util.Set; @@ -245,6 +250,68 @@ void testHandleLogSplitChangesAndFetch() throws Exception { } } + @Test + void testPendingRecordsMetric() throws Exception { + 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(1).build()); + appendRows(tablePath, 5); + + Configuration sourceConf = new Configuration(clientConf); + sourceConf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 1); + MetricListener metricListener = new MetricListener(); + FlinkSourceReaderMetrics sourceReaderMetrics = + new FlinkSourceReaderMetrics( + InternalSourceReaderMetricGroup.mock(metricListener.getMetricGroup())); + + try (FlinkSourceSplitReader splitReader = + new FlinkSourceSplitReader( + sourceConf, + tablePath, + schema.getRowType(), + null, + null, + null, + sourceReaderMetrics)) { + // the metric is registered when the split reader creates the log scanner, and reports + // 0 before anything is fetched + Optional> pendingRecords = + metricListener.getGauge(MetricNames.PENDING_RECORDS); + assertThat(pendingRecords).isPresent(); + assertThat((long) pendingRecords.get().getValue()).isEqualTo(0L); + + TableBucket tableBucket = new TableBucket(tableId, 0); + LogSplit logSplit = new LogSplit(tableBucket, null, 0L); + splitReader.handleSplitsChanges( + new SplitsAddition<>(Collections.singletonList(logSplit))); + + // fetch the rows one by one, the lag should decrease accordingly. Note that a fetch + // may return no records when the poll times out before any record arrives. + int fetchedRows = 0; + while (fetchedRows < 5) { + RecordsWithSplitIds records = splitReader.fetch(); + int rowsInFetch = 0; + if (records.nextSplit() != null) { + while (records.nextRecordFromSplit() != null) { + rowsInFetch++; + } + } + records.recycle(); + if (rowsInFetch > 0) { + fetchedRows += rowsInFetch; + assertThat((long) pendingRecords.get().getValue()).isEqualTo(5 - fetchedRows); + } + } + } + } + @Test void testHandleMixSnapshotLogSplitChangesAndFetch() throws Exception { TablePath tablePath = TablePath.of(DEFAULT_DB, "test-mix-snapshot-log-table"); diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 4d69d66484c..9d8a0031cf8 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 Time difference between reading the data file and file creation. Gauge + + pendingRecords + Flink Source Operator + 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. + Gauge + From e7401793845fc97974bdf271918825cf89e35084 Mon Sep 17 00:00:00 2001 From: Hongshun Wang Date: Wed, 9 Sep 2026 14:34:20 +0800 Subject: [PATCH 2/2] modified based on CR --- .../scanner/log/LogFetchCollectorTest.java | 21 --- .../metrics/FlinkSourceReaderMetrics.java | 31 +++- .../source/reader/FlinkSourceSplitReader.java | 7 +- .../fetcher/FlinkSourceFetcherManager.java | 2 +- .../metrics/FlinkSourceReaderMetricsTest.java | 25 ++- .../source/reader/FlinkSourceReaderTest.java | 167 +++++++++++++++++- .../reader/FlinkSourceSplitReaderTest.java | 109 ++++++++++-- 7 files changed, 305 insertions(+), 57 deletions(-) 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 e00c07e8970..064512a568a 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 @@ -246,27 +246,6 @@ void testShouldContinueConsumeSameCompletedFetchAcrossPolls() throws Exception { assertThat(completedFetch.isConsumed()).isFalse(); } - @Test - void testRecordsLagAggregation() { - TableBucket initializedBucket = new TableBucket(DATA1_TABLE_ID, 0); - TableBucket uninitializedBucket = new TableBucket(DATA1_TABLE_ID, 1); - Map scanBuckets = new HashMap<>(); - scanBuckets.put(initializedBucket, 2L); - scanBuckets.put(uninitializedBucket, LogScanner.EARLIEST_OFFSET); - - LogScannerStatus scannerStatus = new LogScannerStatus(); - scannerStatus.assignScanBuckets(scanBuckets); - scannerStatus.updateHighWatermark(initializedBucket, 10L); - assertThat(scannerStatus.recordsLag()).isEqualTo(8L); - - scannerStatus.updateHighWatermark(uninitializedBucket, 7L); - scannerStatus.updateOffset(uninitializedBucket, 3L); - assertThat(scannerStatus.recordsLag()).isEqualTo(12L); - - scannerStatus.unassignScanBuckets(Collections.singletonList(initializedBucket)); - assertThat(scannerStatus.recordsLag()).isEqualTo(4L); - } - @Test void testFilteredEmptyResponseAdvancesOffset() { Configuration conf = new Configuration(); 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 3d86c3eafb9..124635635e0 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,11 +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.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 { @@ -30,11 +33,12 @@ public class FlinkSourceReaderMetrics { // Source reader metric group private final SourceReaderMetricGroup sourceReaderMetricGroup; + private final Set> recordsLagMetrics = ConcurrentHashMap.newKeySet(); // For currentFetchEventTimeLag metric private volatile long currentFetchEventTimeLag = UNINITIALIZED; - private volatile boolean pendingRecordsGaugeRegistered; + private boolean pendingRecordsGaugeRegistered; public FlinkSourceReaderMetrics(SourceReaderMetricGroup sourceReaderMetricGroup) { this.sourceReaderMetricGroup = sourceReaderMetricGroup; @@ -54,18 +58,31 @@ public void reportRecordEventTime(long lag) { } /** - * Registers the gauge reporting the number of log records that have not been fetched yet, which - * backs the standard Flink pendingRecords metric. Only the first registration takes effect, so - * that re-created split readers won't register the metric again. + * 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 void registerPendingRecordsGauge(Gauge pendingRecordsGauge) { + public synchronized void maybeAddRecordsLagMetric(Gauge recordsLagMetric) { + recordsLagMetrics.add(recordsLagMetric); if (!pendingRecordsGaugeRegistered) { - sourceReaderMetricGroup.setPendingRecordsGauge(pendingRecordsGauge); + sourceReaderMetricGroup.setPendingRecordsGauge(this::getPendingRecords); pendingRecordsGaugeRegistered = true; } } + /** Removes a scanner's records-lag metric from the tracked metrics. */ + public void removeRecordsLagMetric(Gauge recordsLagMetric) { + recordsLagMetrics.remove(recordsLagMetric); + } + public SourceReaderMetricGroup getSourceReaderMetricGroup() { return sourceReaderMetricGroup; } + + private long getPendingRecords() { + long pendingRecords = 0L; + for (Gauge 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 e1f91692038..692775ede48 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 @@ -97,6 +97,7 @@ public class FlinkSourceSplitReader implements SplitReader recordsLagMetric; @Nullable private BoundedSplitReader currentBoundedSplitReader; @Nullable private SourceSplitBase currentBoundedSplit; @@ -152,13 +153,14 @@ public FlinkSourceSplitReader( this.lakeSource = lakeSource; @SuppressWarnings("unchecked") - Gauge recordsLagGauge = + Gauge recordsLagMetric = (Gauge) checkNotNull( flinkMetricRegistry.getFlussMetric(MetricNames.SCANNER_RECORDS_LAG), "The gauge %s should have been registered by the log scanner.", MetricNames.SCANNER_RECORDS_LAG); - flinkSourceReaderMetrics.registerPendingRecordsGauge(recordsLagGauge::getValue); + this.recordsLagMetric = recordsLagMetric; + flinkSourceReaderMetrics.maybeAddRecordsLagMetric(recordsLagMetric); } @Override @@ -643,6 +645,7 @@ public void wakeUp() { @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 9b1c9ff0fdf..fdbe0fd08e3 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 FlinkSourceFetcherManager( public void removePartitions( Map removedPartitions, Consumer> unsubscribeTableBucketsCallback) { - SplitFetcher splitFetcher = fetchers.get(0); + SplitFetcher 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 78121dbd107..b40f5f0dba0 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 @@ -52,7 +52,7 @@ void testCurrentFetchEventTimeLag() { } @Test - void testPendingRecords() { + void testPendingRecordsGaugeTracksActiveScanners() { MetricListener metricListener = new MetricListener(); FlinkSourceReaderMetrics flinkSourceReaderMetrics = new FlinkSourceReaderMetrics( @@ -61,13 +61,28 @@ void testPendingRecords() { // the metric is not registered until a log scanner provides its records lag assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS)).isEmpty(); - AtomicLong recordsLag = new AtomicLong(10L); - flinkSourceReaderMetrics.registerPendingRecordsGauge(recordsLag::get); + AtomicLong firstRecordsLag = new AtomicLong(10L); + org.apache.fluss.metrics.Gauge firstRecordsLagMetric = firstRecordsLag::get; + flinkSourceReaderMetrics.maybeAddRecordsLagMetric(firstRecordsLagMetric); Optional> pendingRecords = metricListener.getGauge(MetricNames.PENDING_RECORDS); assertThat(pendingRecords).isPresent(); assertThat((long) pendingRecords.get().getValue()).isEqualTo(10L); - recordsLag.set(3L); - assertThat((long) pendingRecords.get().getValue()).isEqualTo(3L); + AtomicLong secondRecordsLag = new AtomicLong(7L); + org.apache.fluss.metrics.Gauge 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 91b685deacd..7ae9a32c7b0 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.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.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.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 @@ void testHandlePartitionsRemovedEvent() throws Exception { } } + @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 partitionNameByIds = + waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath); + assertThat(partitionNameByIds).hasSizeGreaterThanOrEqualTo(2); + Iterator> partitionIterator = + partitionNameByIds.entrySet().iterator(); + Map.Entry firstPartition = partitionIterator.next(); + Map.Entry 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 reader = + createReader( + sourceConf, + tablePath, + tableDescriptor.getSchema().getRowType(), + readerContext, + null, + sourceReaderMetrics)) { + assertThat(metricListener.getGauge(MetricNames.PENDING_RECORDS)).isEmpty(); + + reader.addSplits(createLogSplits(tableId, firstPartition)); + Optional> pendingRecords = + metricListener.getGauge(MetricNames.PENDING_RECORDS); + assertThat(pendingRecords).isPresent(); + Gauge registeredPendingRecords = pendingRecords.get(); + retry( + Duration.ofMinutes(1), + () -> assertThat((long) registeredPendingRecords.getValue()).isPositive()); + + TestingReaderOutput 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 @@ private UpsertResult upsert(TablePath tablePath, InternalRow row) throws Excepti } } - private FlinkSourceReader createReader( + private void appendRowsToPartition(TablePath tablePath, String partitionName, int numRows) + throws Exception { + List rows = new ArrayList<>(numRows); + for (int i = 0; i < numRows; i++) { + rows.add(row(i, partitionName)); + } + writeRows(conn, tablePath, rows, true); + } + + private List createLogSplits(long tableId, Map.Entry partition) { + List 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 reader, + Map.Entry partition, + TestingReaderOutput 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 createReader( Configuration flussConf, TablePath tablePath, RowType sourceOutputType, SourceReaderContext context, LakeSource lakeSource) throws Exception { + return createReader( + flussConf, + tablePath, + sourceOutputType, + context, + lakeSource, + new FlinkSourceReaderMetrics(context.metricGroup())); + } + + private TestingFlinkSourceReader createReader( + Configuration flussConf, + TablePath tablePath, + RowType sourceOutputType, + SourceReaderContext context, + LakeSource lakeSource, + FlinkSourceReaderMetrics flinkSourceReaderMetrics) + throws Exception { FutureCompletingBlockingQueue> elementsQueue = new FutureCompletingBlockingQueue<>(); @@ -436,19 +566,46 @@ private FlinkSourceReader createReader( sourceOutputType)); FlinkRecordEmitter 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 extends FlinkSourceReader { + + private TestingFlinkSourceReader( + FutureCompletingBlockingQueue> elementsQueue, + Configuration flussConfig, + TablePath tablePath, + RowType sourceOutputType, + SourceReaderContext context, + FlinkSourceReaderMetrics flinkSourceReaderMetrics, + FlinkRecordEmitter recordEmitter, + LakeSource 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 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 394f2c27a96..139073a76f1 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 @@ -53,6 +53,7 @@ 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; @@ -67,6 +68,7 @@ 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; @@ -252,6 +254,7 @@ void testHandleLogSplitChangesAndFetch() throws Exception { @Test void testPendingRecordsMetric() throws Exception { + int recordsPerBucket = 5; Schema schema = Schema.newBuilder() .column("id", DataTypes.INT()) @@ -261,17 +264,23 @@ void testPendingRecordsMetric() throws Exception { long tableId = createTable( tablePath, - TableDescriptor.builder().schema(schema).distributedBy(1).build()); - appendRows(tablePath, 5); + 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())); - try (FlinkSourceSplitReader splitReader = + FlinkSourceSplitReader splitReader = new FlinkSourceSplitReader( sourceConf, tablePath, @@ -279,7 +288,9 @@ void testPendingRecordsMetric() throws Exception { null, null, null, - sourceReaderMetrics)) { + 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> pendingRecords = @@ -287,28 +298,84 @@ void testPendingRecordsMetric() throws Exception { assertThat(pendingRecords).isPresent(); assertThat((long) pendingRecords.get().getValue()).isEqualTo(0L); - TableBucket tableBucket = new TableBucket(tableId, 0); - LogSplit logSplit = new LogSplit(tableBucket, null, 0L); - splitReader.handleSplitsChanges( - new SplitsAddition<>(Collections.singletonList(logSplit))); - - // fetch the rows one by one, the lag should decrease accordingly. Note that a fetch - // may return no records when the poll times out before any record arrives. - int fetchedRows = 0; - while (fetchedRows < 5) { + 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 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 fetchedRowsBySplit = new HashMap<>(); + Set finishedSplits = new HashSet<>(); + boolean aggregationVerified = false; + while (fetchedRowsBySplit.getOrDefault(firstStreamingSplit.splitId(), 0) + < recordsPerBucket + || fetchedRowsBySplit.getOrDefault(secondStreamingSplit.splitId(), 0) + < recordsPerBucket + || !finishedSplits.contains(boundedSplit.splitId())) { RecordsWithSplitIds records = splitReader.fetch(); int rowsInFetch = 0; - if (records.nextSplit() != null) { + 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) { - fetchedRows += rowsInFetch; - assertThat((long) pendingRecords.get().getValue()).isEqualTo(5 - fetchedRows); + 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 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(); + } } } @@ -598,6 +665,16 @@ private List appendRows(TablePath tablePath, int rows) throws Excep 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(); }