You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/04/12 04:48:46 UTC
(fluss) branch main updated: [client] Fix LogFetchCollector stale fetch detection across poll boundaries (#3053)
This is an automated email from the ASF dual-hosted git repository.
jark 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 9e4724d75 [client] Fix LogFetchCollector stale fetch detection across poll boundaries (#3053)
9e4724d75 is described below
commit 9e4724d75fe381f7dc4a96357acc79fab5f6121c
Author: Yang Wang <oc...@alibaba-inc.com>
AuthorDate: Sun Apr 12 12:48:43 2026 +0800
[client] Fix LogFetchCollector stale fetch detection across poll boundaries (#3053)
---
.../table/scanner/log/LogFetchCollector.java | 7 +-
.../table/scanner/log/LogFetchCollectorTest.java | 49 ++++++++
.../table/scanner/log/LogFetcherFilterITCase.java | 137 ++++++++++++++++++++-
3 files changed, 187 insertions(+), 6 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
index c49ab1ba9..5961de0ec 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
@@ -156,10 +156,7 @@ public class LogFetchCollector {
tb,
nextInLineFetch.fetchOffset());
} else {
- // When server-side filtering is enabled, the CompletedFetch's nextFetchOffset
- // may have been advanced by filteredEndOffset (all batches filtered out).
- // The fetchOffset still matches the original request offset.
- if (nextInLineFetch.fetchOffset() == offset) {
+ if (nextInLineFetch.nextFetchOffset() == offset) {
List<ScanRecord> records = nextInLineFetch.fetchRecords(maxRecords);
LOG.trace(
"Returning {} fetched records at offset {} for assigned bucket {}.",
@@ -183,7 +180,7 @@ public class LogFetchCollector {
LOG.warn(
"Ignoring fetched records for {} at offset {} since the current offset is {}",
nextInLineFetch.tableBucket,
- nextInLineFetch.fetchOffset(),
+ nextInLineFetch.nextFetchOffset(),
offset);
}
}
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 2ce40ce55..be28b31bc 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
@@ -20,6 +20,7 @@ package org.apache.fluss.client.table.scanner.log;
import org.apache.fluss.client.metadata.MetadataUpdater;
import org.apache.fluss.client.metadata.TestingMetadataUpdater;
import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.record.LogRecordBatch;
@@ -205,6 +206,54 @@ public class LogFetchCollectorTest {
assertThat(totalBytesRead).isEqualTo(expectedTotal);
}
+ @Test
+ void testShouldContinueConsumeSameCompletedFetchAcrossPolls() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 2);
+ MetadataUpdater metadataUpdater =
+ new TestingMetadataUpdater(
+ Collections.singletonMap(DATA1_TABLE_PATH, DATA1_TABLE_INFO));
+ LogFetchCollector collector =
+ new LogFetchCollector(DATA1_TABLE_PATH, logScannerStatus, conf, metadataUpdater);
+
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ FetchLogResultForBucket result =
+ new FetchLogResultForBucket(tb, genMemoryLogRecordsByObject(DATA1), 10L);
+ CompletedFetch completedFetch = makeCompletedFetch(tb, result, 0L);
+ logFetchBuffer.add(completedFetch);
+
+ ScanRecords firstPoll = collector.collectFetch(logFetchBuffer);
+ assertThat(firstPoll.records(tb).size()).isEqualTo(2);
+ assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(2L);
+ assertThat(completedFetch.isConsumed()).isFalse();
+
+ ScanRecords secondPoll = collector.collectFetch(logFetchBuffer);
+ assertThat(secondPoll.records(tb).size()).isEqualTo(2);
+ assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(4L);
+ assertThat(completedFetch.isConsumed()).isFalse();
+ }
+
+ @Test
+ void testFilteredEmptyResponseAdvancesOffset() {
+ Configuration conf = new Configuration();
+ conf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 2);
+ MetadataUpdater metadataUpdater =
+ new TestingMetadataUpdater(
+ Collections.singletonMap(DATA1_TABLE_PATH, DATA1_TABLE_INFO));
+ LogFetchCollector collector =
+ new LogFetchCollector(DATA1_TABLE_PATH, logScannerStatus, conf, metadataUpdater);
+
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1);
+ FetchLogResultForBucket filteredEmpty = new FetchLogResultForBucket(tb, 10L, 20L);
+ CompletedFetch completedFetch = makeCompletedFetch(tb, filteredEmpty, 0L);
+ logFetchBuffer.add(completedFetch);
+
+ ScanRecords scanRecords = collector.collectFetch(logFetchBuffer);
+ assertThat(scanRecords.records(tb)).isEmpty();
+ assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(20L);
+ assertThat(completedFetch.isConsumed()).isTrue();
+ }
+
private DefaultCompletedFetch makeCompletedFetch(
TableBucket tableBucket, FetchLogResultForBucket resultForBucket, long offset) {
return new DefaultCompletedFetch(
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherFilterITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherFilterITCase.java
index 915bfa033..ac64fbdbd 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherFilterITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherFilterITCase.java
@@ -22,6 +22,7 @@ import org.apache.fluss.client.metadata.MetadataUpdater;
import org.apache.fluss.client.metrics.TestingScannerMetricGroup;
import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.predicate.Predicate;
import org.apache.fluss.predicate.PredicateBuilder;
@@ -57,6 +58,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class LogFetcherFilterITCase extends ClientToServerITCaseBase {
private LogFetcher logFetcher;
private long tableId;
+ private MetadataUpdater metadataUpdater;
private static final int BUCKET_ID_0 = 0;
private static final int BUCKET_ID_1 = 1;
@@ -90,7 +92,7 @@ public class LogFetcherFilterITCase extends ClientToServerITCaseBase {
FLUSS_CLUSTER_EXTENSION.waitUntilTableReady(tableId);
RpcClient rpcClient = FLUSS_CLUSTER_EXTENSION.getRpcClient();
- MetadataUpdater metadataUpdater = new MetadataUpdater(clientConf, rpcClient);
+ metadataUpdater = new MetadataUpdater(clientConf, rpcClient);
metadataUpdater.checkAndUpdateTableMetadata(Collections.singleton(DATA1_TABLE_PATH));
Map<TableBucket, Long> scanBuckets = new HashMap<>();
@@ -307,6 +309,139 @@ public class LogFetcherFilterITCase extends ClientToServerITCaseBase {
assertThat(logFetcher.getCompletedFetchesSize()).isEqualTo(0);
}
+ @Test
+ void testFilterFetchKeepsConsumingSameCompletedFetchAcrossPolls() throws Exception {
+ // Recreate fetcher with tiny max poll records so one CompletedFetch must span polls.
+ clientConf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 2);
+ logFetcher.close();
+ logFetcher =
+ createFetcherWithBuckets(
+ Collections.singletonMap(new TableBucket(tableId, BUCKET_ID_0), 0L));
+
+ TableBucket tb0 = new TableBucket(tableId, BUCKET_ID_0);
+ List<Object[]> manyMatchingRows =
+ Arrays.asList(
+ new Object[] {6, "u0"},
+ new Object[] {7, "u1"},
+ new Object[] {8, "u2"},
+ new Object[] {9, "u3"},
+ new Object[] {10, "u4"},
+ new Object[] {11, "u5"});
+ addRecordsToBucket(
+ tb0,
+ LogRecordBatchStatisticsTestUtils.createLogRecordsWithStatistics(
+ manyMatchingRows, FILTER_TEST_ROW_TYPE, 0L, DEFAULT_SCHEMA_ID));
+
+ logFetcher.sendFetches();
+ retry(Duration.ofMinutes(1), () -> assertThat(logFetcher.hasAvailableFetches()).isTrue());
+
+ ScanRecords firstPoll = logFetcher.collectFetch();
+ List<ScanRecord> firstRecords = firstPoll.records(tb0);
+ assertThat(firstRecords).hasSize(2);
+ assertThat(
+ firstRecords.stream()
+ .map(r -> r.getRow().getInt(0))
+ .collect(Collectors.toList()))
+ .containsExactly(6, 7);
+ assertThat(logFetcher.getLogScannerStatus().getBucketOffset(tb0)).isEqualTo(2L);
+
+ // This second poll must continue consuming the same in-buffer fetch instead of
+ // being treated as stale and drained.
+ ScanRecords secondPoll = logFetcher.collectFetch();
+ List<ScanRecord> secondRecords = secondPoll.records(tb0);
+ assertThat(secondRecords).hasSize(2);
+ assertThat(
+ secondRecords.stream()
+ .map(r -> r.getRow().getInt(0))
+ .collect(Collectors.toList()))
+ .containsExactly(8, 9);
+ assertThat(logFetcher.getLogScannerStatus().getBucketOffset(tb0)).isEqualTo(4L);
+ }
+
+ @Test
+ void testFilterFetchWithMultipleBucketsKeepsOffsetProgressForSameBucket() throws Exception {
+ // Keep poll size small so one completed fetch spans multiple collectFetch calls.
+ clientConf.setInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 2);
+ logFetcher.close();
+ Map<TableBucket, Long> scanBuckets = new HashMap<>();
+ TableBucket tb0 = new TableBucket(tableId, BUCKET_ID_0);
+ TableBucket tb1 = new TableBucket(tableId, BUCKET_ID_1);
+ scanBuckets.put(tb0, 0L);
+ scanBuckets.put(tb1, 0L);
+ logFetcher = createFetcherWithBuckets(scanBuckets);
+
+ List<Object[]> bucket0Rows =
+ Arrays.asList(
+ new Object[] {6, "b0_0"},
+ new Object[] {7, "b0_1"},
+ new Object[] {8, "b0_2"},
+ new Object[] {9, "b0_3"},
+ new Object[] {10, "b0_4"},
+ new Object[] {11, "b0_5"});
+ List<Object[]> bucket1Rows =
+ Arrays.asList(
+ new Object[] {12, "b1_0"},
+ new Object[] {13, "b1_1"},
+ new Object[] {14, "b1_2"},
+ new Object[] {15, "b1_3"},
+ new Object[] {16, "b1_4"},
+ new Object[] {17, "b1_5"});
+ addRecordsToBucket(
+ tb0,
+ LogRecordBatchStatisticsTestUtils.createLogRecordsWithStatistics(
+ bucket0Rows, FILTER_TEST_ROW_TYPE, 0L, DEFAULT_SCHEMA_ID));
+ addRecordsToBucket(
+ tb1,
+ LogRecordBatchStatisticsTestUtils.createLogRecordsWithStatistics(
+ bucket1Rows, FILTER_TEST_ROW_TYPE, 0L, DEFAULT_SCHEMA_ID));
+
+ logFetcher.sendFetches();
+ retry(Duration.ofMinutes(1), () -> assertThat(logFetcher.hasAvailableFetches()).isTrue());
+
+ ScanRecords firstPoll = logFetcher.collectFetch();
+ List<ScanRecord> firstTb0 = firstPoll.records(tb0);
+ List<ScanRecord> firstTb1 = firstPoll.records(tb1);
+ assertThat(firstTb0.size() + firstTb1.size()).isEqualTo(2);
+ TableBucket firstActiveBucket = firstTb0.isEmpty() ? tb1 : tb0;
+ TableBucket firstInactiveBucket = firstTb0.isEmpty() ? tb0 : tb1;
+ assertThat(firstPoll.records(firstActiveBucket)).hasSize(2);
+ assertThat(firstPoll.records(firstInactiveBucket)).isEmpty();
+ assertThat(logFetcher.getLogScannerStatus().getBucketOffset(firstActiveBucket))
+ .isEqualTo(2L);
+ assertThat(logFetcher.getLogScannerStatus().getBucketOffset(firstInactiveBucket))
+ .isEqualTo(0L);
+
+ // Core assertion: second poll should continue from the same bucket (tb0),
+ // not drain it as stale and jump to tb1.
+ ScanRecords secondPoll = logFetcher.collectFetch();
+ assertThat(secondPoll.records(firstActiveBucket)).hasSize(2);
+ assertThat(secondPoll.records(firstInactiveBucket)).isEmpty();
+ assertThat(logFetcher.getLogScannerStatus().getBucketOffset(firstActiveBucket))
+ .isEqualTo(4L);
+ assertThat(logFetcher.getLogScannerStatus().getBucketOffset(firstInactiveBucket))
+ .isEqualTo(0L);
+ }
+
+ private LogFetcher createFetcherWithBuckets(Map<TableBucket, Long> scanBuckets) {
+ LogScannerStatus scannerStatus = new LogScannerStatus();
+ scannerStatus.assignScanBuckets(scanBuckets);
+
+ PredicateBuilder builder = new PredicateBuilder(FILTER_TEST_ROW_TYPE);
+ Predicate recordBatchFilter = builder.greaterThan(0, 5); // a > 5
+
+ TestingScannerMetricGroup scannerMetricGroup = TestingScannerMetricGroup.newInstance();
+ return new LogFetcher(
+ DATA1_TABLE_INFO,
+ null,
+ recordBatchFilter,
+ scannerStatus,
+ clientConf,
+ metadataUpdater,
+ scannerMetricGroup,
+ new RemoteFileDownloader(1),
+ TEST_SCHEMA_GETTER);
+ }
+
private void addRecordsToBucket(TableBucket tableBucket, MemoryLogRecords logRecords)
throws Exception {
int leader = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tableBucket);