You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/08/06 13:38:45 UTC
(fluss) branch main updated: [server] batch fetch latest bucket snapshots to reduce latency (#3811)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia 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 a72ef624e [server] batch fetch latest bucket snapshots to reduce latency (#3811)
a72ef624e is described below
commit a72ef624e07955c5a836730c3d09cb93ed3982ec
Author: Junfan Zhang <zu...@apache.org>
AuthorDate: Thu Aug 6 21:38:40 2026 +0800
[server] batch fetch latest bucket snapshots to reduce latency (#3811)
Co-authored-by: zhangjunfan <zh...@qiyi.com>
---
.../apache/fluss/server/zk/ZooKeeperClient.java | 52 ++++++---
.../fluss/server/zk/ZooKeeperClientTest.java | 122 +++++++++++++++++++++
2 files changed, 159 insertions(+), 15 deletions(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
index e9eab00e0..1016c9512 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
@@ -1261,24 +1261,45 @@ public class ZooKeeperClient implements AutoCloseable {
partitionId == null
? BucketIdsZNode.pathOfTable(tableId)
: BucketIdsZNode.pathOfPartition(partitionId);
- // iterate all buckets
+ Map<String, TableBucket> snapshotPathToTableBucket = new HashMap<>();
for (String bucketIdStr : getChildren(bucketIdsPath)) {
- // get the bucket id
int bucketId = Integer.parseInt(bucketIdStr);
+ snapshots.put(bucketId, Optional.empty());
TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
- // get the snapshot node for the bucket
- String bucketSnapshotPath = BucketSnapshotsZNode.path(tableBucket);
- // get all the snapshots for the bucket
- List<String> bucketSnapshots = getChildren(bucketSnapshotPath);
-
- Optional<Long> optLatestSnapshotId =
- bucketSnapshots.stream().map(Long::parseLong).reduce(Math::max);
- Optional<BucketSnapshot> optTableBucketSnapshot = Optional.empty();
- if (optLatestSnapshotId.isPresent()) {
- optTableBucketSnapshot =
- getTableBucketSnapshot(tableBucket, optLatestSnapshotId.get());
+ snapshotPathToTableBucket.put(BucketSnapshotsZNode.path(tableBucket), tableBucket);
+ }
+
+ List<ZkGetChildrenResponse> childrenResponses =
+ getChildrenInBackground(snapshotPathToTableBucket.keySet());
+ Map<String, Integer> snapshotDataPathToBucketId = new HashMap<>();
+ for (ZkGetChildrenResponse response : childrenResponses) {
+ if (response.getResultCode() == KeeperException.Code.NONODE) {
+ continue;
+ }
+ response.maybeThrow();
+
+ OptionalLong latestSnapshotId =
+ response.getChildren().stream().mapToLong(Long::parseLong).max();
+ if (latestSnapshotId.isPresent()) {
+ TableBucket tableBucket =
+ checkNotNull(snapshotPathToTableBucket.get(response.getPath()));
+ snapshotDataPathToBucketId.put(
+ BucketSnapshotIdZNode.path(tableBucket, latestSnapshotId.getAsLong()),
+ tableBucket.getBucket());
}
- snapshots.put(bucketId, optTableBucketSnapshot);
+ }
+
+ List<ZkGetDataResponse> dataResponses =
+ getDataInBackground(snapshotDataPathToBucketId.keySet());
+ for (ZkGetDataResponse response : dataResponses) {
+ if (response.getResultCode() == KeeperException.Code.NONODE) {
+ // The snapshot may be deleted between listing the children and reading its data.
+ continue;
+ }
+ response.maybeThrow();
+
+ int bucketId = checkNotNull(snapshotDataPathToBucketId.get(response.getPath()));
+ snapshots.put(bucketId, Optional.of(BucketSnapshotIdZNode.decode(response.getData())));
}
return snapshots;
}
@@ -1893,7 +1914,8 @@ public class ZooKeeperClient implements AutoCloseable {
* @return list of async responses for each path
* @throws Exception if there is an error during the operation
*/
- private List<ZkGetDataResponse> getDataInBackground(Collection<String> paths) throws Exception {
+ @VisibleForTesting
+ List<ZkGetDataResponse> getDataInBackground(Collection<String> paths) throws Exception {
List<ZkGetDataRequest> requests =
paths.stream().map(ZkGetDataRequest::new).collect(Collectors.toList());
return handleRequestInBackground(requests, ZkGetDataResponse::create);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
index ea753c885..cbc0b85c6 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
@@ -42,6 +42,7 @@ import org.apache.fluss.server.zk.data.ServerTags;
import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
import org.apache.fluss.server.zk.data.TabletServerRegistration;
+import org.apache.fluss.server.zk.data.ZkData.BucketIdZNode;
import org.apache.fluss.server.zk.data.lease.KvSnapshotLeaseMetadata;
import org.apache.fluss.shaded.curator5.org.apache.curator.CuratorZookeeperClient;
import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;
@@ -77,6 +78,9 @@ import static org.apache.fluss.server.utils.TableAssignmentUtils.generateAssignm
import static org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.common.ZKConfig.JUTE_MAXBUFFER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
/** Test for {@link ZooKeeperClient}. */
class ZooKeeperClientTest {
@@ -531,6 +535,124 @@ class ZooKeeperClientTest {
assertThat(zookeeperClient.getTableBucketSnapshot(table1Bucket2, 1)).isEmpty();
}
+ @Test
+ void testGetLatestBucketSnapshotsInBatch() throws Exception {
+ long tableId = 1L;
+ TableBucket bucketWithSnapshots = new TableBucket(tableId, 0);
+ TableBucket bucketWithoutPath = new TableBucket(tableId, 1);
+ TableBucket bucketWithoutSnapshotsPath = new TableBucket(tableId, 2);
+ TableBucket bucketWithEmptySnapshotsPath = new TableBucket(tableId, 3);
+ TableBucket unassignedBucketWithoutSnapshotsPath = new TableBucket(tableId, 4);
+ zookeeperClient.registerTableAssignment(
+ tableId,
+ TableAssignment.builder()
+ .add(bucketWithSnapshots.getBucket(), BucketAssignment.of(0))
+ .add(bucketWithoutPath.getBucket(), BucketAssignment.of(0))
+ .add(bucketWithoutSnapshotsPath.getBucket(), BucketAssignment.of(0))
+ .add(bucketWithEmptySnapshotsPath.getBucket(), BucketAssignment.of(0))
+ .build());
+
+ BucketSnapshot olderSnapshot = new BucketSnapshot(1L, 10L, "oss://test/cp1");
+ BucketSnapshot latestSnapshot = new BucketSnapshot(3L, 30L, "oss://test/cp3");
+ zookeeperClient.registerTableBucketSnapshot(bucketWithSnapshots, olderSnapshot);
+ zookeeperClient.registerTableBucketSnapshot(bucketWithSnapshots, latestSnapshot);
+
+ // Create a bucket path without its snapshots child path.
+ zookeeperClient
+ .getCuratorClient()
+ .create()
+ .creatingParentsIfNeeded()
+ .forPath(BucketIdZNode.path(bucketWithoutSnapshotsPath));
+ zookeeperClient
+ .getCuratorClient()
+ .create()
+ .creatingParentsIfNeeded()
+ .forPath(BucketIdZNode.path(unassignedBucketWithoutSnapshotsPath));
+
+ // Deleting the only snapshot leaves an empty snapshots path.
+ BucketSnapshot deletedSnapshot = new BucketSnapshot(2L, 20L, "oss://test/cp2");
+ zookeeperClient.registerTableBucketSnapshot(bucketWithEmptySnapshotsPath, deletedSnapshot);
+ zookeeperClient.deleteTableBucketSnapshot(
+ bucketWithEmptySnapshotsPath, deletedSnapshot.getSnapshotId());
+
+ Map<Integer, Optional<BucketSnapshot>> expectedSnapshots = new HashMap<>();
+ expectedSnapshots.put(bucketWithSnapshots.getBucket(), Optional.of(latestSnapshot));
+ expectedSnapshots.put(bucketWithoutPath.getBucket(), Optional.empty());
+ expectedSnapshots.put(bucketWithoutSnapshotsPath.getBucket(), Optional.empty());
+ expectedSnapshots.put(bucketWithEmptySnapshotsPath.getBucket(), Optional.empty());
+ expectedSnapshots.put(unassignedBucketWithoutSnapshotsPath.getBucket(), Optional.empty());
+
+ assertThat(zookeeperClient.getTableLatestBucketSnapshot(tableId))
+ .isEqualTo(expectedSnapshots);
+
+ long partitionId = 2L;
+ Map<Integer, BucketAssignment> partitionBucketAssignments =
+ Collections.singletonMap(0, BucketAssignment.of(0));
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId,
+ "p1",
+ new PartitionAssignment(tableId, partitionBucketAssignments),
+ remoteDataDir,
+ TablePath.of("db", "partitioned_table"),
+ tableId);
+ TableBucket partitionBucket = new TableBucket(tableId, partitionId, 0);
+ BucketSnapshot partitionSnapshot = new BucketSnapshot(5L, 50L, "oss://test/partition-cp5");
+ zookeeperClient.registerTableBucketSnapshot(partitionBucket, partitionSnapshot);
+
+ assertThat(zookeeperClient.getPartitionLatestBucketSnapshot(partitionId))
+ .containsEntry(partitionBucket.getBucket(), Optional.of(partitionSnapshot));
+ }
+
+ @Test
+ void testGetLatestBucketSnapshotsBeyondMaxInflightRequests() throws Exception {
+ long tableId = 1L;
+ int bucketCount = ConfigOptions.ZOOKEEPER_MAX_INFLIGHT_REQUESTS.defaultValue() + 1;
+ TableAssignment.Builder assignmentBuilder = TableAssignment.builder();
+ Map<Integer, Optional<BucketSnapshot>> expectedSnapshots = new HashMap<>();
+ for (int bucketId = 0; bucketId < bucketCount; bucketId++) {
+ assignmentBuilder.add(bucketId, BucketAssignment.of(0));
+ BucketSnapshot snapshot = new BucketSnapshot(1L, bucketId, "oss://test/cp-" + bucketId);
+ expectedSnapshots.put(bucketId, Optional.of(snapshot));
+ }
+ zookeeperClient.registerTableAssignment(tableId, assignmentBuilder.build());
+
+ for (Map.Entry<Integer, Optional<BucketSnapshot>> entry : expectedSnapshots.entrySet()) {
+ zookeeperClient.registerTableBucketSnapshot(
+ new TableBucket(tableId, entry.getKey()), entry.getValue().get());
+ }
+
+ assertThat(zookeeperClient.getTableLatestBucketSnapshot(tableId))
+ .isEqualTo(expectedSnapshots);
+ }
+
+ @Test
+ void testLatestBucketSnapshotDeletedBetweenBatchReads() throws Exception {
+ long tableId = 1L;
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ zookeeperClient.registerTableAssignment(
+ tableId,
+ TableAssignment.builder()
+ .add(tableBucket.getBucket(), BucketAssignment.of(0))
+ .build());
+ BucketSnapshot olderSnapshot = new BucketSnapshot(1L, 10L, "oss://test/cp1");
+ BucketSnapshot latestSnapshot = new BucketSnapshot(2L, 20L, "oss://test/cp2");
+ zookeeperClient.registerTableBucketSnapshot(tableBucket, olderSnapshot);
+ zookeeperClient.registerTableBucketSnapshot(tableBucket, latestSnapshot);
+
+ ZooKeeperClient raceTestingClient = spy(zookeeperClient);
+ doAnswer(
+ invocation -> {
+ zookeeperClient.deleteTableBucketSnapshot(
+ tableBucket, latestSnapshot.getSnapshotId());
+ return invocation.callRealMethod();
+ })
+ .when(raceTestingClient)
+ .getDataInBackground(anyCollection());
+
+ assertThat(raceTestingClient.getTableLatestBucketSnapshot(tableId))
+ .containsEntry(tableBucket.getBucket(), Optional.empty());
+ }
+
@Test
void testKvSnapshotLease() throws Exception {
Map<Long, FsPath> tableIdToRemotePath = new HashMap<>();