fix set bucket

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/batch/FullScanBatchScanner.java

@@ -40,16 +40,15 @@
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
 /**
- * BatchScanner that performs a FULL_SCAN snapshot across all leaders of a KV table (or a
- * single partition) and exposes the result via pollBatch in a bounded fashion.
+ * BatchScanner that performs a FULL_SCAN snapshot across all leaders of a KV table (or a single
+ * partition) and exposes the result via pollBatch in a bounded fashion.
  */
 @Internal
 public class FullScanBatchScanner implements BatchScanner {
@@ -86,17 +85,13 @@ private void startFullScan() {
             long tableId = tableInfo.getTableId();
             int numBuckets = tableInfo.getNumBuckets();
 
-            // Find leader tablet/servers for this table/partition
-            HashSet<Integer> leaderServers = new HashSet<>();
+            // Send full scan request per bucket to its leader tablet/server
+            List<CompletableFuture<FullScanResponse>> responseFutures = new ArrayList<>();
             for (int bucketId = 0; bucketId < numBuckets; bucketId++) {
                 TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
                 metadataUpdater.checkAndUpdateMetadata(tableInfo.getTablePath(), tableBucket);
                 int leader = metadataUpdater.leaderFor(tableBucket);
-                leaderServers.add(leader);
-            }
 
-            List<CompletableFuture<FullScanResponse>> responseFutures = new ArrayList<>();
-            for (int leader : leaderServers) {
                 TabletServerGateway gateway = metadataUpdater.newTabletServerClientForNode(leader);
                 if (gateway == null) {
                     rowsFuture.completeExceptionally(
@@ -108,8 +103,7 @@ private void startFullScan() {
                 }
                 FullScanRequest request = new FullScanRequest();
                 request.setTableId(tableId);
-                request.setBucketId(0);
-
+                request.setBucketId(bucketId);
                 if (partitionId != null) {
                     request.setPartitionId(partitionId);
                 }

fluss-client/src/test/java/org/apache/fluss/client/table/scanner/batch/FullScanBatchScannerITCase.java

@@ -57,9 +57,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-/**
- * End-to-end IT cases for FullScanBatchScanner functionality.
- */
+/** End-to-end IT cases for FullScanBatchScanner functionality. */
 class FullScanBatchScannerITCase {
 
     @RegisterExtension