This is an automated email from the ASF dual-hosted git repository. wuchong pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/fluss.git
commit a70f96e0eef07ddc83defc592a6f54ac91a5b594 Author: Jark Wu <[email protected]> AuthorDate: Mon May 25 00:30:13 2026 +0800 [client] call_seq_id should start from zero for ScanKv open and continuation requests --- .../client/table/scanner/batch/KvBatchScanner.java | 18 +++++---- .../fluss/client/table/TableKvScanITCase.java | 19 +++++++++- .../table/scanner/batch/KvBatchScannerTest.java | 7 ++-- .../apache/fluss/server/tablet/TabletService.java | 26 ++++++++----- .../fluss/server/tablet/TabletServiceITCase.java | 43 +++++++++++++++++++++- 5 files changed, 90 insertions(+), 23 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java index 5a99d88b2..2062f9c1f 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/KvBatchScanner.java @@ -45,7 +45,6 @@ import javax.annotation.Nullable; import javax.annotation.concurrent.NotThreadSafe; import java.io.IOException; -import java.nio.ByteBuffer; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; @@ -134,6 +133,10 @@ public final class KvBatchScanner implements BatchScanner { } catch (TimeoutException e) { return CloseableIterator.emptyIterator(); } catch (Exception e) { + // TODO: we should retry continuation requests on transient networks errors, but + // for simplicity we currently fail the whole scan. Retrying here would require robust + // handling of duplicate continuations (e.g. call_seq_id rollback) to avoid skipping or + // repeating rows. terminate(); throw new IOException(e); } @@ -183,18 +186,20 @@ public final class KvBatchScanner implements BatchScanner { } ScanKvRequest request = - new ScanKvRequest().setBucketScanReq(bucketReq).setBatchSizeBytes(batchSizeBytes); + new ScanKvRequest() + .setBucketScanReq(bucketReq) + .setBatchSizeBytes(batchSizeBytes) + .setCallSeqId(callSeqId); inFlight = gateway.scanKv(request); } private void sendContinuation() { - // Pre-increment so the first continuation sends call_seq_id=1; the server initializes its - // counter to 0 and expects requestSeqId == contextSeqId + 1. + callSeqId++; ScanKvRequest request = new ScanKvRequest() .setScannerId(scannerId) .setBatchSizeBytes(batchSizeBytes) - .setCallSeqId(++callSeqId); + .setCallSeqId(callSeqId); inFlight = gateway.scanKv(request); } @@ -300,9 +305,8 @@ public final class KvBatchScanner implements BatchScanner { } private List<InternalRow> parseRecords(ScanKvResponse response) { - ByteBuffer recordsBuffer = ByteBuffer.wrap(response.getRecords()); DefaultValueRecordBatch valueRecords = - DefaultValueRecordBatch.pointToByteBuffer(recordsBuffer); + DefaultValueRecordBatch.pointToBytes(response.getRecords()); int recordCount = valueRecords.getRecordCount(); if (recordCount == 0) { return Collections.emptyList(); diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/TableKvScanITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/TableKvScanITCase.java index 3161a053f..6d79c1316 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/table/TableKvScanITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/table/TableKvScanITCase.java @@ -17,10 +17,14 @@ package org.apache.fluss.client.table; +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; import org.apache.fluss.client.admin.ClientToServerITCaseBase; import org.apache.fluss.client.table.scanner.batch.BatchScanUtils; import org.apache.fluss.client.table.scanner.batch.BatchScanner; import org.apache.fluss.client.table.writer.UpsertWriter; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; import org.apache.fluss.metadata.PartitionInfo; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; @@ -126,21 +130,32 @@ class TableKvScanITCase extends ClientToServerITCaseBase { long tableId = createTable(tablePath, singleBucket, true); waitAllReplicasReady(tableId, 1); - try (Table table = conn.getTable(tablePath)) { + // Use a very small fetch max bytes to ensure the first poll returns only partial rows, + // so subsequent collectRows will issue new scanKv RPCs to truly verify snapshot isolation. + Configuration smallFetchConf = new Configuration(clientConf); + smallFetchConf.setString(ConfigOptions.CLIENT_SCANNER_KV_FETCH_MAX_BYTES.key(), "128b"); + + try (Connection smallFetchConn = ConnectionFactory.createConnection(smallFetchConf); + Table table = smallFetchConn.getTable(tablePath)) { int initialRows = 20; upsertRows(table, initialRows); try (BatchScanner scanner = table.newScan().createBatchScanner()) { + // Pull the first batch (snapshot is pinned; small fetch bytes returns partial rows) List<InternalRow> firstBatch = drainOnePoll(scanner); + assertThat(firstBatch.size()).isLessThan(initialRows); + // Write a new row after the snapshot (id=20, name="post-snapshot") upsertRow(table, initialRows, "post-snapshot"); + // Continue pulling remaining data (triggers new scanKv RPCs) List<InternalRow> remaining = BatchScanUtils.collectRows(scanner); List<InternalRow> all = new ArrayList<>(firstBatch); all.addAll(remaining); + // Total rows should be 20 (excluding the row written after the snapshot) assertThat(all).hasSize(initialRows); - assertThat(idsOf(all)).doesNotContain(initialRows); + assertThat(idsOf(all)).doesNotContain(initialRows); // id=20 not visible } } } diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvBatchScannerTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvBatchScannerTest.java index 75f60e664..fad28096f 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvBatchScannerTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/KvBatchScannerTest.java @@ -66,7 +66,7 @@ class KvBatchScannerTest { new TestingSchemaGetter((short) 1, DATA1_SCHEMA_PK); @Test - void firstPollOpensScannerAndDoesNotIncludeCallSeqId() throws Exception { + void firstPollOpensScannerWithCallSeqIdZero() throws Exception { RecordingGateway gateway = new RecordingGateway(); gateway.enqueue(emptyTerminalResponse(SCANNER_ID)); @@ -76,7 +76,8 @@ class KvBatchScannerTest { ScanKvRequest open = gateway.requests.get(0); assertThat(open.hasBucketScanReq()).isTrue(); assertThat(open.hasScannerId()).isFalse(); - assertThat(open.hasCallSeqId()).isFalse(); + assertThat(open.hasCallSeqId()).isTrue(); // open request also carries callSeqId + assertThat(open.getCallSeqId()).isEqualTo(0); // open request with 0 seq_id assertThat(open.getBucketScanReq().getTableId()).isEqualTo(DATA1_TABLE_ID_PK); assertThat(open.getBucketScanReq().getBucketId()).isEqualTo(0); } @@ -97,7 +98,7 @@ class KvBatchScannerTest { } assertThat(gateway.requests).hasSize(4); - assertThat(gateway.requests.get(0).hasCallSeqId()).isFalse(); + assertThat(gateway.requests.get(0).getCallSeqId()).isEqualTo(0); assertThat(gateway.requests.get(1).getCallSeqId()).isEqualTo(1); assertThat(gateway.requests.get(2).getCallSeqId()).isEqualTo(2); assertThat(gateway.requests.get(3).getCallSeqId()).isEqualTo(3); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java index 88731daab..bdeff1818 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java @@ -551,22 +551,28 @@ public final class TabletService extends RpcServiceBase implements TabletServerG } acquiredContext = context; - if (!request.hasBucketScanReq() && request.hasCallSeqId()) { - long expectedSeqId = (long) context.getCallSeqId() + 1L; + // Honour close even on a non-leader: the local session is still ours to release. + // Close requests are exempt from callSeqId validation. + if (isCloseRequest) { + response.setScannerId(context.getScannerId()); + response.setHasMoreResults(false); + return CompletableFuture.completedFuture(response); + } + + // Validate callSeqId for all data-fetching requests (open + continuation). + if (request.hasCallSeqId()) { + int expectedSeqId = context.getCallSeqId() + 1; int requestSeqId = request.getCallSeqId(); - if ((long) requestSeqId != expectedSeqId) { + if (requestSeqId != expectedSeqId) { throw new InvalidScanRequestException( String.format( "Out-of-order scan request: expected callSeqId=%d but got %d.", expectedSeqId, requestSeqId)); } - } - - // Honour close even on a non-leader: the local session is still ours to release. - if (isCloseRequest) { - response.setScannerId(context.getScannerId()); - response.setHasMoreResults(false); - return CompletableFuture.completedFuture(response); + } else { + throw new InvalidScanRequestException( + "call_seq_id is required for ScanKV requests to ensure " + + "in-order processing and at-least-once semantics."); } // Catch a leadership flip ahead of the eventual closeScannersForBucket callback so diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java index 8adb942da..5d2dd1d71 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java @@ -1140,7 +1140,7 @@ public class TabletServiceITCase { DefaultValueRecordBatch.pointToBytes(first.getRecords()).getRecordCount(); ScanKvResponse current = first; - int seq = 0; + int seq = 1; while (current.isHasMoreResults()) { current = leaderGateWay @@ -1340,6 +1340,46 @@ public class TabletServiceITCase { assertThat(bad.getErrorMessage()).contains("Out-of-order"); } + @Test + void testScanKv_missingCallSeqIdIsRejected() throws Exception { + long tableId = + createTable( + FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH_PK, DATA1_TABLE_DESCRIPTOR_PK); + TableBucket tb = new TableBucket(tableId, 0); + FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb); + int leader = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateWay = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader); + + assertPutKvResponse( + leaderGateWay + .putKv( + newPutKvRequest( + tableId, 0, 1, genKvRecordBatch(DATA_1_WITH_KEY_AND_VALUE))) + .get()); + FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tb); + + // Open request without call_seq_id should be rejected. + ScanKvRequest openWithoutSeqId = new ScanKvRequest(); + openWithoutSeqId.setBucketScanReq().setTableId(tableId).setBucketId(0); + openWithoutSeqId.setBatchSizeBytes(1024); + ScanKvResponse openResp = leaderGateWay.scanKv(openWithoutSeqId).get(); + assertThat(openResp.getErrorCode()).isEqualTo(Errors.INVALID_SCAN_REQUEST.code()); + assertThat(openResp.getErrorMessage()).contains("call_seq_id is required"); + + // Continuation request without call_seq_id should also be rejected. + ScanKvResponse open = leaderGateWay.scanKv(newScanKvOpenRequest(tableId, 0, 1)).get(); + assertThat(open.hasErrorCode()).isFalse(); + byte[] scannerId = open.getScannerId(); + + ScanKvRequest contWithoutSeqId = new ScanKvRequest(); + contWithoutSeqId.setScannerId(scannerId); + contWithoutSeqId.setBatchSizeBytes(1024); + ScanKvResponse contResp = leaderGateWay.scanKv(contWithoutSeqId).get(); + assertThat(contResp.getErrorCode()).isEqualTo(Errors.INVALID_SCAN_REQUEST.code()); + assertThat(contResp.getErrorMessage()).contains("call_seq_id is required"); + } + @Test void testScanKv_oversizeBatchSizeBytesIsClamped() throws Exception { long tableId = @@ -1392,6 +1432,7 @@ public class TabletServiceITCase { ScanKvRequest req = new ScanKvRequest(); req.setBucketScanReq().setTableId(tableId).setBucketId(bucketId); req.setBatchSizeBytes(batchSize); + req.setCallSeqId(0); return req; }
