Skip to content

[server] batch fetch latest bucket snapshots to reduce latency #3811

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zuston wants to merge 1 commit into
base: main
Choose a base branch
from
+159 −15
Open

[server] batch fetch latest bucket snapshots to reduce latency #3811

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 37 additions & 15 deletions fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1261,24 +1261,45 @@ private Map<Integer, Optional<BucketSnapshot>> getBucketSnapshots(
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;
}
Expand Down Expand Up @@ -1893,7 +1914,8 @@ private List<ZkGetChildrenResponse> getChildrenInBackground(Collection<String> p
* @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);
Expand Down
122 changes: 122 additions & 0 deletions fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
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;
Expand Down Expand Up @@ -77,6 +78,9 @@
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 {
Expand Down Expand Up @@ -531,6 +535,124 @@ void testTableBucketSnapshot() throws Exception {
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<>();
Expand Down
Loading