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 2025/09/26 10:41:44 UTC
(fluss) branch main updated (f9ea35301 -> 4345663fe)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
from f9ea35301 [hotfix] Revert CURRENT_LOG_MAGIC_VALUE to LOG_MAGIC_VALUE_V0 for compatibility higher version client with lower version server (#1743)
new 32d50b8b6 [server] server process metadata request in async mode (#1463)
new 4345663fe [server] Use unified batch-concurrent implementation for ZooKeeper operations and remove redundant code
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../client/table/AutoPartitionedTableITCase.java | 2 +-
.../apache/fluss/utils/concurrent/FutureUtils.java | 83 +++
.../org.junit.jupiter.api.extension.Extension | 0
.../src/test/resources/log4j2-test.properties | 0
.../org.junit.jupiter.api.extension.Extension | 0
.../src/test/resources/log4j2-test.properties | 0
.../org/apache/fluss/server/RpcServiceBase.java | 257 +++------
.../server/coordinator/CoordinatorService.java | 110 +---
.../fluss/server/coordinator/MetadataManager.java | 10 +-
.../metadata/CoordinatorMetadataProvider.java | 149 +++++
.../fluss/server/metadata/MetadataProvider.java | 74 +++
.../server/metadata/TabletServerMetadataCache.java | 185 +++++--
.../metadata/TabletServerMetadataProvider.java | 100 ++++
.../server/metadata/ZkBasedMetadataProvider.java | 124 +++++
.../apache/fluss/server/tablet/TabletServer.java | 2 +-
.../apache/fluss/server/tablet/TabletService.java | 13 +-
.../apache/fluss/server/zk/ZooKeeperClient.java | 605 ++++++++++++++-------
.../server/metadata/MetadataUpdateITCase.java | 40 +-
.../metadata/TabletServerMetadataCacheTest.java | 207 ++++++-
.../metadata/ZkBasedMetadataProviderTest.java | 349 ++++++++++++
.../fluss/server/replica/ReplicaManagerTest.java | 16 +-
.../fluss/server/replica/ReplicaTestBase.java | 3 +-
.../replica/fetcher/ReplicaFetcherThreadTest.java | 2 +-
23 files changed, 1763 insertions(+), 568 deletions(-)
copy {fluss-client => fluss-flink/fluss-flink-1.20}/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension (100%)
copy fluss-flink/{fluss-flink-1.19 => fluss-flink-1.20}/src/test/resources/log4j2-test.properties (100%)
copy {fluss-client => fluss-flink/fluss-flink-2.1}/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension (100%)
copy fluss-flink/{fluss-flink-1.19 => fluss-flink-2.1}/src/test/resources/log4j2-test.properties (100%)
create mode 100644 fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java
create mode 100644 fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java
create mode 100644 fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java
create mode 100644 fluss-server/src/main/java/org/apache/fluss/server/metadata/ZkBasedMetadataProvider.java
create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java
(fluss) 01/02: [server] server process metadata request in async mode (#1463)
Posted by ja...@apache.org.
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
commit 32d50b8b6d130a6cf871dc24879d3a959e5df39f
Author: ocean.wy <oc...@alibaba-inc.com>
AuthorDate: Wed Aug 27 15:08:59 2025 +0800
[server] server process metadata request in async mode (#1463)
---
.../apache/fluss/exception/CacheMissException.java | 36 +
.../apache/fluss/utils/concurrent/FutureUtils.java | 83 +++
.../java/org/apache/fluss/rpc/protocol/Errors.java | 4 +
fluss-rpc/src/main/proto/FlussApi.proto | 4 +
.../org/apache/fluss/server/RpcServiceBase.java | 355 ++++-----
.../server/coordinator/CoordinatorService.java | 124 +---
.../metadata/CoordinatorMetadataProvider.java | 169 +++++
.../fluss/server/metadata/MetadataProvider.java | 73 ++
.../server/metadata/TabletServerMetadataCache.java | 180 ++++-
.../metadata/TabletServerMetadataProvider.java | 115 +++
.../apache/fluss/server/tablet/TabletService.java | 24 +-
.../apache/fluss/server/zk/ZooKeeperClient.java | 813 +++++++++++++++++----
.../fluss/server/zk/ZooKeeperMetadataUtils.java | 478 ++++++++++++
.../server/metadata/MetadataUpdateITCase.java | 40 +-
.../metadata/TabletServerMetadataCacheTest.java | 204 +++++-
.../fluss/server/replica/ReplicaManagerTest.java | 16 +-
.../fluss/server/tablet/TabletServiceITCase.java | 139 ++++
.../fluss/server/zk/ZooKeeperClientTest.java | 264 +++++++
18 files changed, 2596 insertions(+), 525 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java b/fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java
new file mode 100644
index 000000000..e9ff03373
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.exception;
+
+/**
+ * Exception thrown when requested metadata is not available in the local cache and cache_only is
+ * enabled.
+ *
+ * @since 0.8
+ */
+public class CacheMissException extends ApiException {
+ private static final long serialVersionUID = 1L;
+
+ public CacheMissException(String message) {
+ super(message);
+ }
+
+ public CacheMissException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/FutureUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/FutureUtils.java
index 91061d031..d13862cf1 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/FutureUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/concurrent/FutureUtils.java
@@ -30,6 +30,7 @@ import javax.annotation.Nullable;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -45,6 +46,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
+import java.util.function.Function;
import java.util.function.Supplier;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
@@ -773,6 +775,87 @@ public class FutureUtils {
}
}
+ /**
+ * Core method to execute async-to-sync conversion with unified exception handling. This method
+ * contains all the common exception handling logic.
+ *
+ * @param futureSupplier supplier that provides the CompletableFuture
+ * @param operationName name of the operation for error messages
+ * @param <T> the type of the result
+ * @return the result directly from the CompletableFuture
+ * @throws Exception if any error occurs during the operation
+ */
+ private static <T> T asyncToSyncInternal(
+ Supplier<CompletableFuture<T>> futureSupplier, String operationName) throws Exception {
+ try {
+ return futureSupplier.get().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while waiting for " + operationName, e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof RuntimeException) {
+ throw (RuntimeException) cause;
+ } else if (cause instanceof Exception) {
+ throw (Exception) cause;
+ } else {
+ throw new RuntimeException("Unexpected error getting " + operationName, cause);
+ }
+ }
+ }
+
+ /**
+ * Template method to convert async CompletableFuture to synchronous Optional result.
+ *
+ * @param futureSupplier supplier that provides the CompletableFuture
+ * @param operationName name of the operation for error messages
+ * @param <T> the type of the result
+ * @return Optional containing the result, or empty if the result is null
+ * @throws Exception if any error occurs during the operation
+ */
+ public static <T> Optional<T> asyncToSync(
+ Supplier<CompletableFuture<T>> futureSupplier, String operationName) throws Exception {
+ T result = asyncToSyncInternal(futureSupplier, operationName);
+ return Optional.ofNullable(result);
+ }
+
+ /**
+ * Template method to convert async CompletableFuture to synchronous Optional result with
+ * transformation.
+ *
+ * @param futureSupplier supplier that provides the CompletableFuture
+ * @param transformer function to transform the result before wrapping in Optional
+ * @param operationName name of the operation for error messages
+ * @param <T> the type of the future result
+ * @param <R> the type of the final result
+ * @return Optional containing the transformed result, or empty if the result is null
+ * @throws Exception if any error occurs during the operation
+ */
+ public static <T, R> Optional<R> asyncToSync(
+ Supplier<CompletableFuture<T>> futureSupplier,
+ Function<T, R> transformer,
+ String operationName)
+ throws Exception {
+ T result = asyncToSyncInternal(futureSupplier, operationName);
+ R transformedResult = result == null ? null : transformer.apply(result);
+ return Optional.ofNullable(transformedResult);
+ }
+
+ /**
+ * Template method to convert async CompletableFuture to synchronous result directly (not
+ * wrapped in Optional).
+ *
+ * @param futureSupplier supplier that provides the CompletableFuture
+ * @param operationName name of the operation for error messages
+ * @param <T> the type of the result
+ * @return the result directly (can be null)
+ * @throws Exception if any error occurs during the operation
+ */
+ public static <T> T asyncToSyncDirect(
+ Supplier<CompletableFuture<T>> futureSupplier, String operationName) throws Exception {
+ return asyncToSyncInternal(futureSupplier, operationName);
+ }
+
/**
* Wraps a Runnable so that throwables are caught and logged when a Runnable is run.
*
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
index 3b899baee..e3b5007a3 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
@@ -20,6 +20,7 @@ package org.apache.fluss.rpc.protocol;
import org.apache.fluss.exception.ApiException;
import org.apache.fluss.exception.AuthenticationException;
import org.apache.fluss.exception.AuthorizationException;
+import org.apache.fluss.exception.CacheMissException;
import org.apache.fluss.exception.CorruptMessageException;
import org.apache.fluss.exception.CorruptRecordException;
import org.apache.fluss.exception.DatabaseAlreadyExistException;
@@ -224,6 +225,9 @@ public enum Errors {
55,
"The new ISR contains at least one ineligible replica.",
IneligibleReplicaException::new),
+ CACHE_MISS(
+ 56, "The requested metadata is not available in local cache.", CacheMissException::new);
+ IneligibleReplicaException::new),
INVALID_ALTER_TABLE_EXCEPTION(
56, "The alter table is invalid.", InvalidAlterTableException::new);
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 46db51849..3c6b8b358 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -176,6 +176,10 @@ message MetadataRequest {
// metadata request
// todo: we won't need the assumption after we introduce metadata cache in server
repeated int64 partitions_id = 3 [packed = true];
+
+ // if cache_only is true, return immediately if metadata is not in local cache
+ // instead of waiting for ZooKeeper access to complete
+ optional bool cache_only = 4;
}
message MetadataResponse {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
index bbf135bf4..77983c863 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server;
import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.exception.CacheMissException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.KvSnapshotNotExistException;
import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
@@ -81,18 +82,15 @@ import org.apache.fluss.server.authorizer.Authorizer;
import org.apache.fluss.server.coordinator.CoordinatorService;
import org.apache.fluss.server.coordinator.MetadataManager;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
-import org.apache.fluss.server.metadata.BucketMetadata;
+import org.apache.fluss.server.metadata.MetadataProvider;
import org.apache.fluss.server.metadata.PartitionMetadata;
import org.apache.fluss.server.metadata.ServerMetadataCache;
import org.apache.fluss.server.metadata.TableMetadata;
import org.apache.fluss.server.tablet.TabletService;
import org.apache.fluss.server.utils.ServerRpcMessageUtils;
import org.apache.fluss.server.zk.ZooKeeperClient;
-import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.BucketSnapshot;
import org.apache.fluss.server.zk.data.LakeTableSnapshot;
-import org.apache.fluss.server.zk.data.LeaderAndIsr;
-import org.apache.fluss.server.zk.data.TableAssignment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -107,7 +105,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.function.Function;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclFilter;
@@ -121,7 +119,6 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeListAclsRe
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toGetFileSystemSecurityTokenResponse;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toListPartitionInfosResponse;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toTablePath;
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
import static org.apache.fluss.utils.Preconditions.checkState;
/**
@@ -466,231 +463,171 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
}
}
- protected MetadataResponse makeMetadataResponse(
+ protected void processMetadataRequest(
MetadataRequest request,
String listenerName,
Session session,
Authorizer authorizer,
ServerMetadataCache metadataCache,
- Function<TablePath, TableMetadata> getTableMetadataFunc,
- Function<Long, Optional<PhysicalTablePath>> getPhysicalTablePathFunc,
- Function<PhysicalTablePath, PartitionMetadata> getPartitionMetadataFunc) {
+ MetadataProvider functionProvider,
+ CompletableFuture<MetadataResponse> responseFuture) {
List<PbTablePath> pbTablePaths = request.getTablePathsList();
- List<TableMetadata> tableMetadata = new ArrayList<>();
-
+ List<CompletableFuture<TableMetadata>> pendingTableMetadata = new ArrayList<>();
List<TablePath> tablePaths = new ArrayList<>(pbTablePaths.size());
- for (PbTablePath pbTablePath : pbTablePaths) {
- if (authorizer == null
- || authorizer.isAuthorized(
- session,
- OperationType.DESCRIBE,
- Resource.table(
- pbTablePath.getDatabaseName(), pbTablePath.getTableName()))) {
- TablePath tablePath = toTablePath(pbTablePath);
- tableMetadata.add(getTableMetadataFunc.apply(tablePath));
- tablePaths.add(tablePath);
- }
- }
-
- List<PartitionMetadata> partitionMetadata = new ArrayList<>();
- List<PhysicalTablePath> partitions =
- request.getPartitionsPathsList().stream()
- .map(ServerRpcMessageUtils::toPhysicalTablePath)
- .collect(Collectors.toList());
- long[] partitionIds = request.getPartitionsIds();
- Set<Long> partitionIdsNotExistsInCache = new HashSet<>();
- for (long partitionId : partitionIds) {
- Optional<PhysicalTablePath> physicalTablePath =
- getPhysicalTablePathFunc.apply(partitionId);
- if (physicalTablePath.isPresent()) {
- partitions.add(physicalTablePath.get());
- } else {
- partitionIdsNotExistsInCache.add(partitionId);
- }
- }
-
- if (!partitionIdsNotExistsInCache.isEmpty()) {
- partitionMetadata.addAll(
- getPartitionMetadataFromZk(tablePaths, partitionIdsNotExistsInCache));
- }
+ // result collections - using ConcurrentLinkedQueue for thread safety
+ ConcurrentLinkedQueue<TableMetadata> tableMetadata = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<PartitionMetadata> partitionMetadata = new ConcurrentLinkedQueue<>();
+ boolean cacheOnly = request.hasCacheOnly() && request.isCacheOnly();
+ List<String> cacheMissMessages = new ArrayList<>();
- for (PhysicalTablePath partitionPath : partitions) {
- if (authorizer == null
- || authorizer.isAuthorized(
- session,
- OperationType.DESCRIBE,
- Resource.table(
- partitionPath.getDatabaseName(),
- partitionPath.getTableName()))) {
- partitionMetadata.add(getPartitionMetadataFunc.apply(partitionPath));
+ try {
+ for (PbTablePath pbTablePath : pbTablePaths) {
+ if (authorizer == null
+ || authorizer.isAuthorized(
+ session,
+ OperationType.DESCRIBE,
+ Resource.table(
+ pbTablePath.getDatabaseName(),
+ pbTablePath.getTableName()))) {
+ TablePath tablePath = toTablePath(pbTablePath);
+ Optional<TableMetadata> tableMetadataFromCacheOpt =
+ functionProvider.getTableMetadataFromCache(tablePath);
+ if (!tableMetadataFromCacheOpt.isPresent()) {
+ if (cacheOnly) {
+ cacheMissMessages.add(
+ String.format(
+ "Table metadata for table %s is not available in local cache",
+ tablePath));
+ }
+ // Always add ZK query to fill cache, even for cache_only requests
+ pendingTableMetadata.add(
+ functionProvider
+ .getTableMetadataFromZk(tablePath)
+ .thenApply(
+ meta -> {
+ tableMetadata.add(meta);
+ return meta;
+ }));
+ } else {
+ tableMetadata.add(tableMetadataFromCacheOpt.get());
+ }
+ tablePaths.add(tablePath);
+ }
}
- }
- return buildMetadataResponse(
- metadataCache.getCoordinatorServer(listenerName),
- new HashSet<>(metadataCache.getAllAliveTabletServers(listenerName).values()),
- tableMetadata,
- partitionMetadata);
- }
+ List<CompletableFuture<PartitionMetadata>> pendingPartitionMetadata = new ArrayList<>();
+ CompletableFuture<List<PartitionMetadata>> pendingPartitionMetadataMissingIdsPartition =
+ CompletableFuture.completedFuture(new ArrayList<>());
- public static List<BucketMetadata> getTableMetadataFromZk(
- ZooKeeperClient zkClient, TablePath tablePath, long tableId, boolean isPartitioned) {
- try {
- AssignmentInfo assignmentInfo =
- getAssignmentInfo(tableId, PhysicalTablePath.of(tablePath), zkClient);
- List<BucketMetadata> bucketMetadataList = new ArrayList<>();
- if (assignmentInfo.tableAssignment != null) {
- TableAssignment tableAssignment = assignmentInfo.tableAssignment;
- bucketMetadataList =
- toBucketMetadataList(tableId, null, tableAssignment, null, zkClient);
- } else {
- if (isPartitioned) {
- LOG.warn("No table assignment node found for table {}", tableId);
+ List<PhysicalTablePath> partitions =
+ request.getPartitionsPathsList().stream()
+ .map(ServerRpcMessageUtils::toPhysicalTablePath)
+ .collect(Collectors.toList());
+ long[] partitionIds = request.getPartitionsIds();
+ Set<Long> partitionIdsNotExistsInCache = new HashSet<>();
+ for (long partitionId : partitionIds) {
+ Optional<PhysicalTablePath> physicalTablePath =
+ functionProvider.getPhysicalTablePathFromCache(partitionId);
+ if (physicalTablePath.isPresent()) {
+ partitions.add(physicalTablePath.get());
+ } else {
+ partitionIdsNotExistsInCache.add(partitionId);
}
}
- return bucketMetadataList;
- } catch (Exception e) {
- throw new FlussRuntimeException(
- String.format("Failed to get metadata for %s", tablePath), e);
- }
- }
- public static PartitionMetadata getPartitionMetadataFromZk(
- PhysicalTablePath partitionPath, ZooKeeperClient zkClient) {
- try {
- checkNotNull(
- partitionPath.getPartitionName(),
- "partitionName must be not null, but get: " + partitionPath);
- AssignmentInfo assignmentInfo = getAssignmentInfo(null, partitionPath, zkClient);
- List<BucketMetadata> bucketMetadataList = new ArrayList<>();
- checkNotNull(
- assignmentInfo.partitionId,
- "partition id must be not null for " + partitionPath);
- if (assignmentInfo.tableAssignment != null) {
- TableAssignment tableAssignment = assignmentInfo.tableAssignment;
- bucketMetadataList =
- toBucketMetadataList(
- assignmentInfo.tableId,
- assignmentInfo.partitionId,
- tableAssignment,
- null,
- zkClient);
- } else {
- LOG.warn("No partition assignment node found for partition {}", partitionPath);
+ if (!partitionIdsNotExistsInCache.isEmpty()) {
+ if (cacheOnly) {
+ cacheMissMessages.add(
+ String.format(
+ "Partition metadata for partition IDs %s is not available in local cache",
+ partitionIdsNotExistsInCache));
+ }
+ // Always execute ZK query to fill cache, even for cache_only requests
+ pendingPartitionMetadataMissingIdsPartition =
+ zkClient.batchGetPartitionMetadataFromZkAsync(
+ tablePaths, partitionIdsNotExistsInCache)
+ .thenApply(
+ batchMetadata -> {
+ partitionMetadata.addAll(batchMetadata);
+ return batchMetadata;
+ });
}
- return new PartitionMetadata(
- assignmentInfo.tableId,
- partitionPath.getPartitionName(),
- assignmentInfo.partitionId,
- bucketMetadataList);
- } catch (PartitionNotExistException e) {
- throw e;
- } catch (Exception e) {
- throw new FlussRuntimeException(
- String.format("Failed to get metadata for partition %s", partitionPath), e);
- }
- }
- private static List<BucketMetadata> toBucketMetadataList(
- long tableId,
- @Nullable Long partitionId,
- TableAssignment tableAssignment,
- @Nullable Integer leaderEpoch,
- ZooKeeperClient zkClient)
- throws Exception {
- List<BucketMetadata> bucketMetadataList = new ArrayList<>();
- // iterate each bucket assignment
- for (Map.Entry<Integer, BucketAssignment> assignment :
- tableAssignment.getBucketAssignments().entrySet()) {
- int bucketId = assignment.getKey();
- TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
-
- List<Integer> replicas = assignment.getValue().getReplicas();
-
- // now get the leader
- Optional<LeaderAndIsr> optLeaderAndIsr = zkClient.getLeaderAndIsr(tableBucket);
- Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
- bucketMetadataList.add(new BucketMetadata(bucketId, leader, leaderEpoch, replicas));
- }
- return bucketMetadataList;
- }
-
- private List<PartitionMetadata> getPartitionMetadataFromZk(
- Collection<TablePath> tablePaths, Set<Long> partitionIdSet) {
- // todo: hack logic; currently, we can't get partition metadata by partition ids directly,
- // in here, we always assume the partition ids must belong to the first argument tablePaths;
- // at least, in current client metadata request design, the assumption is true.
- // but the assumption is fragile; we should use metadata cache to help to get partition by
- // partition ids
-
- List<PartitionMetadata> partitionMetadata = new ArrayList<>();
- try {
- for (TablePath tablePath : tablePaths) {
- if (partitionIdSet.isEmpty()) {
- break;
- }
- Set<Long> hitPartitionIds = new HashSet<>();
- // TODO: this is a heavy operation, should be optimized when we have metadata cache
- Map<Long, String> partitionNameById = zkClient.getPartitionIdAndNames(tablePath);
- for (Long partitionId : partitionIdSet) {
- // the partition is under the table, get the metadata
- String partitionName = partitionNameById.get(partitionId);
- if (partitionName != null) {
- partitionMetadata.add(
- getPartitionMetadataFromZk(
- PhysicalTablePath.of(tablePath, partitionName), zkClient));
- hitPartitionIds.add(partitionId);
+ for (PhysicalTablePath partitionPath : partitions) {
+ if (authorizer == null
+ || authorizer.isAuthorized(
+ session,
+ OperationType.DESCRIBE,
+ Resource.table(
+ partitionPath.getDatabaseName(),
+ partitionPath.getTableName()))) {
+ Optional<PartitionMetadata> partitionMetadataOpt =
+ functionProvider.getPartitionMetadataFromCache(partitionPath);
+ if (!partitionMetadataOpt.isPresent()) {
+ if (cacheOnly) {
+ cacheMissMessages.add(
+ String.format(
+ "Partition metadata for partition %s is not available in local cache",
+ partitionPath));
+ }
+ // Always execute ZK query to fill cache, even for cache_only requests
+ pendingPartitionMetadata.add(
+ functionProvider
+ .getPartitionMetadataFromZk(partitionPath)
+ .thenApply(
+ meta -> {
+ partitionMetadata.add(meta);
+ return meta;
+ }));
+ } else {
+ partitionMetadata.add(partitionMetadataOpt.get());
}
}
- partitionIdSet.removeAll(hitPartitionIds);
}
- } catch (Exception e) {
- throw new FlussRuntimeException(
- "Failed to get metadata for partition ids: " + partitionIdSet, e);
- }
- if (!partitionIdSet.isEmpty()) {
- throw new PartitionNotExistException(
- "Partition not exist for partition ids: " + partitionIdSet);
- }
- return partitionMetadata;
- }
- private static AssignmentInfo getAssignmentInfo(
- @Nullable Long tableId, PhysicalTablePath physicalTablePath, ZooKeeperClient zkClient)
- throws Exception {
- // it's a partition, get the partition assignment
- if (physicalTablePath.getPartitionName() != null) {
- Optional<TablePartition> tablePartition =
- zkClient.getPartition(
- physicalTablePath.getTablePath(), physicalTablePath.getPartitionName());
- if (!tablePartition.isPresent()) {
- throw new PartitionNotExistException(
- "Table partition '" + physicalTablePath + "' does not exist.");
+ // Check for cache misses and return error immediately for cache_only requests
+ if (cacheOnly && !cacheMissMessages.isEmpty()) {
+ String combinedMessage = String.join("; ", cacheMissMessages);
+ responseFuture.completeExceptionally(new CacheMissException(combinedMessage));
+ // Continue with ZK queries to fill cache asynchronously, but don't wait for them
+ return;
}
- long partitionId = tablePartition.get().getPartitionId();
- return new AssignmentInfo(
- tablePartition.get().getTableId(),
- zkClient.getPartitionAssignment(partitionId).orElse(null),
- partitionId);
- } else {
- checkNotNull(tableId, "tableId must be not null");
- return new AssignmentInfo(
- tableId, zkClient.getTableAssignment(tableId).orElse(null), null);
- }
- }
-
- private static class AssignmentInfo {
- private final long tableId;
- // null then the bucket doesn't belong to a partition. Otherwise, not null
- private final @Nullable Long partitionId;
- private final @Nullable TableAssignment tableAssignment;
-
- private AssignmentInfo(
- long tableId, TableAssignment tableAssignment, @Nullable Long partitionId) {
- this.tableId = tableId;
- this.tableAssignment = tableAssignment;
- this.partitionId = partitionId;
+ CompletableFuture<Void> allPendingTableMeta =
+ CompletableFuture.allOf(pendingTableMetadata.toArray(new CompletableFuture[0]));
+ CompletableFuture<Void> allPendingPartitionMeta =
+ CompletableFuture.allOf(
+ pendingPartitionMetadata.toArray(new CompletableFuture[0]));
+
+ CompletableFuture<Void> allPendingRequest =
+ CompletableFuture.allOf(
+ allPendingTableMeta,
+ allPendingPartitionMeta,
+ pendingPartitionMetadataMissingIdsPartition);
+
+ allPendingRequest.handle(
+ (v, throwable) -> {
+ if (throwable != null) {
+ responseFuture.completeExceptionally(throwable);
+ return null;
+ }
+ try {
+ responseFuture.complete(
+ buildMetadataResponse(
+ metadataCache.getCoordinatorServer(listenerName),
+ new HashSet<>(
+ metadataCache
+ .getAllAliveTabletServers(listenerName)
+ .values()),
+ new ArrayList<>(tableMetadata),
+ new ArrayList<>(partitionMetadata)));
+ } catch (Exception e) {
+ responseFuture.completeExceptionally(e);
+ }
+ return null;
+ });
+ } catch (Exception e) {
+ responseFuture.completeExceptionally(e);
}
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index 6e6a2fe84..5d1b8a1e4 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -33,12 +33,10 @@ import org.apache.fluss.lake.lakestorage.LakeCatalog;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.DatabaseDescriptor;
import org.apache.fluss.metadata.PartitionSpec;
-import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
-import org.apache.fluss.metadata.TableInfo;
-import org.apache.fluss.metadata.TablePartition;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
import org.apache.fluss.rpc.messages.AdjustIsrRequest;
@@ -96,28 +94,27 @@ import org.apache.fluss.server.entity.CommitKvSnapshotData;
import org.apache.fluss.server.entity.LakeTieringTableInfo;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde;
-import org.apache.fluss.server.metadata.BucketMetadata;
-import org.apache.fluss.server.metadata.PartitionMetadata;
-import org.apache.fluss.server.metadata.ServerMetadataCache;
-import org.apache.fluss.server.metadata.TableMetadata;
+import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
+import org.apache.fluss.server.metadata.CoordinatorMetadataProvider;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.BucketAssignment;
-import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.server.zk.data.PartitionAssignment;
import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
+import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.concurrent.FutureUtils;
import javax.annotation.Nullable;
import java.io.UncheckedIOException;
-import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.function.Supplier;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclBindingFilters;
@@ -145,7 +142,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
private final boolean kvTableAllowCreation;
private final Supplier<EventManager> eventManagerSupplier;
private final Supplier<Integer> coordinatorEpochSupplier;
- private final ServerMetadataCache metadataCache;
+ private final CoordinatorMetadataCache metadataCache;
// null if the cluster hasn't configured datalake format
private final @Nullable DataLakeFormat dataLakeFormat;
@@ -157,7 +154,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
FileSystem remoteFileSystem,
ZooKeeperClient zkClient,
Supplier<CoordinatorEventProcessor> coordinatorEventProcessorSupplier,
- ServerMetadataCache metadataCache,
+ CoordinatorMetadataCache metadataCache,
MetadataManager metadataManager,
@Nullable Authorizer authorizer,
@Nullable LakeCatalog lakeCatalog,
@@ -455,17 +452,28 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
AccessContextEvent<MetadataResponse> metadataResponseAccessContextEvent =
new AccessContextEvent<>(
- ctx ->
- makeMetadataResponse(
- request,
- listenerName,
- session,
- authorizer,
- metadataCache,
- (tablePath) -> getTableMetadata(ctx, tablePath),
- ctx::getPhysicalTablePath,
- (physicalTablePath) ->
- getPartitionMetadata(ctx, physicalTablePath)));
+ ctx -> {
+ CompletableFuture<MetadataResponse> response =
+ new CompletableFuture<>();
+ processMetadataRequest(
+ request,
+ listenerName,
+ session,
+ authorizer,
+ metadataCache,
+ new CoordinatorMetadataProvider(zkClient, ctx, metadataManager),
+ response);
+ try {
+ return response.get();
+ } catch (Exception e) {
+ Throwable strippedException =
+ ExceptionUtils.stripException(e, ExecutionException.class);
+ if (strippedException instanceof RuntimeException) {
+ throw (RuntimeException) strippedException;
+ }
+ throw new RuntimeException(strippedException);
+ }
+ });
eventManagerSupplier.get().put(metadataResponseAccessContextEvent);
return metadataResponseAccessContextEvent.getResultFuture();
}
@@ -624,78 +632,6 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
}
}
- private TableMetadata getTableMetadata(CoordinatorContext ctx, TablePath tablePath) {
- // always get table info from zk.
- TableInfo tableInfo = metadataManager.getTable(tablePath);
- long tableId = ctx.getTableIdByPath(tablePath);
- List<BucketMetadata> bucketMetadataList;
- if (tableId == TableInfo.UNKNOWN_TABLE_ID) {
- // TODO no need to get assignment from zk if refactor client metadata cache. Trace by
- // https://github.com/apache/fluss/issues/483
- // get table assignment from zk.
- bucketMetadataList =
- getTableMetadataFromZk(
- zkClient, tablePath, tableInfo.getTableId(), tableInfo.isPartitioned());
- } else {
- // get table assignment from coordinatorContext.
- bucketMetadataList =
- getBucketMetadataFromContext(
- ctx, tableId, null, ctx.getTableAssignment(tableId));
- }
- return new TableMetadata(tableInfo, bucketMetadataList);
- }
-
- private PartitionMetadata getPartitionMetadata(
- CoordinatorContext ctx, PhysicalTablePath partitionPath) {
- TablePath tablePath =
- new TablePath(partitionPath.getDatabaseName(), partitionPath.getTableName());
- String partitionName = partitionPath.getPartitionName();
- long tableId = ctx.getTableIdByPath(tablePath);
- if (tableId == TableInfo.UNKNOWN_TABLE_ID) {
- // TODO no need to get assignment from zk if refactor client metadata cache. Trace by
- // https://github.com/apache/fluss/issues/483
- return getPartitionMetadataFromZk(partitionPath, zkClient);
- } else {
- Optional<Long> partitionIdOpt = ctx.getPartitionId(partitionPath);
- if (partitionIdOpt.isPresent()) {
- long partitionId = partitionIdOpt.get();
- List<BucketMetadata> bucketMetadataList =
- getBucketMetadataFromContext(
- ctx,
- tableId,
- partitionId,
- ctx.getPartitionAssignment(
- new TablePartition(tableId, partitionId)));
- return new PartitionMetadata(
- tableId, partitionName, partitionId, bucketMetadataList);
- } else {
- return getPartitionMetadataFromZk(partitionPath, zkClient);
- }
- }
- }
-
- private static List<BucketMetadata> getBucketMetadataFromContext(
- CoordinatorContext ctx,
- long tableId,
- @Nullable Long partitionId,
- Map<Integer, List<Integer>> tableAssigment) {
- List<BucketMetadata> bucketMetadataList = new ArrayList<>();
- tableAssigment.forEach(
- (bucketId, serverIds) -> {
- TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
- Optional<LeaderAndIsr> optLeaderAndIsr = ctx.getBucketLeaderAndIsr(tableBucket);
- Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
- BucketMetadata bucketMetadata =
- new BucketMetadata(
- bucketId,
- leader,
- ctx.getBucketLeaderEpoch(tableBucket),
- serverIds);
- bucketMetadataList.add(bucketMetadata);
- });
- return bucketMetadataList;
- }
-
/**
* Validates whether the table creation is allowed based on the table type and configuration.
*
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java
new file mode 100644
index 000000000..6127a4a03
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.metadata;
+
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePartition;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.coordinator.CoordinatorContext;
+import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Coordinator-side implementation of the MetadataProvider interface.
+ *
+ * <p>This provider serves metadata requests from the coordinator server, utilizing both local
+ * coordinator context cache and ZooKeeper as data sources. It provides efficient metadata access
+ * for coordinator operations such as partition assignment, leader election, and table management.
+ *
+ * <p>The provider leverages:
+ *
+ * <ul>
+ * <li>CoordinatorContext - for cached metadata and partition assignments
+ * <li>MetadataManager - for table information management
+ * <li>ZooKeeperClient - for authoritative metadata retrieval from ZK
+ * </ul>
+ *
+ * <p>This implementation is optimized for coordinator usage patterns where table and partition
+ * metadata is frequently accessed for making assignment decisions and handling client requests.
+ */
+public class CoordinatorMetadataProvider implements MetadataProvider {
+
+ private final ZooKeeperClient zkClient;
+
+ private final CoordinatorContext ctx;
+
+ private final MetadataManager metadataManager;
+
+ /**
+ * Creates a new CoordinatorMetadataProvider.
+ *
+ * @param zkClient the ZooKeeper client for accessing distributed metadata
+ * @param ctx the coordinator context providing cached metadata and assignments
+ * @param metadataManager the metadata manager for table information
+ */
+ public CoordinatorMetadataProvider(
+ ZooKeeperClient zkClient, CoordinatorContext ctx, MetadataManager metadataManager) {
+ this.zkClient = zkClient;
+ this.ctx = ctx;
+ this.metadataManager = metadataManager;
+ }
+
+ @Override
+ public Optional<TableMetadata> getTableMetadataFromCache(TablePath tablePath) {
+ long tableId = ctx.getTableIdByPath(tablePath);
+ if (tableId == TableInfo.UNKNOWN_TABLE_ID) {
+ return Optional.empty();
+ }
+ TableInfo tableInfo = ctx.getTableInfoById(tableId);
+
+ List<BucketMetadata> bucketMetadataList =
+ getBucketMetadataFromContext(ctx, tableId, null, ctx.getTableAssignment(tableId));
+ return Optional.of(new TableMetadata(tableInfo, bucketMetadataList));
+ }
+
+ @Override
+ public CompletableFuture<TableMetadata> getTableMetadataFromZk(TablePath tablePath) {
+ TableInfo tableInfo = metadataManager.getTable(tablePath);
+ return ZooKeeperClient.getTableMetadataFromZkAsync(
+ zkClient, tablePath, tableInfo.getTableId(), tableInfo.isPartitioned())
+ .thenApply(bucketMetadata -> new TableMetadata(tableInfo, bucketMetadata));
+ }
+
+ @Override
+ public Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionId) {
+ return ctx.getPhysicalTablePath(partitionId);
+ }
+
+ @Override
+ public Optional<PartitionMetadata> getPartitionMetadataFromCache(
+ PhysicalTablePath partitionPath) {
+ TablePath tablePath = partitionPath.getTablePath();
+ String partitionName = partitionPath.getPartitionName();
+ long tableId = ctx.getTableIdByPath(tablePath);
+ if (tableId == TableInfo.UNKNOWN_TABLE_ID) {
+ return Optional.empty();
+ }
+ Optional<Long> partitionIdOpt = ctx.getPartitionId(partitionPath);
+ if (!partitionIdOpt.isPresent()) {
+ return Optional.empty();
+ }
+ long partitionId = partitionIdOpt.get();
+ List<BucketMetadata> bucketMetadataList =
+ getBucketMetadataFromContext(
+ ctx,
+ tableId,
+ partitionId,
+ ctx.getPartitionAssignment(new TablePartition(tableId, partitionId)));
+ return Optional.of(
+ new PartitionMetadata(tableId, partitionName, partitionId, bucketMetadataList));
+ }
+
+ @Override
+ public CompletableFuture<PartitionMetadata> getPartitionMetadataFromZk(
+ PhysicalTablePath partitionPath) {
+ return ZooKeeperClient.getPartitionMetadataFromZkAsync(partitionPath, zkClient);
+ }
+
+ /**
+ * Constructs bucket metadata list from coordinator context information.
+ *
+ * <p>This method builds a complete list of bucket metadata by combining assignment information
+ * from the provided table assignment map with leader and epoch data from the coordinator
+ * context. Each bucket's metadata includes its ID, current leader server, leader epoch, and
+ * replica assignments.
+ *
+ * @param ctx the coordinator context containing leader and epoch information
+ * @param tableId the table identifier
+ * @param partitionId the partition identifier, null for non-partitioned tables
+ * @param tableAssigment the assignment map from bucket ID to list of replica server IDs
+ * @return a list of bucket metadata objects containing complete bucket information
+ */
+ private static List<BucketMetadata> getBucketMetadataFromContext(
+ CoordinatorContext ctx,
+ long tableId,
+ @Nullable Long partitionId,
+ Map<Integer, List<Integer>> tableAssigment) {
+ List<BucketMetadata> bucketMetadataList = new ArrayList<>();
+ tableAssigment.forEach(
+ (bucketId, serverIds) -> {
+ TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
+ Optional<LeaderAndIsr> optLeaderAndIsr = ctx.getBucketLeaderAndIsr(tableBucket);
+ Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
+ BucketMetadata bucketMetadata =
+ new BucketMetadata(
+ bucketId,
+ leader,
+ ctx.getBucketLeaderEpoch(tableBucket),
+ serverIds);
+ bucketMetadataList.add(bucketMetadata);
+ });
+ return bucketMetadataList;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java
new file mode 100644
index 000000000..12c25773e
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.metadata;
+
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TablePath;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Interface for providing metadata information about tables and partitions from various sources.
+ * This provider supports retrieving metadata from both local cache and ZooKeeper, allowing for
+ * flexible metadata access patterns with different consistency and performance characteristics.
+ */
+public interface MetadataProvider {
+
+ /**
+ * Retrieves table metadata from local cache.
+ *
+ * @param tablePath the path identifying the table
+ * @return an Optional containing the table metadata if found in cache, empty otherwise
+ */
+ Optional<TableMetadata> getTableMetadataFromCache(TablePath tablePath);
+
+ /**
+ * Retrieves table metadata from ZooKeeper asynchronously.
+ *
+ * @param tablePath the path identifying the table
+ * @return a CompletableFuture containing the table metadata from ZooKeeper
+ */
+ CompletableFuture<TableMetadata> getTableMetadataFromZk(TablePath tablePath);
+
+ /**
+ * Retrieves the physical table path from local cache using partition ID.
+ *
+ * @param partitionId the partition identifier
+ * @return an Optional containing the physical table path if found in cache, empty otherwise
+ */
+ Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionId);
+
+ /**
+ * Retrieves partition metadata from local cache.
+ *
+ * @param physicalTablePath the physical path identifying the table partition
+ * @return an Optional containing the partition metadata if found in cache, empty otherwise
+ */
+ Optional<PartitionMetadata> getPartitionMetadataFromCache(PhysicalTablePath physicalTablePath);
+
+ /**
+ * Retrieves partition metadata from ZooKeeper asynchronously.
+ *
+ * @param physicalTablePath the physical path identifying the table partition
+ * @return a CompletableFuture containing the partition metadata from ZooKeeper
+ */
+ CompletableFuture<PartitionMetadata> getPartitionMetadataFromZk(
+ PhysicalTablePath physicalTablePath);
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
index ff6855e13..5479b523a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
@@ -42,8 +42,6 @@ import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import static org.apache.fluss.server.RpcServiceBase.getPartitionMetadataFromZk;
-import static org.apache.fluss.server.RpcServiceBase.getTableMetadataFromZk;
import static org.apache.fluss.server.metadata.PartitionMetadata.DELETED_PARTITION_ID;
import static org.apache.fluss.server.metadata.PartitionMetadata.DELETED_PARTITION_NAME;
import static org.apache.fluss.server.metadata.TableMetadata.DELETED_TABLE_ID;
@@ -115,32 +113,31 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
return serverMetadataSnapshot.getPhysicalTablePath(partitionId);
}
- public TableMetadata getTableMetadata(TablePath tablePath) {
- // always get table info from zk.
- TableInfo tableInfo = metadataManager.getTable(tablePath);
+ public Optional<TableMetadata> getTableMetadata(TablePath tablePath) {
+ // Only get data from cache, do not access ZK.
ServerMetadataSnapshot snapshot = serverMetadataSnapshot;
OptionalLong tableIdOpt = snapshot.getTableId(tablePath);
- List<BucketMetadata> bucketMetadataList;
if (!tableIdOpt.isPresent()) {
- // TODO no need to get assignment from zk if refactor client metadata cache. Trace by
- // https://github.com/apache/fluss/issues/483
- // get table assignment from zk.
- bucketMetadataList =
- getTableMetadataFromZk(
- zkClient, tablePath, tableInfo.getTableId(), tableInfo.isPartitioned());
- } else {
- // get table assignment from cache.
- bucketMetadataList =
- new ArrayList<>(
- snapshot.getBucketMetadataForTable(tableIdOpt.getAsLong()).values());
+ return Optional.empty();
}
- return new TableMetadata(tableInfo, bucketMetadataList);
+ long tableId = tableIdOpt.getAsLong();
+
+ // Try to get table info from ZK only if we have the table ID in cache
+ try {
+ TableInfo tableInfo = metadataManager.getTable(tablePath);
+ List<BucketMetadata> bucketMetadataList =
+ new ArrayList<>(snapshot.getBucketMetadataForTable(tableId).values());
+ return Optional.of(new TableMetadata(tableInfo, bucketMetadataList));
+ } catch (Exception e) {
+ // If table doesn't exist in ZK but exists in cache, return empty
+ // This maintains backward compatibility while fixing the semantic issue
+ return Optional.empty();
+ }
}
- public PartitionMetadata getPartitionMetadata(PhysicalTablePath partitionPath) {
- TablePath tablePath =
- new TablePath(partitionPath.getDatabaseName(), partitionPath.getTableName());
+ public Optional<PartitionMetadata> getPartitionMetadata(PhysicalTablePath partitionPath) {
+ TablePath tablePath = partitionPath.getTablePath();
String partitionName = partitionPath.getPartitionName();
ServerMetadataSnapshot snapshot = serverMetadataSnapshot;
@@ -149,15 +146,16 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
if (tableIdOpt.isPresent() && partitionIdOpt.isPresent()) {
long tableId = tableIdOpt.getAsLong();
long partitionId = partitionIdOpt.get();
- return new PartitionMetadata(
- tableId,
- partitionName,
- partitionId,
- new ArrayList<>(snapshot.getBucketMetadataForPartition(partitionId).values()));
+ return Optional.of(
+ new PartitionMetadata(
+ tableId,
+ partitionName,
+ partitionId,
+ new ArrayList<>(
+ snapshot.getBucketMetadataForPartition(partitionId).values())));
} else {
- // TODO no need to get assignment from zk if refactor client metadata cache. Trace by
- // https://github.com/apache/fluss/issues/483
- return getPartitionMetadataFromZk(partitionPath, zkClient);
+
+ return Optional.empty();
}
}
@@ -291,4 +289,128 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
Collections.emptyMap());
});
}
+
+ /**
+ * Update a single table metadata to the local cache. This method is thread-safe and will merge
+ * the new table metadata with existing cache.
+ *
+ * @param tableMetadata the table metadata to update
+ */
+ public void updateTableMetadata(TableMetadata tableMetadata) {
+ inLock(
+ metadataLock,
+ () -> {
+ TableInfo tableInfo = tableMetadata.getTableInfo();
+ TablePath tablePath = tableInfo.getTablePath();
+ long tableId = tableInfo.getTableId();
+
+ // Get current snapshot
+ ServerMetadataSnapshot currentSnapshot = serverMetadataSnapshot;
+
+ // Create new maps based on current state
+ Map<TablePath, Long> tableIdByPath =
+ new HashMap<>(currentSnapshot.getTableIdByPath());
+ Map<Long, TablePath> pathByTableId = new HashMap<>();
+ Map<Long, Map<Integer, BucketMetadata>> bucketMetadataMapForTables =
+ new HashMap<>(currentSnapshot.getBucketMetadataMapForTables());
+
+ // Update table mapping
+ tableIdByPath.put(tablePath, tableId);
+ pathByTableId.put(tableId, tablePath);
+
+ // Update bucket metadata for this table
+ Map<Integer, BucketMetadata> tableBucketMetadata = new HashMap<>();
+ for (BucketMetadata bucketMetadata : tableMetadata.getBucketMetadataList()) {
+ tableBucketMetadata.put(bucketMetadata.getBucketId(), bucketMetadata);
+ }
+ bucketMetadataMapForTables.put(tableId, tableBucketMetadata);
+
+ // Copy other existing data
+ Map<PhysicalTablePath, Long> partitionIdByPath =
+ new HashMap<>(currentSnapshot.getPartitionIdByPath());
+ Map<Long, Map<Integer, BucketMetadata>> bucketMetadataMapForPartitions =
+ new HashMap<>(currentSnapshot.getBucketMetadataMapForPartitions());
+
+ // Build pathByTableId from tableIdByPath
+ tableIdByPath.forEach((path, id) -> pathByTableId.put(id, path));
+
+ // Create new snapshot
+ serverMetadataSnapshot =
+ new ServerMetadataSnapshot(
+ currentSnapshot.getCoordinatorServer(),
+ currentSnapshot.getAliveTabletServers(),
+ tableIdByPath,
+ pathByTableId,
+ partitionIdByPath,
+ bucketMetadataMapForTables,
+ bucketMetadataMapForPartitions);
+ });
+ }
+
+ /**
+ * Update a single partition metadata to the local cache. This method is thread-safe and will
+ * merge the new partition metadata with existing cache.
+ *
+ * @param partitionMetadata the partition metadata to update
+ */
+ public void updatePartitionMetadata(PartitionMetadata partitionMetadata) {
+ inLock(
+ metadataLock,
+ () -> {
+ long tableId = partitionMetadata.getTableId();
+ String partitionName = partitionMetadata.getPartitionName();
+ long partitionId = partitionMetadata.getPartitionId();
+
+ // Get current snapshot
+ ServerMetadataSnapshot currentSnapshot = serverMetadataSnapshot;
+
+ // Get table path from tableId
+ Optional<TablePath> tablePathOpt = currentSnapshot.getTablePath(tableId);
+ if (!tablePathOpt.isPresent()) {
+ // If table doesn't exist in cache, we can't update partition metadata
+ return;
+ }
+ TablePath tablePath = tablePathOpt.get();
+ PhysicalTablePath physicalTablePath =
+ PhysicalTablePath.of(tablePath, partitionName);
+
+ // Create new maps based on current state
+ Map<TablePath, Long> tableIdByPath =
+ new HashMap<>(currentSnapshot.getTableIdByPath());
+ Map<Long, TablePath> pathByTableId = new HashMap<>();
+ Map<PhysicalTablePath, Long> partitionIdByPath =
+ new HashMap<>(currentSnapshot.getPartitionIdByPath());
+ Map<Long, Map<Integer, BucketMetadata>> bucketMetadataMapForPartitions =
+ new HashMap<>(currentSnapshot.getBucketMetadataMapForPartitions());
+
+ // Update partition mapping
+ partitionIdByPath.put(physicalTablePath, partitionId);
+
+ // Update bucket metadata for this partition
+ Map<Integer, BucketMetadata> partitionBucketMetadata = new HashMap<>();
+ for (BucketMetadata bucketMetadata :
+ partitionMetadata.getBucketMetadataList()) {
+ partitionBucketMetadata.put(bucketMetadata.getBucketId(), bucketMetadata);
+ }
+ bucketMetadataMapForPartitions.put(partitionId, partitionBucketMetadata);
+
+ // Copy other existing data
+ Map<Long, Map<Integer, BucketMetadata>> bucketMetadataMapForTables =
+ new HashMap<>(currentSnapshot.getBucketMetadataMapForTables());
+
+ // Build pathByTableId from tableIdByPath
+ tableIdByPath.forEach((path, id) -> pathByTableId.put(id, path));
+
+ // Create new snapshot
+ serverMetadataSnapshot =
+ new ServerMetadataSnapshot(
+ currentSnapshot.getCoordinatorServer(),
+ currentSnapshot.getAliveTabletServers(),
+ tableIdByPath,
+ pathByTableId,
+ partitionIdByPath,
+ bucketMetadataMapForTables,
+ bucketMetadataMapForPartitions);
+ });
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java
new file mode 100644
index 000000000..b52269ea2
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.metadata;
+
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Tablet server-side implementation of the MetadataProvider interface.
+ *
+ * <p>This provider serves metadata requests from tablet servers, utilizing a local metadata cache
+ * for efficient access and ZooKeeper as the authoritative data source. It implements a cache-aside
+ * pattern where metadata is first checked in the local cache, and if not found, retrieved from
+ * ZooKeeper and then cached locally for future access.
+ *
+ * <p>Key characteristics:
+ *
+ * <ul>
+ * <li>Local caching - Uses {@link TabletServerMetadataCache} for fast metadata access
+ * <li>Cache updates - Automatically updates local cache when fetching from ZooKeeper
+ * <li>Optimized for tablet server workloads - Frequent metadata lookups with good locality
+ * </ul>
+ *
+ * <p>This implementation is particularly suited for tablet servers that need to frequently access
+ * metadata for partition management, leader election participation, and client request handling,
+ * while maintaining consistency with the coordinator's view of metadata.
+ */
+public class TabletServerMetadataProvider implements MetadataProvider {
+
+ private final ZooKeeperClient zkClient;
+
+ private final TabletServerMetadataCache metadataCache;
+
+ private final MetadataManager metadataManager;
+
+ /**
+ * Creates a new TabletServerMetadataProvider.
+ *
+ * @param zkClient the ZooKeeper client for accessing distributed metadata
+ * @param metadataCache the local metadata cache for efficient metadata access
+ * @param metadataManager the metadata manager for table information
+ */
+ public TabletServerMetadataProvider(
+ ZooKeeperClient zkClient,
+ TabletServerMetadataCache metadataCache,
+ MetadataManager metadataManager) {
+ this.zkClient = zkClient;
+ this.metadataCache = metadataCache;
+ this.metadataManager = metadataManager;
+ }
+
+ @Override
+ public Optional<TableMetadata> getTableMetadataFromCache(TablePath tablePath) {
+ return metadataCache.getTableMetadata(tablePath);
+ }
+
+ @Override
+ public CompletableFuture<TableMetadata> getTableMetadataFromZk(TablePath tablePath) {
+ TableInfo tableInfo = metadataManager.getTable(tablePath);
+ return ZooKeeperClient.getTableMetadataFromZkAsync(
+ zkClient, tablePath, tableInfo.getTableId(), tableInfo.isPartitioned())
+ .thenApply(
+ bucketMetadataList -> {
+ TableMetadata tableMetadata =
+ new TableMetadata(tableInfo, bucketMetadataList);
+ // Update local cache after successfully fetching from ZooKeeper
+ metadataCache.updateTableMetadata(tableMetadata);
+ return tableMetadata;
+ });
+ }
+
+ @Override
+ public Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionId) {
+ return metadataCache.getPhysicalTablePath(partitionId);
+ }
+
+ @Override
+ public Optional<PartitionMetadata> getPartitionMetadataFromCache(
+ PhysicalTablePath physicalTablePath) {
+ return metadataCache.getPartitionMetadata(physicalTablePath);
+ }
+
+ @Override
+ public CompletableFuture<PartitionMetadata> getPartitionMetadataFromZk(
+ PhysicalTablePath physicalTablePath) {
+ return ZooKeeperClient.getPartitionMetadataFromZkAsync(physicalTablePath, zkClient)
+ .thenApply(
+ partitionMetadata -> {
+ // Update local cache after successfully fetching from ZooKeeper
+ metadataCache.updatePartitionMetadata(partitionMetadata);
+ return partitionMetadata;
+ });
+ }
+}
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 3bf13aa02..4114b1ab4 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
@@ -72,6 +72,7 @@ import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.log.FetchParams;
import org.apache.fluss.server.log.ListOffsetsParam;
import org.apache.fluss.server.metadata.TabletServerMetadataCache;
+import org.apache.fluss.server.metadata.TabletServerMetadataProvider;
import org.apache.fluss.server.replica.ReplicaManager;
import org.apache.fluss.server.utils.ServerRpcMessageUtils;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -122,6 +123,7 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
private final String serviceName;
private final ReplicaManager replicaManager;
private final TabletServerMetadataCache metadataCache;
+ private final TabletServerMetadataProvider metadataFunctionProvider;
public TabletService(
int serverId,
@@ -135,6 +137,8 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
this.serviceName = "server-" + serverId;
this.replicaManager = replicaManager;
this.metadataCache = metadataCache;
+ this.metadataFunctionProvider =
+ new TabletServerMetadataProvider(zkClient, metadataCache, metadataManager);
}
@Override
@@ -282,16 +286,16 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
@Override
public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) {
- return CompletableFuture.completedFuture(
- makeMetadataResponse(
- request,
- currentListenerName(),
- currentSession(),
- authorizer,
- metadataCache,
- metadataCache::getTableMetadata,
- metadataCache::getPhysicalTablePath,
- metadataCache::getPartitionMetadata));
+ CompletableFuture<MetadataResponse> response = new CompletableFuture<>();
+ processMetadataRequest(
+ request,
+ currentListenerName(),
+ currentSession(),
+ authorizer,
+ metadataCache,
+ metadataFunctionProvider,
+ response);
+ return response;
}
@Override
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 7173d8cdf..872b06d7b 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
@@ -20,6 +20,10 @@ package org.apache.fluss.server.zk;
import org.apache.fluss.annotation.Internal;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.PartitionNotExistException;
+import org.apache.fluss.exception.TableNotExistException;
+import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaInfo;
@@ -31,10 +35,15 @@ import org.apache.fluss.security.acl.Resource;
import org.apache.fluss.security.acl.ResourceType;
import org.apache.fluss.server.authorizer.DefaultAuthorizer.VersionedAcls;
import org.apache.fluss.server.entity.RegisterTableBucketLeadAndIsrInfo;
+import org.apache.fluss.server.metadata.BucketMetadata;
+import org.apache.fluss.server.metadata.PartitionMetadata;
import org.apache.fluss.server.zk.ZkAsyncRequest.ZkGetChildrenRequest;
import org.apache.fluss.server.zk.ZkAsyncRequest.ZkGetDataRequest;
import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetChildrenResponse;
import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetDataResponse;
+import org.apache.fluss.server.zk.ZooKeeperMetadataUtils.AssignmentInfo;
+import org.apache.fluss.server.zk.ZooKeeperMetadataUtils.PartitionPathInfo;
+import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.BucketSnapshot;
import org.apache.fluss.server.zk.data.CoordinatorAddress;
import org.apache.fluss.server.zk.data.DatabaseRegistration;
@@ -78,11 +87,14 @@ import org.apache.fluss.shaded.curator5.org.apache.curator.framework.api.transac
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.CreateMode;
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.data.Stat;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.concurrent.FutureUtils;
import org.apache.fluss.utils.types.Tuple2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.ArrayList;
@@ -97,12 +109,15 @@ import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.fluss.metadata.ResolvedPartitionSpec.fromPartitionName;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
/**
* This class includes methods for write/read various metadata (leader address, tablet server
@@ -114,6 +129,7 @@ public class ZooKeeperClient implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperClient.class);
public static final int UNKNOWN_VERSION = -2;
private static final int MAX_BATCH_SIZE = 1024;
+
private final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
private final CuratorFramework zkClient;
@@ -147,54 +163,107 @@ public class ZooKeeperClient implements AutoCloseable {
}
/**
- * Send a pipelined sequence of requests and wait for all of their responses.
+ * Send a pipelined sequence of requests and return a CompletableFuture for all their responses.
*
* <p>The watch flag on each outgoing request will be set if we've already registered a handler
* for the path associated with the request.
*
- * @param requests a sequence of requests to send and wait on.
- * @return the responses for the requests. If all requests have the same type, the responses
- * will have the respective response type.
+ * @param requests a sequence of requests to send
+ * @param respCreator function to create response objects from curator events
+ * @return CompletableFuture containing the responses for the requests
*/
private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
- List<Resp> handleRequestAsync(
- List<Req> requests, Function<CuratorEvent, Resp> respCreator) throws Exception {
+ CompletableFuture<List<Resp>> handleRequestInBackgroundAsync(
+ List<Req> requests, Function<CuratorEvent, Resp> respCreator) {
if (requests == null || requests.isEmpty()) {
- return Collections.emptyList();
+ return CompletableFuture.completedFuture(Collections.emptyList());
}
- ArrayBlockingQueue<Resp> responseQueue = new ArrayBlockingQueue<>(requests.size());
- CountDownLatch countDownLatch = new CountDownLatch(requests.size());
+ CompletableFuture<List<Resp>> future = new CompletableFuture<>();
+ BackgroundCallback callback = getZkBackgroundCallback(requests, respCreator, future);
- BackgroundCallback callback =
- (client, event) -> {
- Resp response = respCreator.apply(event);
- responseQueue.add(response);
+ try {
+ for (Req request : requests) {
+ try {
+ inFlightRequests.acquire();
+ if (request instanceof ZkGetDataRequest) {
+ zkClient.getData().inBackground(callback).forPath(request.getPath());
+
+ } else if (request instanceof ZkGetChildrenRequest) {
+ zkClient.getChildren().inBackground(callback).forPath(request.getPath());
+
+ } else {
+ inFlightRequests.release();
+ throw new IllegalArgumentException(
+ "Unsupported request type: " + request.getClass());
+ }
+ } catch (Exception e) {
inFlightRequests.release();
- countDownLatch.countDown();
- };
+ throw e;
+ }
+ }
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
- for (Req request : requests) {
- try {
- inFlightRequests.acquire();
- if (request instanceof ZkGetDataRequest) {
- zkClient.getData().inBackground(callback).forPath(request.getPath());
+ return future;
+ }
+
+ @Nonnull
+ private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
+ BackgroundCallback getZkBackgroundCallback(
+ List<Req> requests,
+ Function<CuratorEvent, Resp> respCreator,
+ CompletableFuture<List<Resp>> future) {
+ ArrayBlockingQueue<Resp> responseQueue = new ArrayBlockingQueue<>(requests.size());
+ CountDownLatch countDownLatch = new CountDownLatch(requests.size());
- } else if (request instanceof ZkGetChildrenRequest) {
- zkClient.getChildren().inBackground(callback).forPath(request.getPath());
+ // Complete future when all responses received
+ return (client, event) -> {
+ try {
+ Resp response = respCreator.apply(event);
+ responseQueue.add(response);
+ inFlightRequests.release();
+ countDownLatch.countDown();
- } else {
- throw new IllegalArgumentException(
- "Unsupported request type: " + request.getClass());
+ // Complete future when all responses received
+ if (countDownLatch.getCount() == 0) {
+ future.complete(new ArrayList<>(responseQueue));
}
} catch (Exception e) {
inFlightRequests.release();
- throw e;
+ future.completeExceptionally(e);
}
- }
+ };
+ }
- countDownLatch.await();
- return new ArrayList<>(responseQueue);
+ /**
+ * Send a pipelined sequence of requests and wait for all of their responses synchronously in
+ * background.
+ *
+ * <p>The watch flag on each outgoing request will be set if we've already registered a handler
+ * for the path associated with the request.
+ *
+ * @param requests a sequence of requests to send and wait on.
+ * @return the responses for the requests. If all requests have the same type, the responses
+ * will have the respective response type.
+ */
+ private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
+ List<Resp> handleRequestInBackground(
+ List<Req> requests, Function<CuratorEvent, Resp> respCreator) throws Exception {
+ try {
+ return handleRequestInBackgroundAsync(requests, respCreator).get();
+ } catch (ExecutionException e) {
+ Throwable cause = ExceptionUtils.stripExecutionException(e);
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ } else {
+ throw new Exception("Async request handling failed", cause);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new Exception("Request handling was interrupted", e);
+ }
}
// --------------------------------------------------------------------------------------------
@@ -251,22 +320,12 @@ public class ZooKeeperClient implements AutoCloseable {
.boxed()
.collect(Collectors.toMap(ServerIdZNode::path, id -> id));
- List<ZkGetDataResponse> responses = getDataAsync(path2IdMap.keySet());
- // tablet server id -> TabletServerRegistration
- Map<Integer, TabletServerRegistration> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- result.put(
- path2IdMap.get(response.getPath()),
- ServerIdZNode.decode(response.getData()));
- } else {
- LOG.warn(
- "Failed to get data for path {}: {}",
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
+ List<ZkGetDataResponse> responses = getDataInBackground(path2IdMap.keySet());
+ return ZooKeeperMetadataUtils.processMultipleDataResponses(
+ responses,
+ response -> path2IdMap.get(response.getPath()),
+ ServerIdZNode::decode,
+ "tablet server registration");
}
/** Gets the list of sorted server Ids. */
@@ -292,13 +351,20 @@ public class ZooKeeperClient implements AutoCloseable {
/** Get the table assignment in ZK. */
public Optional<TableAssignment> getTableAssignment(long tableId) throws Exception {
- Optional<byte[]> bytes = getOrEmpty(TableIdZNode.path(tableId));
- return bytes.map(
- data ->
- // we'll put a laketable node under TableIdZNode,
- // so it won't be Optional#empty
- // but will with a zero-length array
- data.length == 0 ? null : TableIdZNode.decode(data));
+ return FutureUtils.asyncToSync(() -> getTableAssignmentAsync(tableId), "table assignment");
+ }
+
+ /** Get the table assignment in ZK asynchronously. */
+ public CompletableFuture<TableAssignment> getTableAssignmentAsync(long tableId) {
+ String path = TableIdZNode.path(tableId);
+
+ return getDataInBackgroundAsync(Collections.singleton(path))
+ .thenApply(
+ responses ->
+ ZooKeeperMetadataUtils.processSingleDataResponse(
+ responses,
+ data -> data.length == 0 ? null : TableIdZNode.decode(data),
+ path));
}
/** Get the tables assignments in ZK. */
@@ -306,31 +372,37 @@ public class ZooKeeperClient implements AutoCloseable {
Map<String, Long> path2TableIdMap =
tableIds.stream().collect(Collectors.toMap(TableIdZNode::path, id -> id));
- List<ZkGetDataResponse> responses = getDataAsync(path2TableIdMap.keySet());
- // tabletId -> TableAssignment
- Map<Long, TableAssignment> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK
- && response.getData().length > 0) {
- result.put(
- path2TableIdMap.get(response.getPath()),
- TableIdZNode.decode(response.getData()));
- } else {
- LOG.warn(
- "Failed to get data for path {}: {}, data length = {}",
- response.getPath(),
- response.getResultCode(),
- response.getData() == null ? 0 : response.getData().length);
- }
- }
-
- return result;
+ List<ZkGetDataResponse> responses = getDataInBackground(path2TableIdMap.keySet());
+ return ZooKeeperMetadataUtils.processMultipleDataResponsesWithCondition(
+ responses,
+ response -> path2TableIdMap.get(response.getPath()),
+ TableIdZNode::decode,
+ response -> response.getData().length > 0,
+ "table assignment");
}
/** Get the partition assignment in ZK. */
public Optional<PartitionAssignment> getPartitionAssignment(long partitionId) throws Exception {
- Optional<byte[]> bytes = getOrEmpty(PartitionIdZNode.path(partitionId));
- return bytes.map(PartitionIdZNode::decode);
+ return FutureUtils.asyncToSync(
+ () -> getPartitionAssignmentAsync(partitionId),
+ assignment ->
+ (PartitionAssignment)
+ assignment, // PartitionAssignment extends TableAssignment
+ "partition assignment");
+ }
+
+ /** Get the partition assignment in ZK asynchronously. */
+ private CompletableFuture<TableAssignment> getPartitionAssignmentAsync(long partitionId) {
+ String path = PartitionIdZNode.path(partitionId);
+
+ return getDataInBackgroundAsync(Collections.singleton(path))
+ .thenApply(
+ responses ->
+ ZooKeeperMetadataUtils.processSingleDataResponse(
+ responses,
+ PartitionIdZNode::decode, // PartitionAssignment extends
+ // TableAssignment
+ path));
}
/** Get the partitions assignments in ZK. */
@@ -339,22 +411,12 @@ public class ZooKeeperClient implements AutoCloseable {
Map<String, Long> path2PartitionIdMap =
partitionIds.stream().collect(Collectors.toMap(PartitionIdZNode::path, id -> id));
- List<ZkGetDataResponse> responses = getDataAsync(path2PartitionIdMap.keySet());
- // tabletId -> PartitionAssignment
- Map<Long, PartitionAssignment> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- result.put(
- path2PartitionIdMap.get(response.getPath()),
- PartitionIdZNode.decode(response.getData()));
- } else {
- LOG.warn(
- "Failed to get data for path {}: {}",
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
+ List<ZkGetDataResponse> responses = getDataInBackground(path2PartitionIdMap.keySet());
+ return ZooKeeperMetadataUtils.processMultipleDataResponses(
+ responses,
+ response -> path2PartitionIdMap.get(response.getPath()),
+ PartitionIdZNode::decode,
+ "partition assignment");
}
public void updateTableAssignment(long tableId, TableAssignment tableAssignment)
@@ -448,6 +510,27 @@ public class ZooKeeperClient implements AutoCloseable {
return bytes.map(LeaderAndIsrZNode::decode);
}
+ /** Get the buckets LeaderAndIsr in ZK asynchronously. */
+ private CompletableFuture<Map<TableBucket, Optional<LeaderAndIsr>>> getLeaderAndIsrsAsync(
+ Collection<TableBucket> tableBuckets) {
+ if (tableBuckets.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.emptyMap());
+ }
+
+ Map<String, TableBucket> path2TableBucketMap =
+ tableBuckets.stream()
+ .collect(Collectors.toMap(LeaderAndIsrZNode::path, bucket -> bucket));
+
+ return getDataInBackgroundAsync(path2TableBucketMap.keySet())
+ .thenApply(
+ responses ->
+ ZooKeeperMetadataUtils.processMultipleDataResponsesWithOptional(
+ responses,
+ response -> path2TableBucketMap.get(response.getPath()),
+ LeaderAndIsrZNode::decode,
+ "leader and isr"));
+ }
+
/** Get the buckets LeaderAndIsr in ZK. */
public Map<TableBucket, LeaderAndIsr> getLeaderAndIsrs(Collection<TableBucket> tableBuckets)
throws Exception {
@@ -455,22 +538,12 @@ public class ZooKeeperClient implements AutoCloseable {
tableBuckets.stream()
.collect(Collectors.toMap(LeaderAndIsrZNode::path, bucket -> bucket));
- List<ZkGetDataResponse> responses = getDataAsync(path2TableBucketMap.keySet());
- // TableBucket -> LeaderAndIsr
- Map<TableBucket, LeaderAndIsr> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- result.put(
- path2TableBucketMap.get(response.getPath()),
- LeaderAndIsrZNode.decode(response.getData()));
- } else {
- LOG.warn(
- "Failed to get data for path {}: {}",
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
+ List<ZkGetDataResponse> responses = getDataInBackground(path2TableBucketMap.keySet());
+ return ZooKeeperMetadataUtils.processMultipleDataResponses(
+ responses,
+ response -> path2TableBucketMap.get(response.getPath()),
+ LeaderAndIsrZNode::decode,
+ "leader and isr");
}
public void updateLeaderAndIsr(TableBucket tableBucket, LeaderAndIsr leaderAndIsr)
@@ -605,25 +678,23 @@ public class ZooKeeperClient implements AutoCloseable {
/** Get the tables in ZK. */
public Map<TablePath, TableRegistration> getTables(Collection<TablePath> tablePaths)
throws Exception {
+ return FutureUtils.asyncToSyncDirect(() -> getTablesAsync(tablePaths), "tables");
+ }
+
+ /** Get the tables in ZK asynchronously. */
+ public CompletableFuture<Map<TablePath, TableRegistration>> getTablesAsync(
+ Collection<TablePath> tablePaths) {
Map<String, TablePath> path2TablePathMap =
tablePaths.stream().collect(Collectors.toMap(TableZNode::path, path -> path));
- List<ZkGetDataResponse> responses = getDataAsync(path2TablePathMap.keySet());
- // TablePath -> TableRegistration
- Map<TablePath, TableRegistration> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- result.put(
- path2TablePathMap.get(response.getPath()),
- TableZNode.decode(response.getData()));
- } else {
- LOG.warn(
- "Failed to get data for path {}: {}",
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
+ return getDataInBackgroundAsync(path2TablePathMap.keySet())
+ .thenApply(
+ responses ->
+ ZooKeeperMetadataUtils.processMultipleDataResponses(
+ responses,
+ response -> path2TablePathMap.get(response.getPath()),
+ TableZNode::decode,
+ "table registration"));
}
/** Update the table in ZK. */
@@ -665,19 +736,11 @@ public class ZooKeeperClient implements AutoCloseable {
Map<String, TablePath> path2TablePathMap =
tablePaths.stream().collect(Collectors.toMap(PartitionsZNode::path, path -> path));
- List<ZkGetChildrenResponse> responses = getChildrenAsync(path2TablePathMap.keySet());
- Map<TablePath, List<String>> result = new HashMap<>();
- for (ZkGetChildrenResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- result.put(path2TablePathMap.get(response.getPath()), response.getChildren());
- } else {
- LOG.warn(
- "Failed to get children for path {}: {}",
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
+ List<ZkGetChildrenResponse> responses = getChildrenInBackground(path2TablePathMap.keySet());
+ return ZooKeeperMetadataUtils.processMultipleChildrenResponses(
+ responses,
+ response -> path2TablePathMap.get(response.getPath()),
+ "partitions for tables");
}
/** Get the partition and the id for the partitions of a table in ZK. */
@@ -711,7 +774,7 @@ public class ZooKeeperClient implements AutoCloseable {
}
}
- List<ZkGetDataResponse> responses = getDataAsync(zkPath2TablePath.keySet());
+ List<ZkGetDataResponse> responses = getDataInBackground(zkPath2TablePath.keySet());
for (ZkGetDataResponse response : responses) {
if (response.getResultCode() == KeeperException.Code.OK) {
String zkPath = response.getPath();
@@ -767,8 +830,20 @@ public class ZooKeeperClient implements AutoCloseable {
/** Get a partition of a table in ZK. */
public Optional<TablePartition> getPartition(TablePath tablePath, String partitionName)
throws Exception {
+ return FutureUtils.asyncToSync(
+ () -> getPartitionAsync(tablePath, partitionName), "partition");
+ }
+
+ /** Get a partition of a table in ZK asynchronously. */
+ public CompletableFuture<TablePartition> getPartitionAsync(
+ TablePath tablePath, String partitionName) {
String path = PartitionZNode.path(tablePath, partitionName);
- return getOrEmpty(path).map(PartitionZNode::decode);
+
+ return getDataInBackgroundAsync(Collections.singleton(path))
+ .thenApply(
+ responses ->
+ ZooKeeperMetadataUtils.processSingleDataResponse(
+ responses, PartitionZNode::decode, path));
}
/** Get partition num of a table in ZK. */
@@ -1191,29 +1266,56 @@ public class ZooKeeperClient implements AutoCloseable {
}
/**
- * Gets the child nodes at given zk node paths asynchronously.
+ * Gets the child nodes at given zk node paths in background asynchronously.
+ *
+ * @param paths the paths to list children
+ * @return CompletableFuture containing list of async responses for each path
+ */
+ public CompletableFuture<List<ZkGetChildrenResponse>> getChildrenInBackgroundAsync(
+ Collection<String> paths) {
+ List<ZkGetChildrenRequest> requests =
+ paths.stream().map(ZkGetChildrenRequest::new).collect(Collectors.toList());
+ return handleRequestInBackgroundAsync(requests, ZkGetChildrenResponse::create);
+ }
+
+ /**
+ * Gets the child nodes at given zk node paths in background.
*
* @param paths the paths to list children
* @return list of async responses for each path
* @throws Exception if there is an error during the operation
*/
- public List<ZkGetChildrenResponse> getChildrenAsync(Collection<String> paths) throws Exception {
+ public List<ZkGetChildrenResponse> getChildrenInBackground(Collection<String> paths)
+ throws Exception {
List<ZkGetChildrenRequest> requests =
paths.stream().map(ZkGetChildrenRequest::new).collect(Collectors.toList());
- return handleRequestAsync(requests, ZkGetChildrenResponse::create);
+ return handleRequestInBackground(requests, ZkGetChildrenResponse::create);
+ }
+
+ /**
+ * Gets the data of given zk node paths in background asynchronously.
+ *
+ * @param paths the paths to fetch data
+ * @return CompletableFuture containing list of async responses for each path
+ */
+ public CompletableFuture<List<ZkGetDataResponse>> getDataInBackgroundAsync(
+ Collection<String> paths) {
+ List<ZkGetDataRequest> requests =
+ paths.stream().map(ZkGetDataRequest::new).collect(Collectors.toList());
+ return handleRequestInBackgroundAsync(requests, ZkGetDataResponse::create);
}
/**
- * Gets the data of given zk node paths asynchronously.
+ * Gets the data of given zk node paths in background.
*
* @param paths the paths to fetch data
* @return list of async responses for each path
* @throws Exception if there is an error during the operation
*/
- public List<ZkGetDataResponse> getDataAsync(Collection<String> paths) throws Exception {
+ public List<ZkGetDataResponse> getDataInBackground(Collection<String> paths) throws Exception {
List<ZkGetDataRequest> requests =
paths.stream().map(ZkGetDataRequest::new).collect(Collectors.toList());
- return handleRequestAsync(requests, ZkGetDataResponse::create);
+ return handleRequestInBackground(requests, ZkGetDataResponse::create);
}
/** Gets the data and stat of a given zk node path. */
@@ -1238,6 +1340,433 @@ public class ZooKeeperClient implements AutoCloseable {
return zkClient;
}
+ // --------------------------------------------------------------------------------------------
+ // Table and Partition Metadata
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Asynchronously retrieves table metadata from ZooKeeper for the specified table.
+ *
+ * <p>This method fetches the table assignment information from ZooKeeper and converts it into a
+ * list of bucket metadata. The operation is performed asynchronously to avoid blocking the
+ * calling thread.
+ *
+ * @param tablePath the path of the table to retrieve metadata for
+ * @param tableId the unique identifier of the table
+ * @param isPartitioned whether the table is partitioned (affects warning behavior for missing
+ * assignments)
+ * @return a CompletableFuture containing a list of BucketMetadata objects representing all
+ * buckets for this table. Returns an empty list if no table assignment is found.
+ * @throws TableNotExistException if the table does not exist in ZooKeeper
+ * @throws FlussRuntimeException if there's an error retrieving or processing the metadata
+ */
+ public CompletableFuture<List<BucketMetadata>> getTableMetadataFromZkAsync(
+ TablePath tablePath, long tableId, boolean isPartitioned) {
+
+ PhysicalTablePath physicalTablePath = PhysicalTablePath.of(tablePath);
+
+ return getAssignmentInfoAsync(tableId, physicalTablePath)
+ .thenCompose(
+ assignmentInfo -> {
+ if (assignmentInfo.tableAssignment == null) {
+ if (isPartitioned) {
+ LOG.warn(
+ "No table assignment node found for table {}", tableId);
+ }
+ return CompletableFuture.completedFuture(
+ new ArrayList<BucketMetadata>());
+ }
+
+ TableAssignment tableAssignment = assignmentInfo.tableAssignment;
+ if (tableAssignment == null) {
+ return CompletableFuture.completedFuture(
+ new ArrayList<BucketMetadata>());
+ }
+ tableAssignment = checkNotNull(tableAssignment);
+
+ Map<Integer, BucketAssignment> bucketAssignments =
+ tableAssignment.getBucketAssignments();
+ if (bucketAssignments.isEmpty()) {
+ return CompletableFuture.completedFuture(
+ new ArrayList<BucketMetadata>());
+ }
+
+ // Create TableBucket collection for batch processing
+ List<TableBucket> tableBuckets = new ArrayList<>();
+ Map<TableBucket, Map.Entry<Integer, BucketAssignment>>
+ bucketToAssignmentMap = new HashMap<>();
+
+ for (Map.Entry<Integer, BucketAssignment> assignment :
+ bucketAssignments.entrySet()) {
+ int bucketId = assignment.getKey();
+ TableBucket tableBucket = new TableBucket(tableId, null, bucketId);
+ tableBuckets.add(tableBucket);
+ bucketToAssignmentMap.put(tableBucket, assignment);
+ }
+
+ // Use batch async method to get all LeaderAndIsr information in one ZK
+ // call
+ return getLeaderAndIsrsAsync(tableBuckets)
+ .thenApply(
+ leaderAndIsrMap ->
+ ZooKeeperMetadataUtils
+ .assembleBucketMetadataList(
+ tableBuckets,
+ bucketToAssignmentMap,
+ leaderAndIsrMap,
+ null));
+ })
+ .exceptionally(
+ throwable ->
+ ZooKeeperMetadataUtils.handleTableMetadataException(
+ throwable, tablePath, tableId));
+ }
+
+ public static CompletableFuture<List<BucketMetadata>> getTableMetadataFromZkAsync(
+ ZooKeeperClient zkClient, TablePath tablePath, long tableId, boolean isPartitioned) {
+ return zkClient.getTableMetadataFromZkAsync(tablePath, tableId, isPartitioned);
+ }
+
+ /**
+ * Asynchronously retrieves partition metadata from ZooKeeper for the specified partition.
+ *
+ * <p>This method fetches the partition assignment information from ZooKeeper and converts it
+ * into partition metadata including all bucket metadata. The operation is performed
+ * asynchronously to avoid blocking the calling thread.
+ *
+ * @param partitionPath the physical path of the partition to retrieve metadata for
+ * @return a CompletableFuture containing a PartitionMetadata object representing the partition
+ * and all its buckets. Returns a PartitionMetadata with empty bucket list if no partition
+ * assignment is found.
+ * @throws PartitionNotExistException if the partition does not exist in ZooKeeper
+ * @throws FlussRuntimeException if there's an error retrieving or processing the metadata
+ * @throws IllegalArgumentException if the partition ID is null (should never happen for valid
+ * partitions)
+ */
+ public CompletableFuture<PartitionMetadata> getPartitionMetadataFromZkAsync(
+ PhysicalTablePath partitionPath) {
+
+ return getAssignmentInfoAsync(null, partitionPath)
+ .thenCompose(
+ assignmentInfo -> {
+ checkNotNull(
+ assignmentInfo.partitionId,
+ "partition id must be not null for " + partitionPath);
+
+ if (assignmentInfo.tableAssignment == null) {
+ LOG.warn(
+ "No partition assignment node found for partition {}",
+ partitionPath);
+
+ PartitionMetadata emptyMetadata =
+ new PartitionMetadata(
+ assignmentInfo.tableId,
+ partitionPath.getPartitionName(),
+ assignmentInfo.partitionId != null
+ ? assignmentInfo.partitionId
+ : -1L,
+ new ArrayList<>());
+
+ return CompletableFuture.completedFuture(emptyMetadata);
+ }
+
+ TableAssignment tableAssignment =
+ checkNotNull(assignmentInfo.tableAssignment);
+ Map<Integer, BucketAssignment> bucketAssignments =
+ tableAssignment.getBucketAssignments();
+ if (bucketAssignments.isEmpty()) {
+ return CompletableFuture.completedFuture(
+ new PartitionMetadata(
+ assignmentInfo.tableId,
+ partitionPath.getPartitionName(),
+ assignmentInfo.partitionId != null
+ ? assignmentInfo.partitionId
+ : -1L,
+ new ArrayList<>()));
+ }
+
+ // Create TableBucket collection for batch processing
+ List<TableBucket> tableBuckets = new ArrayList<>();
+ Map<TableBucket, Map.Entry<Integer, BucketAssignment>>
+ bucketToAssignmentMap = new HashMap<>();
+
+ for (Map.Entry<Integer, BucketAssignment> assignment :
+ bucketAssignments.entrySet()) {
+ int bucketId = assignment.getKey();
+ TableBucket tableBucket =
+ new TableBucket(
+ assignmentInfo.tableId,
+ assignmentInfo.partitionId,
+ bucketId);
+ tableBuckets.add(tableBucket);
+ bucketToAssignmentMap.put(tableBucket, assignment);
+ }
+
+ // Use batch async method to get all LeaderAndIsr information in one ZK
+ // call
+ return getLeaderAndIsrsAsync(tableBuckets)
+ .thenApply(
+ leaderAndIsrMap -> {
+ List<BucketMetadata> bucketMetadataList =
+ ZooKeeperMetadataUtils
+ .assembleBucketMetadataList(
+ tableBuckets,
+ bucketToAssignmentMap,
+ leaderAndIsrMap,
+ null);
+ return new PartitionMetadata(
+ assignmentInfo.tableId,
+ partitionPath.getPartitionName(),
+ assignmentInfo.partitionId != null
+ ? assignmentInfo.partitionId
+ : -1L,
+ bucketMetadataList);
+ });
+ })
+ .exceptionally(
+ throwable ->
+ ZooKeeperMetadataUtils.handlePartitionException(
+ throwable,
+ "Failed to get metadata for partition {}",
+ partitionPath,
+ String.format(
+ "Failed to get metadata for partition %s",
+ partitionPath)));
+ }
+
+ public static CompletableFuture<PartitionMetadata> getPartitionMetadataFromZkAsync(
+ PhysicalTablePath partitionPath, ZooKeeperClient zkClient) {
+ return zkClient.getPartitionMetadataFromZkAsync(partitionPath);
+ }
+
+ /**
+ * Asynchronously retrieves partition metadata from ZooKeeper for multiple partitions in batch.
+ *
+ * <p>This method efficiently fetches partition metadata for multiple partitions across multiple
+ * tables in a single batched operation. It performs the following steps:
+ *
+ * <ul>
+ * <li>Maps table paths to their partition information (name and ID mappings)
+ * <li>Validates that all requested partition IDs exist
+ * <li>Builds partition path information with resolved table IDs
+ * <li>Batch retrieves partition assignments and bucket metadata
+ * </ul>
+ *
+ * <p>The batch operation is optimized to minimize ZooKeeper calls by grouping related
+ * operations together, making it much more efficient than calling {@link
+ * #getPartitionMetadataFromZkAsync(PhysicalTablePath)} individually for each partition.
+ *
+ * @param tablePaths the collection of table paths that may contain the requested partitions.
+ * This is used to scope the search and improve efficiency. Cannot be empty if
+ * partitionIdSet is not empty.
+ * @param partitionIdSet the set of partition IDs to retrieve metadata for. Cannot be empty if
+ * tablePaths is not empty.
+ * @return a CompletableFuture containing a list of PartitionMetadata objects for all found
+ * partitions. The list may be smaller than the input partition ID set if some partitions
+ * have no assignment data. Returns an empty list if either input parameter is empty.
+ * @throws PartitionNotExistException if any of the requested partition IDs do not exist in any
+ * of the provided tables
+ * @throws FlussRuntimeException if there's an error during ZooKeeper operations or metadata
+ * processing
+ */
+ public CompletableFuture<List<PartitionMetadata>> batchGetPartitionMetadataFromZkAsync(
+ Collection<TablePath> tablePaths, Set<Long> partitionIdSet) {
+ if (tablePaths.isEmpty() || partitionIdSet.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.<PartitionMetadata>emptyList());
+ }
+
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return getPartitionNameAndIdsForTables(new ArrayList<>(tablePaths));
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to get partition name and ids for tables", e);
+ }
+ })
+ .thenCompose(
+ tableToPartitionInfo -> {
+ // Inline collectValidPartitionsAndTables logic
+ Set<Long> remainingPartitionIds = new HashSet<>(partitionIdSet);
+ Set<TablePath> tablesNeedingIds = new HashSet<>();
+ Map<TablePath, List<Tuple2<String, Long>>> tablePartitionMapping =
+ new HashMap<>();
+
+ for (Map.Entry<TablePath, Map<String, Long>> entry :
+ tableToPartitionInfo.entrySet()) {
+ TablePath tablePath = entry.getKey();
+ Map<String, Long> partitionInfo = entry.getValue();
+
+ List<Tuple2<String, Long>> validPartitions =
+ ZooKeeperMetadataUtils.findValidPartitionsForTable(
+ partitionInfo,
+ partitionIdSet,
+ remainingPartitionIds);
+
+ if (!validPartitions.isEmpty()) {
+ tablesNeedingIds.add(tablePath);
+ tablePartitionMapping.put(tablePath, validPartitions);
+ }
+ }
+
+ if (!remainingPartitionIds.isEmpty()) {
+ throw new PartitionNotExistException(
+ "Partition not exist for partition ids: "
+ + remainingPartitionIds);
+ }
+
+ if (tablePartitionMapping.isEmpty()) {
+ return CompletableFuture.completedFuture(
+ Collections.<PartitionPathInfo>emptyList());
+ }
+
+ return getTablesAsync(tablesNeedingIds)
+ .thenApply(
+ tableRegistrations ->
+ ZooKeeperMetadataUtils
+ .buildPartitionPathInfoList(
+ tablePartitionMapping,
+ tableRegistrations));
+ })
+ .thenCompose(
+ partitionPathInfos -> {
+ if (partitionPathInfos.isEmpty()) {
+ return CompletableFuture.completedFuture(
+ Collections.<PartitionMetadata>emptyList());
+ }
+
+ List<Long> partitionIds =
+ partitionPathInfos.stream()
+ .map(info -> info.partitionId)
+ .collect(Collectors.toList());
+
+ CompletableFuture<Map<Long, PartitionAssignment>>
+ partitionAssignmentsFuture;
+ if (partitionIds.isEmpty()) {
+ partitionAssignmentsFuture =
+ CompletableFuture.completedFuture(Collections.emptyMap());
+ } else {
+ Map<String, Long> path2PartitionIdMap =
+ partitionIds.stream()
+ .collect(
+ Collectors.toMap(
+ PartitionIdZNode::path, id -> id));
+
+ partitionAssignmentsFuture =
+ getDataInBackgroundAsync(path2PartitionIdMap.keySet())
+ .thenApply(
+ responses ->
+ ZooKeeperMetadataUtils
+ .processMultipleDataResponses(
+ responses,
+ response ->
+ path2PartitionIdMap
+ .get(
+ response
+ .getPath()),
+ PartitionIdZNode
+ ::decode,
+ "partition assignment"));
+ }
+
+ return partitionAssignmentsFuture.thenCompose(
+ partitionAssignments -> {
+ List<AssignmentInfo> assignmentInfos = new ArrayList<>();
+ for (PartitionPathInfo pathInfo : partitionPathInfos) {
+ PartitionAssignment partitionAssignment =
+ partitionAssignments.get(pathInfo.partitionId);
+ AssignmentInfo assignmentInfo =
+ new AssignmentInfo(
+ pathInfo.tableId,
+ partitionAssignment,
+ pathInfo.partitionId);
+ assignmentInfos.add(assignmentInfo);
+ }
+
+ List<TableBucket> allTableBuckets = new ArrayList<>();
+ Map<TableBucket, Integer> bucketToIndexMap =
+ new HashMap<>();
+
+ for (int i = 0; i < assignmentInfos.size(); i++) {
+ AssignmentInfo assignmentInfo = assignmentInfos.get(i);
+ PartitionPathInfo pathInfo = partitionPathInfos.get(i);
+
+ if (assignmentInfo.tableAssignment != null) {
+ for (Integer bucketId :
+ assignmentInfo.tableAssignment
+ .getBuckets()) {
+ TableBucket tableBucket =
+ new TableBucket(
+ pathInfo.tableId,
+ pathInfo.partitionId,
+ bucketId);
+ allTableBuckets.add(tableBucket);
+ bucketToIndexMap.put(tableBucket, i);
+ }
+ }
+ }
+
+ return getLeaderAndIsrsAsync(allTableBuckets)
+ .thenApply(
+ leaderAndIsrMap ->
+ ZooKeeperMetadataUtils
+ .assemblePartitionMetadata(
+ partitionPathInfos,
+ assignmentInfos,
+ bucketToIndexMap,
+ leaderAndIsrMap));
+ });
+ })
+ .exceptionally(
+ throwable ->
+ ZooKeeperMetadataUtils.handlePartitionException(
+ throwable,
+ "Failed to batch get partition metadata for partition ids: {}",
+ partitionIdSet,
+ "Failed to batch get partition metadata for partition ids: "
+ + partitionIdSet));
+ }
+
+ private CompletableFuture<AssignmentInfo> getAssignmentInfoAsync(
+ @Nullable Long tableId, PhysicalTablePath physicalTablePath) {
+
+ if (physicalTablePath.getPartitionName() != null) {
+ return getPartitionAsync(
+ physicalTablePath.getTablePath(), physicalTablePath.getPartitionName())
+ .thenApply(
+ tablePartition -> {
+ if (tablePartition == null) {
+ throw new PartitionNotExistException(
+ "Table partition '"
+ + physicalTablePath
+ + "' does not exist.");
+ }
+ return tablePartition;
+ })
+ .thenCompose(
+ tablePartition -> {
+ long partitionId = tablePartition.getPartitionId();
+ long partitionTableId = tablePartition.getTableId();
+
+ return getPartitionAssignmentAsync(partitionId)
+ .thenApply(
+ tableAssignment ->
+ new AssignmentInfo(
+ partitionTableId,
+ tableAssignment,
+ partitionId));
+ });
+ } else {
+ checkNotNull(tableId, "tableId must be not null");
+ long realTableId = tableId != null ? tableId : -1L;
+
+ return getTableAssignmentAsync(realTableId)
+ .thenApply(
+ tableAssignment ->
+ new AssignmentInfo(realTableId, tableAssignment, null));
+ }
+ }
+
/** Close the underlying ZooKeeperClient. */
@Override
public void close() {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperMetadataUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperMetadataUtils.java
new file mode 100644
index 000000000..803db46b2
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperMetadataUtils.java
@@ -0,0 +1,478 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.zk;
+
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.PartitionNotExistException;
+import org.apache.fluss.exception.TableNotExistException;
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.metadata.BucketMetadata;
+import org.apache.fluss.server.metadata.PartitionMetadata;
+import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetChildrenResponse;
+import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetDataResponse;
+import org.apache.fluss.server.zk.data.BucketAssignment;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+import org.apache.fluss.server.zk.data.TableAssignment;
+import org.apache.fluss.server.zk.data.TableRegistration;
+import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+
+/** Utility class containing helper methods for ZooKeeper metadata operations. */
+public class ZooKeeperMetadataUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperMetadataUtils.class);
+
+ /**
+ * Template method to process multiple ZooKeeper data responses with decoder.
+ *
+ * @param responses list of ZkGetDataResponse from ZooKeeper
+ * @param keyExtractor function to extract key from response
+ * @param decoder function to decode byte array to target type
+ * @param operationName name of the operation for error messages
+ * @param <K> the type of the result map key
+ * @param <V> the type of the result map value
+ * @return Map containing decoded results for successful responses
+ */
+ public static <K, V> Map<K, V> processMultipleDataResponses(
+ List<ZkGetDataResponse> responses,
+ Function<ZkGetDataResponse, K> keyExtractor,
+ Function<byte[], V> decoder,
+ String operationName) {
+ Map<K, V> result = new HashMap<>();
+ for (ZkGetDataResponse response : responses) {
+ if (response.getResultCode() == KeeperException.Code.OK) {
+ K key = keyExtractor.apply(response);
+ V value = decoder.apply(response.getData());
+ result.put(key, value);
+ } else {
+ LOG.warn(
+ "Failed to get {} for path {}: {}",
+ operationName,
+ response.getPath(),
+ response.getResultCode());
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Template method to process multiple ZooKeeper data responses with decoder and condition.
+ *
+ * @param responses list of ZkGetDataResponse from ZooKeeper
+ * @param keyExtractor function to extract key from response
+ * @param decoder function to decode byte array to target type
+ * @param condition additional condition to check before processing (beyond OK status)
+ * @param operationName name of the operation for error messages
+ * @param <K> the type of the result map key
+ * @param <V> the type of the result map value
+ * @return Map containing decoded results for successful responses
+ */
+ public static <K, V> Map<K, V> processMultipleDataResponsesWithCondition(
+ List<ZkGetDataResponse> responses,
+ Function<ZkGetDataResponse, K> keyExtractor,
+ Function<byte[], V> decoder,
+ Function<ZkGetDataResponse, Boolean> condition,
+ String operationName) {
+ Map<K, V> result = new HashMap<>();
+ for (ZkGetDataResponse response : responses) {
+ if (response.getResultCode() == KeeperException.Code.OK && condition.apply(response)) {
+ K key = keyExtractor.apply(response);
+ V value = decoder.apply(response.getData());
+ result.put(key, value);
+ } else {
+ String reason =
+ response.getResultCode() != KeeperException.Code.OK
+ ? response.getResultCode().toString()
+ : "condition check failed";
+ LOG.warn(
+ "Failed to get {} for path {}: {}",
+ operationName,
+ response.getPath(),
+ reason);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Template method to process multiple ZooKeeper data responses with decoder and Optional
+ * result. This method handles NONODE as a valid case returning Optional.empty().
+ *
+ * @param responses list of ZkGetDataResponse from ZooKeeper
+ * @param keyExtractor function to extract key from response
+ * @param decoder function to decode byte array to target type
+ * @param operationName name of the operation for error messages
+ * @param <K> the type of the result map key
+ * @param <V> the type of the result map value
+ * @return Map containing Optional results for all responses (including NONODE cases)
+ */
+ public static <K, V> Map<K, Optional<V>> processMultipleDataResponsesWithOptional(
+ List<ZkGetDataResponse> responses,
+ Function<ZkGetDataResponse, K> keyExtractor,
+ Function<byte[], V> decoder,
+ String operationName) {
+ Map<K, Optional<V>> result = new HashMap<>();
+ for (ZkGetDataResponse response : responses) {
+ K key = keyExtractor.apply(response);
+ if (response.getResultCode() == KeeperException.Code.OK) {
+ V value = decoder.apply(response.getData());
+ result.put(key, Optional.of(value));
+ } else if (response.getResultCode() == KeeperException.Code.NONODE) {
+ result.put(key, Optional.empty());
+ } else {
+ LOG.warn(
+ "Failed to get {} for path {}: {}",
+ operationName,
+ response.getPath(),
+ response.getResultCode());
+ result.put(key, Optional.empty());
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Template method to process multiple ZooKeeper children responses with decoder.
+ *
+ * @param responses list of ZkGetChildrenResponse from ZooKeeper
+ * @param keyExtractor function to extract key from response
+ * @param operationName name of the operation for error messages
+ * @param <K> the type of the result map key
+ * @return Map containing children lists for successful responses
+ */
+ public static <K> Map<K, List<String>> processMultipleChildrenResponses(
+ List<ZkGetChildrenResponse> responses,
+ Function<ZkGetChildrenResponse, K> keyExtractor,
+ String operationName) {
+ Map<K, List<String>> result = new HashMap<>();
+ for (ZkGetChildrenResponse response : responses) {
+ if (response.getResultCode() == KeeperException.Code.OK) {
+ K key = keyExtractor.apply(response);
+ result.put(key, response.getChildren());
+ } else {
+ LOG.warn(
+ "Failed to get {} for path {}: {}",
+ operationName,
+ response.getPath(),
+ response.getResultCode());
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Template method to process single ZooKeeper data response with decoder.
+ *
+ * @param responses list of ZkGetDataResponse from ZooKeeper
+ * @param decoder function to decode byte array to target type
+ * @param path ZooKeeper path for error messages
+ * @param <T> the type of the decoded result
+ * @return decoded result, or null if no data or NONODE
+ */
+ public static <T> T processSingleDataResponse(
+ List<ZkGetDataResponse> responses, Function<byte[], T> decoder, String path) {
+ if (responses.isEmpty()) {
+ return null;
+ }
+
+ ZkGetDataResponse response = responses.get(0);
+ if (response.getResultCode() == KeeperException.Code.OK) {
+ return decoder.apply(response.getData());
+ } else if (response.getResultCode() == KeeperException.Code.NONODE) {
+ return null;
+ } else {
+ throw new RuntimeException(KeeperException.create(response.getResultCode(), path));
+ }
+ }
+
+ /**
+ * Handle exceptions that occur during table metadata retrieval.
+ *
+ * @param throwable the exception that occurred
+ * @param tablePath the table path for error context
+ * @param tableId the table ID for error context
+ * @return never returns normally, always throws an exception
+ * @throws TableNotExistException if the original cause was table not existing
+ * @throws FlussRuntimeException for other errors
+ */
+ public static List<BucketMetadata> handleTableMetadataException(
+ Throwable throwable, TablePath tablePath, long tableId) {
+
+ Throwable cause = ExceptionUtils.stripCompletionException(throwable);
+ cause = ExceptionUtils.stripException(cause, RuntimeException.class);
+
+ if (cause instanceof TableNotExistException) {
+ throw (TableNotExistException) cause;
+ }
+
+ LOG.error("Failed to get metadata for table {}, id {}", tablePath, tableId, cause);
+ throw new FlussRuntimeException(
+ String.format("Failed to get metadata for table %s, id %d", tablePath, tableId),
+ cause);
+ }
+
+ /**
+ * Assemble a list of BucketMetadata from table buckets, assignments, and leader information.
+ *
+ * @param tableBuckets the list of table buckets to process
+ * @param bucketToAssignmentMap mapping from table buckets to their assignment entries
+ * @param leaderAndIsrMap mapping from table buckets to their leader and ISR information
+ * @param leaderEpoch the leader epoch to use for all buckets (can be null)
+ * @return a list of BucketMetadata objects
+ */
+ public static List<BucketMetadata> assembleBucketMetadataList(
+ List<TableBucket> tableBuckets,
+ Map<TableBucket, Map.Entry<Integer, BucketAssignment>> bucketToAssignmentMap,
+ Map<TableBucket, Optional<LeaderAndIsr>> leaderAndIsrMap,
+ @Nullable Integer leaderEpoch) {
+
+ List<BucketMetadata> result = new ArrayList<>();
+
+ for (TableBucket tableBucket : tableBuckets) {
+ Map.Entry<Integer, BucketAssignment> assignment =
+ bucketToAssignmentMap.get(tableBucket);
+ int bucketId = assignment.getKey();
+ List<Integer> replicas = assignment.getValue().getReplicas();
+
+ Optional<LeaderAndIsr> optLeaderAndIsr = leaderAndIsrMap.get(tableBucket);
+ Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
+
+ result.add(new BucketMetadata(bucketId, leader, leaderEpoch, replicas));
+ }
+
+ return result;
+ }
+
+ /**
+ * Common exception handler for partition-related operations.
+ *
+ * @param throwable the exception that occurred
+ * @param logMessageTemplate the template for log message (with {} placeholder)
+ * @param logContext the context object to be logged
+ * @param errorMessage the error message for FlussRuntimeException
+ * @param <T> the return type (never actually returns)
+ * @return never returns normally, always throws an exception
+ * @throws PartitionNotExistException if the original cause was partition not existing
+ * @throws FlussRuntimeException for other errors
+ */
+ public static <T> T handlePartitionException(
+ Throwable throwable,
+ String logMessageTemplate,
+ Object logContext,
+ String errorMessage) {
+
+ Throwable cause = ExceptionUtils.stripCompletionException(throwable);
+ cause = ExceptionUtils.stripException(cause, RuntimeException.class);
+
+ if (cause instanceof PartitionNotExistException) {
+ throw (PartitionNotExistException) cause;
+ }
+
+ LOG.error(logMessageTemplate, logContext, cause);
+ throw new FlussRuntimeException(errorMessage, cause);
+ }
+
+ /**
+ * Find valid partitions for a specific table.
+ *
+ * @param partitionInfo partition information for the table
+ * @param partitionIdSet the set of partition IDs to look for
+ * @param remainingPartitionIds mutable set to track unmatched partition IDs
+ * @return list of valid partitions (name, id) tuples
+ */
+ public static List<Tuple2<String, Long>> findValidPartitionsForTable(
+ Map<String, Long> partitionInfo,
+ Set<Long> partitionIdSet,
+ Set<Long> remainingPartitionIds) {
+
+ List<Tuple2<String, Long>> validPartitions = new ArrayList<>();
+
+ for (Map.Entry<String, Long> partitionEntry : partitionInfo.entrySet()) {
+ Long partitionId = partitionEntry.getValue();
+ if (partitionIdSet.contains(partitionId)) {
+ String partitionName = partitionEntry.getKey();
+ validPartitions.add(Tuple2.of(partitionName, partitionId));
+ remainingPartitionIds.remove(partitionId);
+ }
+ }
+
+ return validPartitions;
+ }
+
+ /**
+ * Build the list of PartitionPathInfo from table registrations and partition mappings.
+ *
+ * @param tablePartitionMapping mapping of tables to their valid partitions
+ * @param tableRegistrations table registration information containing table IDs
+ * @return list of PartitionPathInfo objects
+ */
+ public static List<PartitionPathInfo> buildPartitionPathInfoList(
+ Map<TablePath, List<Tuple2<String, Long>>> tablePartitionMapping,
+ Map<TablePath, TableRegistration> tableRegistrations) {
+
+ List<PartitionPathInfo> targetPartitionInfos = new ArrayList<>();
+
+ for (Map.Entry<TablePath, List<Tuple2<String, Long>>> entry :
+ tablePartitionMapping.entrySet()) {
+ TablePath tablePath = entry.getKey();
+ TableRegistration tableRegistration = tableRegistrations.get(tablePath);
+
+ if (tableRegistration != null) {
+ long tableId = tableRegistration.tableId;
+ addPartitionPathInfosForTable(
+ targetPartitionInfos, tablePath, tableId, entry.getValue());
+ }
+ }
+
+ return targetPartitionInfos;
+ }
+
+ /**
+ * Add PartitionPathInfo objects for all partitions of a specific table.
+ *
+ * @param targetPartitionInfos target list to add to
+ * @param tablePath the table path
+ * @param tableId the table ID
+ * @param partitionInfos list of partition (name, id) tuples
+ */
+ public static void addPartitionPathInfosForTable(
+ List<PartitionPathInfo> targetPartitionInfos,
+ TablePath tablePath,
+ long tableId,
+ List<Tuple2<String, Long>> partitionInfos) {
+
+ for (Tuple2<String, Long> partitionInfo : partitionInfos) {
+ String partitionName = partitionInfo.f0;
+ Long partitionId = partitionInfo.f1;
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(tablePath, partitionName);
+ targetPartitionInfos.add(new PartitionPathInfo(partitionPath, partitionId, tableId));
+ }
+ }
+
+ /**
+ * Assemble partition metadata from the collected bucket information.
+ *
+ * @param partitionPathInfos the partition path information
+ * @param assignmentInfos the assignment information for each partition
+ * @param bucketToIndexMap mapping from table buckets to partition index
+ * @param leaderAndIsrMap mapping from table buckets to leader information
+ * @return list of assembled partition metadata
+ */
+ public static List<PartitionMetadata> assemblePartitionMetadata(
+ List<PartitionPathInfo> partitionPathInfos,
+ List<AssignmentInfo> assignmentInfos,
+ Map<TableBucket, Integer> bucketToIndexMap,
+ Map<TableBucket, Optional<LeaderAndIsr>> leaderAndIsrMap) {
+
+ Map<Integer, List<BucketMetadata>> indexToBuckets = new HashMap<>();
+
+ for (Map.Entry<TableBucket, Optional<LeaderAndIsr>> entry : leaderAndIsrMap.entrySet()) {
+ TableBucket tableBucket = entry.getKey();
+ Optional<LeaderAndIsr> optLeaderAndIsr = entry.getValue();
+ Integer assignmentIndex = bucketToIndexMap.get(tableBucket);
+
+ if (assignmentIndex != null) {
+ AssignmentInfo assignmentInfo = assignmentInfos.get(assignmentIndex);
+ if (assignmentInfo.tableAssignment != null) {
+ BucketAssignment bucketAssignment =
+ assignmentInfo
+ .tableAssignment
+ .getBucketAssignments()
+ .get(tableBucket.getBucket());
+
+ if (bucketAssignment != null) {
+ Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
+ BucketMetadata bucketMetadata =
+ new BucketMetadata(
+ tableBucket.getBucket(),
+ leader,
+ // leaderEpoch
+ null,
+ bucketAssignment.getReplicas());
+
+ indexToBuckets
+ .computeIfAbsent(assignmentIndex, k -> new ArrayList<>())
+ .add(bucketMetadata);
+ }
+ }
+ }
+ }
+
+ List<PartitionMetadata> result = new ArrayList<>();
+ for (int i = 0; i < partitionPathInfos.size(); i++) {
+ PartitionPathInfo pathInfo = partitionPathInfos.get(i);
+ List<BucketMetadata> bucketMetadataList =
+ indexToBuckets.getOrDefault(i, new ArrayList<>());
+
+ result.add(
+ new PartitionMetadata(
+ pathInfo.tableId,
+ pathInfo.partitionPath.getPartitionName(),
+ pathInfo.partitionId,
+ bucketMetadataList));
+ }
+
+ return result;
+ }
+
+ /** Data structure to hold partition path and its corresponding partition ID. */
+ public static class PartitionPathInfo {
+ public final PhysicalTablePath partitionPath;
+ public final long partitionId;
+ public final long tableId;
+
+ public PartitionPathInfo(PhysicalTablePath partitionPath, long partitionId, long tableId) {
+ this.partitionPath = partitionPath;
+ this.partitionId = partitionId;
+ this.tableId = tableId;
+ }
+ }
+
+ /** Data structure to hold assignment information. */
+ public static class AssignmentInfo {
+ public final long tableId;
+ // null then the bucket doesn't belong to a partition. Otherwise, not null
+ public final @Nullable Long partitionId;
+ public final @Nullable TableAssignment tableAssignment;
+
+ public AssignmentInfo(
+ long tableId,
+ @Nullable TableAssignment tableAssignment,
+ @Nullable Long partitionId) {
+ this.tableId = tableId;
+ this.tableAssignment = tableAssignment;
+ this.partitionId = partitionId;
+ }
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java
index 43c49a321..d2cd6b320 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/MetadataUpdateITCase.java
@@ -20,8 +20,6 @@ package org.apache.fluss.server.metadata;
import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
-import org.apache.fluss.exception.PartitionNotExistException;
-import org.apache.fluss.exception.TableNotExistException;
import org.apache.fluss.metadata.PartitionSpec;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
@@ -63,7 +61,6 @@ import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDropTable
import static org.apache.fluss.server.testutils.TableMetadataAssert.assertTableMetadata;
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT Case for metadata update. */
class MetadataUpdateITCase {
@@ -372,22 +369,18 @@ class MetadataUpdateITCase {
TableMetadata tableMetadataFromZk =
getTableMetadataFromZk(tablePath, tableId);
assertTableMetadata(
- serverMetadataCache.getTableMetadata(
- tablePath))
+ serverMetadataCache
+ .getTableMetadata(tablePath)
+ .get())
.isEqualTo(tableMetadataFromZk);
}
} else {
assertThat(serverMetadataCache.getTablePath(tableId))
.isNotPresent();
- assertThatThrownBy(
- () ->
- serverMetadataCache.getTableMetadata(
- tableContext.tablePath))
- .isInstanceOf(TableNotExistException.class)
- .hasMessageContaining(
- "Table '"
- + tableContext.tablePath
- + "' does not exist.");
+ assertThat(
+ serverMetadataCache.getTableMetadata(
+ tableContext.tablePath))
+ .isEmpty();
}
});
expectedPartitionNameById.forEach(
@@ -407,24 +400,19 @@ class MetadataUpdateITCase {
tableContext.partitionName,
partitionId);
assertPartitionMetadata(
- serverMetadataCache.getPartitionMetadata(
- physicalTablePath))
+ serverMetadataCache
+ .getPartitionMetadata(physicalTablePath)
+ .get())
.isEqualTo(partitionMetadataFromZk);
} else {
assertThat(
serverMetadataCache.getPhysicalTablePath(
partitionId))
.isNotPresent();
- assertThatThrownBy(
- () ->
- serverMetadataCache
- .getPartitionMetadata(
- physicalTablePath))
- .isInstanceOf(PartitionNotExistException.class)
- .hasMessageContaining(
- "Table partition '"
- + physicalTablePath
- + "' does not exist.");
+ assertThat(
+ serverMetadataCache.getPartitionMetadata(
+ physicalTablePath))
+ .isEmpty();
}
});
});
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java
index 90f012b83..47f445926 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java
@@ -306,7 +306,7 @@ public class TabletServerMetadataCacheTest {
TableInfo expectedTableInfo,
List<BucketMetadata> expectedBucketMetadataList) {
TablePath tablePath = serverMetadataCache.getTablePath(tableId).get();
- TableMetadata tableMetadata = serverMetadataCache.getTableMetadata(tablePath);
+ TableMetadata tableMetadata = serverMetadataCache.getTableMetadata(tablePath).get();
assertThat(tableMetadata.getTableInfo()).isEqualTo(expectedTableInfo);
assertThat(tableMetadata.getBucketMetadataList())
.hasSameElementsAs(expectedBucketMetadataList);
@@ -322,8 +322,10 @@ public class TabletServerMetadataCacheTest {
serverMetadataCache.getPhysicalTablePath(partitionId).get().getPartitionName();
assertThat(actualPartitionName).isEqualTo(expectedPartitionName);
PartitionMetadata partitionMetadata =
- serverMetadataCache.getPartitionMetadata(
- PhysicalTablePath.of(partitionedTablePath, actualPartitionName));
+ serverMetadataCache
+ .getPartitionMetadata(
+ PhysicalTablePath.of(partitionedTablePath, actualPartitionName))
+ .get();
assertThat(partitionMetadata.getTableId()).isEqualTo(expectedTableId);
assertThat(partitionMetadata.getPartitionId()).isEqualTo(expectedPartitionId);
assertThat(partitionMetadata.getPartitionName()).isEqualTo(actualPartitionName);
@@ -331,6 +333,195 @@ public class TabletServerMetadataCacheTest {
.hasSameElementsAs(expectedBucketMetadataList);
}
+ // Tests merged from TabletServerMetadataCacheExtendedTest - specific metadata operations
+
+ @Test
+ void testUpdateTableMetadata() {
+ // Given: a table metadata using existing test data
+ TableMetadata tableMetadata = new TableMetadata(DATA1_TABLE_INFO, initialBucketMetadata);
+
+ // When: update table metadata
+ serverMetadataCache.updateTableMetadata(tableMetadata);
+
+ // Then: verify the table metadata is cached correctly
+ assertThat(serverMetadataCache.getTablePath(DATA1_TABLE_ID)).isPresent();
+ assertThat(serverMetadataCache.getTablePath(DATA1_TABLE_ID).get())
+ .isEqualTo(DATA1_TABLE_PATH);
+
+ TableMetadata cachedMetadata = serverMetadataCache.getTableMetadata(DATA1_TABLE_PATH).get();
+ assertThat(cachedMetadata).isNotNull();
+ assertThat(cachedMetadata.getTableInfo()).isEqualTo(DATA1_TABLE_INFO);
+ assertThat(cachedMetadata.getBucketMetadataList()).hasSameElementsAs(initialBucketMetadata);
+ }
+
+ @Test
+ void testUpdateTableMetadataMergeWithExisting() {
+ // Given: existing table metadata in cache
+ TableMetadata existingMetadata = new TableMetadata(DATA1_TABLE_INFO, initialBucketMetadata);
+ serverMetadataCache.updateTableMetadata(existingMetadata);
+
+ // When: update with new bucket metadata
+ List<BucketMetadata> newBucketMetadata =
+ Arrays.asList(
+ new BucketMetadata(0, 1, 1, Arrays.asList(1, 2, 3)),
+ new BucketMetadata(2, 2, 0, Arrays.asList(2, 3, 4)));
+ TableMetadata newMetadata = new TableMetadata(DATA1_TABLE_INFO, newBucketMetadata);
+ serverMetadataCache.updateTableMetadata(newMetadata);
+
+ // Then: verify the new metadata replaces the old one
+ TableMetadata cachedMetadata = serverMetadataCache.getTableMetadata(DATA1_TABLE_PATH).get();
+ assertThat(cachedMetadata.getBucketMetadataList()).hasSameElementsAs(newBucketMetadata);
+ assertThat(cachedMetadata.getBucketMetadataList())
+ .doesNotContainAnyElementsOf(initialBucketMetadata);
+ }
+
+ @Test
+ void testUpdatePartitionMetadata() {
+ // Given: table metadata exists in cache (using existing partition table)
+ TableMetadata tableMetadata = new TableMetadata(partitionTableInfo, initialBucketMetadata);
+ serverMetadataCache.updateTableMetadata(tableMetadata);
+
+ // When: update partition metadata
+ String partitionName = "p_new";
+ long partitionId = 2001L;
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(partitionedTablePath, partitionName);
+ PartitionMetadata partitionMetadata =
+ new PartitionMetadata(
+ partitionTableId, partitionName, partitionId, initialBucketMetadata);
+
+ serverMetadataCache.updatePartitionMetadata(partitionMetadata);
+
+ // Then: verify the partition metadata is cached correctly
+ assertThat(serverMetadataCache.getPhysicalTablePath(partitionId)).isPresent();
+ assertThat(serverMetadataCache.getPhysicalTablePath(partitionId).get())
+ .isEqualTo(partitionPath);
+
+ PartitionMetadata cachedMetadata =
+ serverMetadataCache.getPartitionMetadata(partitionPath).get();
+ assertThat(cachedMetadata).isNotNull();
+ assertThat(cachedMetadata.getTableId()).isEqualTo(partitionTableId);
+ assertThat(cachedMetadata.getPartitionId()).isEqualTo(partitionId);
+ assertThat(cachedMetadata.getPartitionName()).isEqualTo(partitionName);
+ assertThat(cachedMetadata.getBucketMetadataList()).hasSameElementsAs(initialBucketMetadata);
+ }
+
+ @Test
+ void testUpdatePartitionMetadataWithoutTable() {
+ // Given: no table metadata in cache
+ String partitionName = "p_no_table";
+ long partitionId = 2002L;
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(partitionedTablePath, partitionName);
+ PartitionMetadata partitionMetadata =
+ new PartitionMetadata(
+ partitionTableId, partitionName, partitionId, initialBucketMetadata);
+
+ // When: update partition metadata without table
+ serverMetadataCache.updatePartitionMetadata(partitionMetadata);
+
+ // Then: partition metadata should not be cached
+ assertThat(serverMetadataCache.getPhysicalTablePath(partitionId)).isEmpty();
+ assertThat(serverMetadataCache.getPartitionMetadata(partitionPath)).isEmpty();
+ }
+
+ @Test
+ void testUpdatePartitionMetadataMergeWithExisting() {
+ // Given: existing partition metadata in cache (using existing partition table)
+ TableMetadata tableMetadata = new TableMetadata(partitionTableInfo, initialBucketMetadata);
+ serverMetadataCache.updateTableMetadata(tableMetadata);
+
+ String partitionName = "p_merge";
+ long partitionId = 2003L;
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(partitionedTablePath, partitionName);
+ PartitionMetadata existingPartitionMetadata =
+ new PartitionMetadata(
+ partitionTableId, partitionName, partitionId, initialBucketMetadata);
+ serverMetadataCache.updatePartitionMetadata(existingPartitionMetadata);
+
+ // When: update with new bucket metadata
+ List<BucketMetadata> newBucketMetadata =
+ Arrays.asList(new BucketMetadata(0, 1, 1, Arrays.asList(1, 2, 3)));
+ PartitionMetadata newPartitionMetadata =
+ new PartitionMetadata(
+ partitionTableId, partitionName, partitionId, newBucketMetadata);
+ serverMetadataCache.updatePartitionMetadata(newPartitionMetadata);
+
+ // Then: verify the new metadata replaces the old one
+ PartitionMetadata cachedMetadata =
+ serverMetadataCache.getPartitionMetadata(partitionPath).get();
+ assertThat(cachedMetadata.getBucketMetadataList()).hasSameElementsAs(newBucketMetadata);
+ assertThat(cachedMetadata.getBucketMetadataList())
+ .doesNotContainAnyElementsOf(initialBucketMetadata);
+ }
+
+ @Test
+ void testConcurrentUpdateTableMetadata() throws InterruptedException {
+ // Given: multiple threads updating table metadata with existing test data
+ int threadCount = 10;
+ Thread[] threads = new Thread[threadCount];
+
+ // When: concurrent updates
+ for (int i = 0; i < threadCount; i++) {
+ threads[i] =
+ new Thread(
+ () -> {
+ // Use a different bucket configuration for each thread
+ List<BucketMetadata> threadBucketMetadata =
+ Arrays.asList(
+ new BucketMetadata(
+ 0,
+ 0,
+ Thread.currentThread().hashCode() % 3,
+ Arrays.asList(0, 1, 2)));
+ TableMetadata tableMetadata =
+ new TableMetadata(DATA1_TABLE_INFO, threadBucketMetadata);
+ serverMetadataCache.updateTableMetadata(tableMetadata);
+ });
+ }
+
+ for (Thread thread : threads) {
+ thread.start();
+ }
+
+ for (Thread thread : threads) {
+ thread.join();
+ }
+
+ // Then: verify the table path is still available (last update wins)
+ assertThat(serverMetadataCache.getTablePath(DATA1_TABLE_ID)).isPresent();
+ assertThat(serverMetadataCache.getTablePath(DATA1_TABLE_ID).get())
+ .isEqualTo(DATA1_TABLE_PATH);
+ }
+
+ @Test
+ void testClearTableMetadata() {
+ // Given: table and partition metadata in cache (using existing test data)
+ TableMetadata tableMetadata = new TableMetadata(DATA1_TABLE_INFO, initialBucketMetadata);
+ serverMetadataCache.updateTableMetadata(tableMetadata);
+
+ TableMetadata partitionTableMetadata =
+ new TableMetadata(partitionTableInfo, initialBucketMetadata);
+ serverMetadataCache.updateTableMetadata(partitionTableMetadata);
+
+ String partitionName = "p_clear";
+ long partitionId = 2004L;
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(partitionedTablePath, partitionName);
+ PartitionMetadata partitionMetadata =
+ new PartitionMetadata(
+ partitionTableId, partitionName, partitionId, initialBucketMetadata);
+ serverMetadataCache.updatePartitionMetadata(partitionMetadata);
+
+ // When: clear table metadata
+ serverMetadataCache.clearTableMetadata();
+
+ // Then: verify all table metadata is cleared
+ assertThat(serverMetadataCache.getTablePath(DATA1_TABLE_ID)).isEmpty();
+ assertThat(serverMetadataCache.getTableMetadata(DATA1_TABLE_PATH)).isEmpty();
+ assertThat(serverMetadataCache.getTablePath(partitionTableId)).isEmpty();
+ assertThat(serverMetadataCache.getTableMetadata(partitionedTablePath)).isEmpty();
+ assertThat(serverMetadataCache.getPhysicalTablePath(partitionId)).isEmpty();
+ assertThat(serverMetadataCache.getPartitionMetadata(partitionPath)).isEmpty();
+ }
+
private static final class TestingMetadataManager extends MetadataManager {
private final Map<TablePath, TableInfo> tableInfoMap = new HashMap<>();
@@ -340,8 +531,13 @@ public class TabletServerMetadataCacheTest {
tableInfos.forEach(tableInfo -> tableInfoMap.put(tableInfo.getTablePath(), tableInfo));
}
+ @Override
public TableInfo getTable(TablePath tablePath) throws TableNotExistException {
- return tableInfoMap.get(tablePath);
+ TableInfo tableInfo = tableInfoMap.get(tablePath);
+ if (tableInfo == null) {
+ throw new TableNotExistException("Table '" + tablePath + "' does not exist.");
+ }
+ return tableInfo;
}
}
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index 0d90a9a25..62aaff65d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -23,8 +23,6 @@ import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.exception.InvalidCoordinatorException;
import org.apache.fluss.exception.InvalidRequiredAcksException;
import org.apache.fluss.exception.NotLeaderOrFollowerException;
-import org.apache.fluss.exception.PartitionNotExistException;
-import org.apache.fluss.exception.TableNotExistException;
import org.apache.fluss.exception.UnknownTableOrBucketException;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
@@ -1847,11 +1845,10 @@ class ReplicaManagerTest extends ReplicaTestBase {
expectedTableMetadataById.forEach(
(k, v) -> {
if (v != null) {
- assertTableMetadata(serverMetadataCache.getTableMetadata(k)).isEqualTo(v);
+ assertTableMetadata(serverMetadataCache.getTableMetadata(k).get())
+ .isEqualTo(v);
} else {
- assertThatThrownBy(() -> serverMetadataCache.getTableMetadata(k))
- .isInstanceOf(TableNotExistException.class)
- .hasMessageContaining("Table '" + k + "' does not exist.");
+ assertThat(serverMetadataCache.getTableMetadata(k)).isEmpty();
}
});
@@ -1872,13 +1869,10 @@ class ReplicaManagerTest extends ReplicaTestBase {
expectedPartitionMetadataById.forEach(
(k, v) -> {
if (v != null) {
- assertPartitionMetadata(serverMetadataCache.getPartitionMetadata(k))
+ assertPartitionMetadata(serverMetadataCache.getPartitionMetadata(k).get())
.isEqualTo(v);
} else {
- assertThatThrownBy(() -> serverMetadataCache.getPartitionMetadata(k))
- .isInstanceOf(PartitionNotExistException.class)
- .hasMessageContaining(
- "Table partition '" + k + "' does not exist.");
+ assertThat(serverMetadataCache.getPartitionMetadata(k)).isEmpty();
}
});
}
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 afd10c943..454aac244 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
@@ -18,6 +18,7 @@
package org.apache.fluss.server.tablet;
import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.exception.CacheMissException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.InvalidRequiredAcksException;
import org.apache.fluss.metadata.LogFormat;
@@ -30,11 +31,14 @@ import org.apache.fluss.record.DefaultKvRecordBatch;
import org.apache.fluss.record.DefaultValueRecordBatch;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.FetchLogResponse;
import org.apache.fluss.rpc.messages.InitWriterRequest;
import org.apache.fluss.rpc.messages.InitWriterResponse;
import org.apache.fluss.rpc.messages.ListOffsetsResponse;
+import org.apache.fluss.rpc.messages.MetadataRequest;
+import org.apache.fluss.rpc.messages.MetadataResponse;
import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest;
import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse;
import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
@@ -78,6 +82,7 @@ import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V0;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V1;
@@ -102,6 +107,7 @@ import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newFetchLogR
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newLimitScanRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newListOffsetsRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newLookupRequest;
+import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newMetadataRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPrefixLookupRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newProduceLogRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest;
@@ -966,4 +972,137 @@ public class TabletServiceITCase {
return ServerRpcMessageUtils.makeNotifyLeaderAndIsrRequest(
0, Collections.singletonList(reqForBucket));
}
+
+ @Test
+ void testCacheOnlyTrue_NonExistentTable_ShouldThrowTableNotExistException() {
+ AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ // Create request for a non-existent table with cache_only = true
+ // Note: For non-existent tables, system throws TableNotExistException
+ // rather than CacheMissException since table validation happens first
+ TablePath nonExistentTable = TablePath.of("non_existent_db", "non_existent_table");
+ MetadataRequest request =
+ newMetadataRequest(Collections.singletonList(nonExistentTable)).setCacheOnly(true);
+
+ // Execute request and verify TableNotExistException is thrown
+ // This demonstrates that cache_only doesn't bypass table existence validation
+ assertThatThrownBy(() -> gateway.metadata(request).get())
+ .isInstanceOf(ExecutionException.class)
+ .hasMessageContaining("does not exist");
+ }
+
+ @Test
+ void testCacheOnlyFalse_NonExistentTable_ShouldNotThrowCacheMissException() {
+ AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ // Create request for a non-existent table with cache_only = false (default)
+ TablePath nonExistentTable = TablePath.of("non_existent_db", "non_existent_table");
+ MetadataRequest request = newMetadataRequest(Collections.singletonList(nonExistentTable));
+
+ // Execute request - should not throw CacheMissException but may throw other exceptions
+ // for non-existent table (which is expected behavior)
+ assertThatThrownBy(() -> gateway.metadata(request).get())
+ .isInstanceOf(ExecutionException.class);
+
+ // Verify that the exception is NOT CacheMissException
+ try {
+ gateway.metadata(request).get();
+ } catch (ExecutionException | InterruptedException e) {
+ assertThat(e.getCause()).isNotInstanceOf(CacheMissException.class);
+ }
+ }
+
+ @Test
+ void testCacheOnlyTrue_ExistingTable_ShouldReturnFromCache() throws Exception {
+ // First create a table to populate cache
+ long tableId =
+ createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
+ TableBucket tb = new TableBucket(tableId, 0);
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
+
+ // Wait for metadata to be available in all gateways
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
+
+ AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ // Create request for existing table with cache_only = true
+ MetadataRequest request =
+ newMetadataRequest(Collections.singletonList(DATA1_TABLE_PATH)).setCacheOnly(true);
+
+ // Execute request - should succeed and return metadata from cache
+ MetadataResponse response = gateway.metadata(request).get();
+ assertThat(response).isNotNull();
+ assertThat(response.hasCoordinatorServer()).isTrue();
+ assertThat(response.getTabletServersCount()).isGreaterThan(0);
+ }
+
+ @Test
+ void testCacheOnlyTrue_EmptyRequest_ShouldSucceed() throws Exception {
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
+
+ AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ // Create empty request with cache_only = true
+ MetadataRequest request = new MetadataRequest().setCacheOnly(true);
+
+ // Execute request - should succeed without any cache misses
+ MetadataResponse response = gateway.metadata(request).get();
+ assertThat(response).isNotNull();
+ // Empty request should return coordinator and tablet servers info
+ assertThat(response.hasCoordinatorServer()).isTrue();
+ assertThat(response.getTabletServersCount()).isGreaterThan(0);
+ }
+
+ @Test
+ void testCacheOnlyTrue_MultipleTables_MixedCacheStatus_ShouldThrowCacheMissException()
+ throws Exception {
+ // Create one table to ensure it's in cache
+ long tableId =
+ createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
+ TableBucket tb = new TableBucket(tableId, 0);
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
+
+ AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ // Create request with both existing table (in cache) and non-existent table (not in cache)
+ List<TablePath> tablePaths =
+ Arrays.asList(
+ DATA1_TABLE_PATH, // exists in cache
+ TablePath.of("non_existent_db", "non_existent_table") // not in cache
+ );
+ MetadataRequest request = newMetadataRequest(tablePaths).setCacheOnly(true);
+
+ // Execute request and verify that an exception is thrown
+ // Note: With existing + non-existent tables, TableNotExistException is thrown first
+ // for the non-existent table, rather than CacheMissException
+ assertThatThrownBy(() -> gateway.metadata(request).get())
+ .isInstanceOf(ExecutionException.class)
+ .hasMessageContaining("does not exist");
+ }
+
+ @Test
+ void testCacheOnlyTrue_RequestPartitionIds_CacheMiss_ShouldThrowCacheMissException()
+ throws Exception {
+ // First create a table
+ long tableId =
+ createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
+ TableBucket tb = new TableBucket(tableId, 0);
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
+
+ AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ // Create request with cache_only = true, but requesting non-existent partition ID
+ // This should trigger CacheMissException since the partition ID doesn't exist in cache
+ MetadataRequest request =
+ newMetadataRequest(Collections.singletonList(DATA1_TABLE_PATH)).setCacheOnly(true);
+ request.addPartitionsId(99999L); // Non-existent partition ID
+
+ // Execute request and verify CacheMissException is thrown
+ assertThatThrownBy(() -> gateway.metadata(request).get())
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(CacheMissException.class)
+ .hasMessageContaining("is not available in local cache");
+ }
}
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 9c15121e3..ded494026 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
@@ -21,6 +21,7 @@ import org.apache.fluss.cluster.Endpoint;
import org.apache.fluss.cluster.TabletServerInfo;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
@@ -28,6 +29,8 @@ import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePartition;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.entity.RegisterTableBucketLeadAndIsrInfo;
+import org.apache.fluss.server.metadata.BucketMetadata;
+import org.apache.fluss.server.metadata.PartitionMetadata;
import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.BucketSnapshot;
import org.apache.fluss.server.zk.data.CoordinatorAddress;
@@ -56,10 +59,12 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import static org.apache.fluss.server.utils.TableAssignmentUtils.generateAssignment;
@@ -576,4 +581,263 @@ class ZooKeeperClientTest {
.isEqualTo("zookeeper2");
}
}
+
+ @Test
+ void testGetTableMetadataFromZkAsync() throws Exception {
+ // Prepare test data using TestData constants
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ long tableId = 1001L;
+
+ // Create table assignment
+ TableAssignment tableAssignment =
+ TableAssignment.builder()
+ .add(0, BucketAssignment.of(1, 2, 3))
+ .add(1, BucketAssignment.of(2, 3, 4))
+ .build();
+
+ zookeeperClient.registerTableAssignment(tableId, tableAssignment);
+
+ // Create leader and isr for buckets
+ TableBucket tableBucket0 = new TableBucket(tableId, 0);
+ TableBucket tableBucket1 = new TableBucket(tableId, 1);
+
+ LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3, 4), 200, 2000);
+
+ zookeeperClient.registerLeaderAndIsr(tableBucket0, leaderAndIsr0);
+ zookeeperClient.registerLeaderAndIsr(tableBucket1, leaderAndIsr1);
+
+ // Test getTableMetadataFromZkAsync
+ CompletableFuture<List<BucketMetadata>> future =
+ zookeeperClient.getTableMetadataFromZkAsync(tablePath, tableId, false);
+
+ List<BucketMetadata> bucketMetadataList = future.get();
+
+ assertThat(bucketMetadataList).hasSize(2);
+
+ // Verify bucket metadata
+ Map<Integer, BucketMetadata> bucketMap =
+ bucketMetadataList.stream()
+ .collect(Collectors.toMap(BucketMetadata::getBucketId, b -> b));
+
+ BucketMetadata bucket0Metadata = bucketMap.get(0);
+ assertThat(extractLeaderFromBucketMetadata(bucket0Metadata)).isEqualTo(1);
+ assertThat(bucket0Metadata.getReplicas()).containsExactly(1, 2, 3);
+
+ BucketMetadata bucket1Metadata = bucketMap.get(1);
+ assertThat(extractLeaderFromBucketMetadata(bucket1Metadata)).isEqualTo(2);
+ assertThat(bucket1Metadata.getReplicas()).containsExactly(2, 3, 4);
+ }
+
+ @Test
+ void testGetPartitionMetadataFromZkAsync() throws Exception {
+ // Prepare test data
+ TablePath tablePath = TablePath.of("test_db", "test_partition_table");
+ long tableId = 2001L;
+ long partitionId = 1L;
+ String partitionName = "p1";
+
+ long currentMillis = System.currentTimeMillis();
+ TableRegistration tableReg =
+ createTestTableRegistration(tableId, "partition table", currentMillis);
+ zookeeperClient.registerTable(tablePath, tableReg);
+
+ // Create partition assignment
+ Map<Integer, BucketAssignment> bucketAssignments = new HashMap<>();
+ bucketAssignments.put(0, BucketAssignment.of(1, 2));
+ bucketAssignments.put(1, BucketAssignment.of(2, 3));
+ PartitionAssignment partitionAssignment =
+ new PartitionAssignment(tableId, bucketAssignments);
+
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId, partitionName, partitionAssignment, tablePath, tableId);
+
+ // Create leader and isr for partition buckets
+ TableBucket partitionBucket0 = new TableBucket(tableId, partitionId, 0);
+ TableBucket partitionBucket1 = new TableBucket(tableId, partitionId, 1);
+
+ LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000);
+
+ zookeeperClient.registerLeaderAndIsr(partitionBucket0, leaderAndIsr0);
+ zookeeperClient.registerLeaderAndIsr(partitionBucket1, leaderAndIsr1);
+
+ // Test getPartitionMetadataFromZkAsync
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(tablePath, partitionName);
+ CompletableFuture<PartitionMetadata> future =
+ zookeeperClient.getPartitionMetadataFromZkAsync(partitionPath);
+
+ PartitionMetadata partitionMetadata = future.get();
+
+ assertThat(partitionMetadata.getTableId()).isEqualTo(tableId);
+ assertThat(partitionMetadata.getPartitionName()).isEqualTo(partitionName);
+ assertThat(partitionMetadata.getPartitionId()).isEqualTo(partitionId);
+
+ List<BucketMetadata> bucketMetadataList = partitionMetadata.getBucketMetadataList();
+ assertThat(bucketMetadataList).hasSize(2);
+
+ Map<Integer, BucketMetadata> bucketMap =
+ bucketMetadataList.stream()
+ .collect(Collectors.toMap(BucketMetadata::getBucketId, b -> b));
+
+ BucketMetadata bucket0Metadata = bucketMap.get(0);
+ assertThat(extractLeaderFromBucketMetadata(bucket0Metadata)).isEqualTo(1);
+ assertThat(bucket0Metadata.getReplicas()).containsExactly(1, 2);
+
+ BucketMetadata bucket1Metadata = bucketMap.get(1);
+ assertThat(extractLeaderFromBucketMetadata(bucket1Metadata)).isEqualTo(2);
+ assertThat(bucket1Metadata.getReplicas()).containsExactly(2, 3);
+ }
+
+ @Test
+ void testBatchGetPartitionMetadataFromZkAsync() throws Exception {
+ // Prepare test data with multiple tables and partitions
+ TablePath tablePath1 = TablePath.of("test_db", "table1");
+ TablePath tablePath2 = TablePath.of("test_db", "table2");
+ long tableId1 = 3001L;
+ long tableId2 = 3002L;
+
+ long currentMillis = System.currentTimeMillis();
+ TableRegistration tableReg1 =
+ createTestTableRegistration(tableId1, "table1", currentMillis);
+ TableRegistration tableReg2 =
+ createTestTableRegistration(tableId2, "table2", currentMillis);
+
+ zookeeperClient.registerTable(tablePath1, tableReg1);
+ zookeeperClient.registerTable(tablePath2, tableReg2);
+
+ // Create partitions for table1
+ long partitionId1 = 11L;
+ long partitionId2 = 12L;
+ String partitionName1 = "p1";
+ String partitionName2 = "p2";
+
+ PartitionAssignment partitionAssignment1 =
+ new PartitionAssignment(
+ tableId1, Collections.singletonMap(0, BucketAssignment.of(1, 2)));
+ PartitionAssignment partitionAssignment2 =
+ new PartitionAssignment(
+ tableId1, Collections.singletonMap(1, BucketAssignment.of(2, 3)));
+
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId1, partitionName1, partitionAssignment1, tablePath1, tableId1);
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId2, partitionName2, partitionAssignment2, tablePath1, tableId1);
+
+ // Create partition for table2
+ long partitionId3 = 21L;
+ String partitionName3 = "p1";
+
+ PartitionAssignment partitionAssignment3 =
+ new PartitionAssignment(
+ tableId2, Collections.singletonMap(0, BucketAssignment.of(1, 3)));
+
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId3, partitionName3, partitionAssignment3, tablePath2, tableId2);
+
+ // Create leader and isr for all partition buckets
+ TableBucket bucket1 = new TableBucket(tableId1, partitionId1, 0);
+ TableBucket bucket2 = new TableBucket(tableId1, partitionId2, 1);
+ TableBucket bucket3 = new TableBucket(tableId2, partitionId3, 0);
+
+ zookeeperClient.registerLeaderAndIsr(
+ bucket1, new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000));
+ zookeeperClient.registerLeaderAndIsr(
+ bucket2, new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000));
+ zookeeperClient.registerLeaderAndIsr(
+ bucket3, new LeaderAndIsr(1, 30, Arrays.asList(1, 3), 300, 3000));
+
+ // Test batchGetPartitionMetadataFromZkAsync
+ List<TablePath> tablePaths = Arrays.asList(tablePath1, tablePath2);
+ Set<Long> partitionIdSet =
+ new HashSet<>(Arrays.asList(partitionId1, partitionId2, partitionId3));
+
+ CompletableFuture<List<PartitionMetadata>> future =
+ zookeeperClient.batchGetPartitionMetadataFromZkAsync(tablePaths, partitionIdSet);
+
+ List<PartitionMetadata> partitionMetadataList = future.get();
+
+ assertThat(partitionMetadataList).hasSize(3);
+
+ // Verify partition metadata
+ Map<Long, PartitionMetadata> partitionMap =
+ partitionMetadataList.stream()
+ .collect(Collectors.toMap(PartitionMetadata::getPartitionId, p -> p));
+
+ // Verify partition 1
+ PartitionMetadata partition1Metadata = partitionMap.get(partitionId1);
+ assertThat(partition1Metadata.getTableId()).isEqualTo(tableId1);
+ assertThat(partition1Metadata.getPartitionName()).isEqualTo(partitionName1);
+ assertThat(partition1Metadata.getBucketMetadataList()).hasSize(1);
+ assertThat(partition1Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(0);
+ assertThat(
+ extractLeaderFromBucketMetadata(
+ partition1Metadata.getBucketMetadataList().get(0)))
+ .isEqualTo(1);
+
+ // Verify partition 2
+ PartitionMetadata partition2Metadata = partitionMap.get(partitionId2);
+ assertThat(partition2Metadata.getTableId()).isEqualTo(tableId1);
+ assertThat(partition2Metadata.getPartitionName()).isEqualTo(partitionName2);
+ assertThat(partition2Metadata.getBucketMetadataList()).hasSize(1);
+ assertThat(partition2Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(1);
+ assertThat(
+ extractLeaderFromBucketMetadata(
+ partition2Metadata.getBucketMetadataList().get(0)))
+ .isEqualTo(2);
+
+ // Verify partition 3
+ PartitionMetadata partition3Metadata = partitionMap.get(partitionId3);
+ assertThat(partition3Metadata.getTableId()).isEqualTo(tableId2);
+ assertThat(partition3Metadata.getPartitionName()).isEqualTo(partitionName3);
+ assertThat(partition3Metadata.getBucketMetadataList()).hasSize(1);
+ assertThat(partition3Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(0);
+ assertThat(
+ extractLeaderFromBucketMetadata(
+ partition3Metadata.getBucketMetadataList().get(0)))
+ .isEqualTo(1);
+ }
+
+ @Test
+ void testBatchGetPartitionMetadataFromZkAsyncWithNonExistentPartition() {
+ // Test error handling when partition doesn't exist
+ TablePath tablePath = TablePath.of("test_db", "table1");
+ Set<Long> nonExistentPartitionIds = new HashSet<>(Arrays.asList(999L, 1000L));
+
+ CompletableFuture<List<PartitionMetadata>> future =
+ zookeeperClient.batchGetPartitionMetadataFromZkAsync(
+ Collections.singletonList(tablePath), nonExistentPartitionIds);
+
+ assertThatThrownBy(future::get)
+ .hasCauseInstanceOf(org.apache.fluss.exception.PartitionNotExistException.class);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helper methods for test data creation
+ // --------------------------------------------------------------------------------------------
+
+ private TableRegistration createTestTableRegistration(
+ long tableId, String comment, long currentMillis) {
+ Map<String, String> options = new HashMap<>();
+ options.put("option-1", "100");
+ return new TableRegistration(
+ tableId,
+ comment,
+ Arrays.asList("a", "b"),
+ new TableDescriptor.TableDistribution(3, Collections.singletonList("a")),
+ options,
+ Collections.emptyMap(),
+ currentMillis,
+ currentMillis);
+ }
+
+ /**
+ * Helper method to extract leader ID from BucketMetadata, handling OptionalInt. Returns null if
+ * leader is not present.
+ */
+ private static Integer extractLeaderFromBucketMetadata(BucketMetadata bucketMetadata) {
+ return bucketMetadata.getLeaderId().isPresent()
+ ? bucketMetadata.getLeaderId().getAsInt()
+ : null;
+ }
}
(fluss) 02/02: [server] Use unified batch-concurrent implementation for ZooKeeper operations and remove redundant code
Posted by ja...@apache.org.
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
commit 4345663fe7d5d474d078260aef69363e0290cb91
Author: Jark Wu <ja...@apache.org>
AuthorDate: Thu Sep 25 09:04:40 2025 +0800
[server] Use unified batch-concurrent implementation for ZooKeeper operations and remove redundant code
This commit replaces all single-operation async calls
with a unified batch-concurrent implementation. This eliminates the need
to maintain three separate code paths (batch concurrent, single async,
and sync), significantly reducing code duplication in ZooKeeperClient.
Going forward, all sync ZooKeeper operations will be replaced by the batch-concurrent
approach, which improves coordinator performance through better I/O
utilization and reduced latency.
Additionally, the `cacheOnly` flag has been removed from the current
metadata request. Support for cache-only semantics will be reintroduced
in a future `MetadataRequestV2` with a cleaner design.
---
.../client/table/AutoPartitionedTableITCase.java | 2 +-
.../apache/fluss/exception/CacheMissException.java | 36 -
.../org.junit.jupiter.api.extension.Extension | 19 +
.../src/test/resources/log4j2-test.properties | 32 +
.../org.junit.jupiter.api.extension.Extension | 19 +
.../src/test/resources/log4j2-test.properties | 32 +
.../java/org/apache/fluss/rpc/protocol/Errors.java | 4 -
fluss-rpc/src/main/proto/FlussApi.proto | 4 -
.../org/apache/fluss/server/RpcServiceBase.java | 238 ++---
.../server/coordinator/CoordinatorService.java | 24 +-
.../fluss/server/coordinator/MetadataManager.java | 10 +-
.../metadata/CoordinatorMetadataProvider.java | 32 +-
.../fluss/server/metadata/MetadataProvider.java | 29 +-
.../server/metadata/TabletServerMetadataCache.java | 5 +-
.../metadata/TabletServerMetadataProvider.java | 57 +-
.../server/metadata/ZkBasedMetadataProvider.java | 124 +++
.../apache/fluss/server/tablet/TabletServer.java | 2 +-
.../apache/fluss/server/tablet/TabletService.java | 21 +-
.../apache/fluss/server/zk/ZooKeeperClient.java | 1050 +++++++-------------
.../fluss/server/zk/ZooKeeperMetadataUtils.java | 478 ---------
.../metadata/TabletServerMetadataCacheTest.java | 3 +-
.../metadata/ZkBasedMetadataProviderTest.java | 349 +++++++
.../fluss/server/replica/ReplicaTestBase.java | 3 +-
.../replica/fetcher/ReplicaFetcherThreadTest.java | 2 +-
.../fluss/server/tablet/TabletServiceITCase.java | 139 ---
.../fluss/server/zk/ZooKeeperClientTest.java | 264 -----
26 files changed, 1102 insertions(+), 1876 deletions(-)
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/AutoPartitionedTableITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/AutoPartitionedTableITCase.java
index c6280d09e..6ae9ed7be 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/table/AutoPartitionedTableITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/table/AutoPartitionedTableITCase.java
@@ -327,7 +327,7 @@ class AutoPartitionedTableITCase extends ClientToServerITCaseBase {
LogScanner logScanner = table.newScan().createLogScanner();
assertThatThrownBy(() -> logScanner.subscribe(100L, 0, 0))
.isInstanceOf(PartitionNotExistException.class)
- .hasMessageContaining("Partition not exist for partition ids: [100]");
+ .hasMessageContaining("The partition id '100' does not exist");
logScanner.close();
// todo: test the case that client produce to a partition to a server, but
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java b/fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java
deleted file mode 100644
index e9ff03373..000000000
--- a/fluss-common/src/main/java/org/apache/fluss/exception/CacheMissException.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.fluss.exception;
-
-/**
- * Exception thrown when requested metadata is not available in the local cache and cache_only is
- * enabled.
- *
- * @since 0.8
- */
-public class CacheMissException extends ApiException {
- private static final long serialVersionUID = 1L;
-
- public CacheMissException(String message) {
- super(message);
- }
-
- public CacheMissException(String message, Throwable cause) {
- super(message, cause);
- }
-}
diff --git a/fluss-flink/fluss-flink-1.20/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/fluss-flink/fluss-flink-1.20/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
new file mode 100644
index 000000000..ca0e907f6
--- /dev/null
+++ b/fluss-flink/fluss-flink-1.20/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.fluss.testutils.common.TestLoggerExtension
\ No newline at end of file
diff --git a/fluss-flink/fluss-flink-1.20/src/test/resources/log4j2-test.properties b/fluss-flink/fluss-flink-1.20/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..12b05f186
--- /dev/null
+++ b/fluss-flink/fluss-flink-1.20/src/test/resources/log4j2-test.properties
@@ -0,0 +1,32 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
+# suppress the duplicated logger extension
+logger.flink.name = org.apache.flink.util.TestLoggerExtension
+logger.flink.level = OFF
diff --git a/fluss-flink/fluss-flink-2.1/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/fluss-flink/fluss-flink-2.1/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
new file mode 100644
index 000000000..ca0e907f6
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.1/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.fluss.testutils.common.TestLoggerExtension
\ No newline at end of file
diff --git a/fluss-flink/fluss-flink-2.1/src/test/resources/log4j2-test.properties b/fluss-flink/fluss-flink-2.1/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..12b05f186
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.1/src/test/resources/log4j2-test.properties
@@ -0,0 +1,32 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
+# suppress the duplicated logger extension
+logger.flink.name = org.apache.flink.util.TestLoggerExtension
+logger.flink.level = OFF
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
index e3b5007a3..3b899baee 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
@@ -20,7 +20,6 @@ package org.apache.fluss.rpc.protocol;
import org.apache.fluss.exception.ApiException;
import org.apache.fluss.exception.AuthenticationException;
import org.apache.fluss.exception.AuthorizationException;
-import org.apache.fluss.exception.CacheMissException;
import org.apache.fluss.exception.CorruptMessageException;
import org.apache.fluss.exception.CorruptRecordException;
import org.apache.fluss.exception.DatabaseAlreadyExistException;
@@ -225,9 +224,6 @@ public enum Errors {
55,
"The new ISR contains at least one ineligible replica.",
IneligibleReplicaException::new),
- CACHE_MISS(
- 56, "The requested metadata is not available in local cache.", CacheMissException::new);
- IneligibleReplicaException::new),
INVALID_ALTER_TABLE_EXCEPTION(
56, "The alter table is invalid.", InvalidAlterTableException::new);
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 3c6b8b358..46db51849 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -176,10 +176,6 @@ message MetadataRequest {
// metadata request
// todo: we won't need the assumption after we introduce metadata cache in server
repeated int64 partitions_id = 3 [packed = true];
-
- // if cache_only is true, return immediately if metadata is not in local cache
- // instead of waiting for ZooKeeper access to complete
- optional bool cache_only = 4;
}
message MetadataResponse {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
index 77983c863..d862813c5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java
@@ -17,8 +17,8 @@
package org.apache.fluss.server;
+import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.cluster.ServerType;
-import org.apache.fluss.exception.CacheMissException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.KvSnapshotNotExistException;
import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
@@ -105,7 +105,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclFilter;
@@ -463,171 +462,104 @@ public abstract class RpcServiceBase extends RpcGatewayService implements AdminR
}
}
- protected void processMetadataRequest(
+ protected MetadataResponse processMetadataRequest(
MetadataRequest request,
String listenerName,
Session session,
Authorizer authorizer,
ServerMetadataCache metadataCache,
- MetadataProvider functionProvider,
- CompletableFuture<MetadataResponse> responseFuture) {
- List<PbTablePath> pbTablePaths = request.getTablePathsList();
- List<CompletableFuture<TableMetadata>> pendingTableMetadata = new ArrayList<>();
- List<TablePath> tablePaths = new ArrayList<>(pbTablePaths.size());
- // result collections - using ConcurrentLinkedQueue for thread safety
- ConcurrentLinkedQueue<TableMetadata> tableMetadata = new ConcurrentLinkedQueue<>();
- ConcurrentLinkedQueue<PartitionMetadata> partitionMetadata = new ConcurrentLinkedQueue<>();
- boolean cacheOnly = request.hasCacheOnly() && request.isCacheOnly();
- List<String> cacheMissMessages = new ArrayList<>();
-
- try {
- for (PbTablePath pbTablePath : pbTablePaths) {
- if (authorizer == null
- || authorizer.isAuthorized(
- session,
- OperationType.DESCRIBE,
- Resource.table(
- pbTablePath.getDatabaseName(),
- pbTablePath.getTableName()))) {
- TablePath tablePath = toTablePath(pbTablePath);
- Optional<TableMetadata> tableMetadataFromCacheOpt =
- functionProvider.getTableMetadataFromCache(tablePath);
- if (!tableMetadataFromCacheOpt.isPresent()) {
- if (cacheOnly) {
- cacheMissMessages.add(
- String.format(
- "Table metadata for table %s is not available in local cache",
- tablePath));
- }
- // Always add ZK query to fill cache, even for cache_only requests
- pendingTableMetadata.add(
- functionProvider
- .getTableMetadataFromZk(tablePath)
- .thenApply(
- meta -> {
- tableMetadata.add(meta);
- return meta;
- }));
- } else {
- tableMetadata.add(tableMetadataFromCacheOpt.get());
- }
- tablePaths.add(tablePath);
- }
+ MetadataProvider metadataProvider) {
+ List<TablePath> authorizedTables = new ArrayList<>();
+ for (PbTablePath pbTablePath : request.getTablePathsList()) {
+ if (authorizer == null
+ || authorizer.isAuthorized(
+ session,
+ OperationType.DESCRIBE,
+ Resource.table(
+ pbTablePath.getDatabaseName(), pbTablePath.getTableName()))) {
+ authorizedTables.add(ServerRpcMessageUtils.toTablePath(pbTablePath));
}
-
- List<CompletableFuture<PartitionMetadata>> pendingPartitionMetadata = new ArrayList<>();
- CompletableFuture<List<PartitionMetadata>> pendingPartitionMetadataMissingIdsPartition =
- CompletableFuture.completedFuture(new ArrayList<>());
-
- List<PhysicalTablePath> partitions =
- request.getPartitionsPathsList().stream()
- .map(ServerRpcMessageUtils::toPhysicalTablePath)
- .collect(Collectors.toList());
- long[] partitionIds = request.getPartitionsIds();
- Set<Long> partitionIdsNotExistsInCache = new HashSet<>();
- for (long partitionId : partitionIds) {
- Optional<PhysicalTablePath> physicalTablePath =
- functionProvider.getPhysicalTablePathFromCache(partitionId);
- if (physicalTablePath.isPresent()) {
- partitions.add(physicalTablePath.get());
- } else {
- partitionIdsNotExistsInCache.add(partitionId);
- }
+ }
+ List<TableMetadata> tablesMetadata = new ArrayList<>();
+ List<TablePath> unknownTables = new ArrayList<>();
+ for (TablePath tablePath : authorizedTables) {
+ Optional<TableMetadata> metadataFromCache =
+ metadataProvider.getTableMetadataFromCache(tablePath);
+ if (metadataFromCache.isPresent()) {
+ tablesMetadata.add(metadataFromCache.get());
+ } else {
+ unknownTables.add(tablePath);
}
+ }
+ // fetch unknown table metadata from ZK
+ tablesMetadata.addAll(metadataProvider.getTablesMetadataFromZK(unknownTables));
+
+ // handle partition ids request
+ Set<PhysicalTablePath> partitionPaths =
+ request.getPartitionsPathsList().stream()
+ .map(ServerRpcMessageUtils::toPhysicalTablePath)
+ .collect(Collectors.toSet());
+ long[] partitionIds = request.getPartitionsIds();
+ List<Long> partitionIdsNotExistsInCache = new ArrayList<>();
+ for (long partitionId : partitionIds) {
+ Optional<PhysicalTablePath> physicalTablePath =
+ metadataProvider.getPhysicalTablePathFromCache(partitionId);
+ if (physicalTablePath.isPresent()) {
+ partitionPaths.add(physicalTablePath.get());
+ } else {
+ partitionIdsNotExistsInCache.add(partitionId);
+ }
+ }
- if (!partitionIdsNotExistsInCache.isEmpty()) {
- if (cacheOnly) {
- cacheMissMessages.add(
+ if (!partitionIdsNotExistsInCache.isEmpty()) {
+ Map<Long, PhysicalTablePath> partitionIdAndPaths;
+ try {
+ partitionIdAndPaths = zkClient.getPartitionIdAndPaths(authorizedTables);
+ } catch (Exception e) {
+ throw new FlussRuntimeException("Failed to get partition paths from ZK.", e);
+ }
+ for (long partitionId : partitionIdsNotExistsInCache) {
+ if (partitionIdAndPaths.containsKey(partitionId)) {
+ partitionPaths.add(partitionIdAndPaths.get(partitionId));
+ } else {
+ throw new PartitionNotExistException(
String.format(
- "Partition metadata for partition IDs %s is not available in local cache",
- partitionIdsNotExistsInCache));
+ "The partition id '%d' does not exist or you don't have permission to access it.",
+ partitionId));
}
- // Always execute ZK query to fill cache, even for cache_only requests
- pendingPartitionMetadataMissingIdsPartition =
- zkClient.batchGetPartitionMetadataFromZkAsync(
- tablePaths, partitionIdsNotExistsInCache)
- .thenApply(
- batchMetadata -> {
- partitionMetadata.addAll(batchMetadata);
- return batchMetadata;
- });
}
+ }
- for (PhysicalTablePath partitionPath : partitions) {
- if (authorizer == null
- || authorizer.isAuthorized(
- session,
- OperationType.DESCRIBE,
- Resource.table(
- partitionPath.getDatabaseName(),
- partitionPath.getTableName()))) {
- Optional<PartitionMetadata> partitionMetadataOpt =
- functionProvider.getPartitionMetadataFromCache(partitionPath);
- if (!partitionMetadataOpt.isPresent()) {
- if (cacheOnly) {
- cacheMissMessages.add(
- String.format(
- "Partition metadata for partition %s is not available in local cache",
- partitionPath));
- }
- // Always execute ZK query to fill cache, even for cache_only requests
- pendingPartitionMetadata.add(
- functionProvider
- .getPartitionMetadataFromZk(partitionPath)
- .thenApply(
- meta -> {
- partitionMetadata.add(meta);
- return meta;
- }));
- } else {
- partitionMetadata.add(partitionMetadataOpt.get());
- }
- }
+ // collect partition metadata
+ List<PhysicalTablePath> authorizedPartitions = new ArrayList<>();
+ for (PhysicalTablePath path : partitionPaths) {
+ if (authorizer == null
+ || authorizer.isAuthorized(
+ session,
+ OperationType.DESCRIBE,
+ Resource.table(path.getDatabaseName(), path.getTableName()))) {
+ authorizedPartitions.add(path);
}
-
- // Check for cache misses and return error immediately for cache_only requests
- if (cacheOnly && !cacheMissMessages.isEmpty()) {
- String combinedMessage = String.join("; ", cacheMissMessages);
- responseFuture.completeExceptionally(new CacheMissException(combinedMessage));
- // Continue with ZK queries to fill cache asynchronously, but don't wait for them
- return;
+ }
+ List<PartitionMetadata> partitionsMetadata = new ArrayList<>();
+ List<PhysicalTablePath> unknownPartitions = new ArrayList<>();
+ for (PhysicalTablePath partitionPath : authorizedPartitions) {
+ Optional<PartitionMetadata> metadataFromCache =
+ metadataProvider.getPartitionMetadataFromCache(partitionPath);
+ if (metadataFromCache.isPresent()) {
+ partitionsMetadata.add(metadataFromCache.get());
+ } else {
+ unknownPartitions.add(partitionPath);
}
-
- CompletableFuture<Void> allPendingTableMeta =
- CompletableFuture.allOf(pendingTableMetadata.toArray(new CompletableFuture[0]));
- CompletableFuture<Void> allPendingPartitionMeta =
- CompletableFuture.allOf(
- pendingPartitionMetadata.toArray(new CompletableFuture[0]));
-
- CompletableFuture<Void> allPendingRequest =
- CompletableFuture.allOf(
- allPendingTableMeta,
- allPendingPartitionMeta,
- pendingPartitionMetadataMissingIdsPartition);
-
- allPendingRequest.handle(
- (v, throwable) -> {
- if (throwable != null) {
- responseFuture.completeExceptionally(throwable);
- return null;
- }
- try {
- responseFuture.complete(
- buildMetadataResponse(
- metadataCache.getCoordinatorServer(listenerName),
- new HashSet<>(
- metadataCache
- .getAllAliveTabletServers(listenerName)
- .values()),
- new ArrayList<>(tableMetadata),
- new ArrayList<>(partitionMetadata)));
- } catch (Exception e) {
- responseFuture.completeExceptionally(e);
- }
- return null;
- });
- } catch (Exception e) {
- responseFuture.completeExceptionally(e);
}
+ // fetch unknown partition metadata from ZK
+ partitionsMetadata.addAll(metadataProvider.getPartitionsMetadataFromZK(unknownPartitions));
+
+ // build response
+ ServerNode coordinatorServer = metadataCache.getCoordinatorServer(listenerName);
+ Set<ServerNode> aliveTabletServers =
+ new HashSet<>(metadataCache.getAllAliveTabletServers(listenerName).values());
+ return buildMetadataResponse(
+ coordinatorServer, aliveTabletServers, tablesMetadata, partitionsMetadata);
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index 5d1b8a1e4..8b0ffba40 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -34,8 +34,6 @@ import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.DatabaseDescriptor;
import org.apache.fluss.metadata.PartitionSpec;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
-import org.apache.fluss.metadata.TableChange;
-import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
@@ -101,7 +99,6 @@ import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.PartitionAssignment;
import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
-import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.concurrent.FutureUtils;
@@ -111,10 +108,7 @@ import java.io.UncheckedIOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
import java.util.function.Supplier;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclBindingFilters;
@@ -453,26 +447,14 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
AccessContextEvent<MetadataResponse> metadataResponseAccessContextEvent =
new AccessContextEvent<>(
ctx -> {
- CompletableFuture<MetadataResponse> response =
- new CompletableFuture<>();
- processMetadataRequest(
+ return processMetadataRequest(
request,
listenerName,
session,
authorizer,
metadataCache,
- new CoordinatorMetadataProvider(zkClient, ctx, metadataManager),
- response);
- try {
- return response.get();
- } catch (Exception e) {
- Throwable strippedException =
- ExceptionUtils.stripException(e, ExecutionException.class);
- if (strippedException instanceof RuntimeException) {
- throw (RuntimeException) strippedException;
- }
- throw new RuntimeException(strippedException);
- }
+ new CoordinatorMetadataProvider(
+ zkClient, metadataManager, ctx));
});
eventManagerSupplier.get().put(metadataResponseAccessContextEvent);
return metadataResponseAccessContextEvent.getResultFuture();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
index 322c45103..f65814c65 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java
@@ -409,12 +409,20 @@ public class MetadataManager {
try {
Map<TablePath, TableRegistration> tablePath2TableRegistrations =
zookeeperClient.getTables(tablePaths);
+ // currently, we don't support schema evolution, so all schemas are version 1
+ Map<TablePath, SchemaInfo> tablePath2SchemaInfos =
+ zookeeperClient.getV1Schemas(tablePaths);
for (TablePath tablePath : tablePaths) {
if (!tablePath2TableRegistrations.containsKey(tablePath)) {
throw new TableNotExistException("Table '" + tablePath + "' does not exist.");
}
+ if (!tablePath2SchemaInfos.containsKey(tablePath)) {
+ throw new SchemaNotExistException(
+ "Schema for '" + tablePath + "' with schema_id=1 does not exist.");
+ }
TableRegistration tableReg = tablePath2TableRegistrations.get(tablePath);
- SchemaInfo schemaInfo = getLatestSchema(tablePath);
+ SchemaInfo schemaInfo = tablePath2SchemaInfos.get(tablePath);
+
result.put(
tablePath,
tableReg.toTableInfo(tablePath, schemaInfo, defaultTableLakeOptions));
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java
index 6127a4a03..f119c49db 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java
@@ -33,7 +33,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
/**
* Coordinator-side implementation of the MetadataProvider interface.
@@ -53,26 +52,21 @@ import java.util.concurrent.CompletableFuture;
* <p>This implementation is optimized for coordinator usage patterns where table and partition
* metadata is frequently accessed for making assignment decisions and handling client requests.
*/
-public class CoordinatorMetadataProvider implements MetadataProvider {
-
- private final ZooKeeperClient zkClient;
+public class CoordinatorMetadataProvider extends ZkBasedMetadataProvider {
private final CoordinatorContext ctx;
- private final MetadataManager metadataManager;
-
/**
* Creates a new CoordinatorMetadataProvider.
*
* @param zkClient the ZooKeeper client for accessing distributed metadata
- * @param ctx the coordinator context providing cached metadata and assignments
* @param metadataManager the metadata manager for table information
+ * @param ctx the coordinator context providing cached metadata and assignments
*/
public CoordinatorMetadataProvider(
- ZooKeeperClient zkClient, CoordinatorContext ctx, MetadataManager metadataManager) {
- this.zkClient = zkClient;
+ ZooKeeperClient zkClient, MetadataManager metadataManager, CoordinatorContext ctx) {
+ super(zkClient, metadataManager);
this.ctx = ctx;
- this.metadataManager = metadataManager;
}
@Override
@@ -88,19 +82,6 @@ public class CoordinatorMetadataProvider implements MetadataProvider {
return Optional.of(new TableMetadata(tableInfo, bucketMetadataList));
}
- @Override
- public CompletableFuture<TableMetadata> getTableMetadataFromZk(TablePath tablePath) {
- TableInfo tableInfo = metadataManager.getTable(tablePath);
- return ZooKeeperClient.getTableMetadataFromZkAsync(
- zkClient, tablePath, tableInfo.getTableId(), tableInfo.isPartitioned())
- .thenApply(bucketMetadata -> new TableMetadata(tableInfo, bucketMetadata));
- }
-
- @Override
- public Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionId) {
- return ctx.getPhysicalTablePath(partitionId);
- }
-
@Override
public Optional<PartitionMetadata> getPartitionMetadataFromCache(
PhysicalTablePath partitionPath) {
@@ -126,9 +107,8 @@ public class CoordinatorMetadataProvider implements MetadataProvider {
}
@Override
- public CompletableFuture<PartitionMetadata> getPartitionMetadataFromZk(
- PhysicalTablePath partitionPath) {
- return ZooKeeperClient.getPartitionMetadataFromZkAsync(partitionPath, zkClient);
+ public Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionId) {
+ return ctx.getPhysicalTablePath(partitionId);
}
/**
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java
index 12c25773e..bd7960f33 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/MetadataProvider.java
@@ -20,8 +20,9 @@ package org.apache.fluss.server.metadata;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TablePath;
+import java.util.Collection;
+import java.util.List;
import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
/**
* Interface for providing metadata information about tables and partitions from various sources.
@@ -39,12 +40,12 @@ public interface MetadataProvider {
Optional<TableMetadata> getTableMetadataFromCache(TablePath tablePath);
/**
- * Retrieves table metadata from ZooKeeper asynchronously.
+ * Retrieves partition metadata from local cache.
*
- * @param tablePath the path identifying the table
- * @return a CompletableFuture containing the table metadata from ZooKeeper
+ * @param physicalTablePath the physical path identifying the table partition
+ * @return an Optional containing the partition metadata if found in cache, empty otherwise
*/
- CompletableFuture<TableMetadata> getTableMetadataFromZk(TablePath tablePath);
+ Optional<PartitionMetadata> getPartitionMetadataFromCache(PhysicalTablePath physicalTablePath);
/**
* Retrieves the physical table path from local cache using partition ID.
@@ -55,19 +56,19 @@ public interface MetadataProvider {
Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionId);
/**
- * Retrieves partition metadata from local cache.
+ * Retrieves a batch of table metadata from ZooKeeper.
*
- * @param physicalTablePath the physical path identifying the table partition
- * @return an Optional containing the partition metadata if found in cache, empty otherwise
+ * @param tablePaths the path identifying the table
+ * @return a CompletableFuture containing the table metadata from ZooKeeper
*/
- Optional<PartitionMetadata> getPartitionMetadataFromCache(PhysicalTablePath physicalTablePath);
+ List<TableMetadata> getTablesMetadataFromZK(Collection<TablePath> tablePaths);
/**
- * Retrieves partition metadata from ZooKeeper asynchronously.
+ * Retrieves a batch of partition metadata from ZooKeeper.
*
- * @param physicalTablePath the physical path identifying the table partition
- * @return a CompletableFuture containing the partition metadata from ZooKeeper
+ * @param partitionPaths the physical path identifying the table partition
+ * @return an Optional containing the partition metadata if found in cache, empty otherwise
*/
- CompletableFuture<PartitionMetadata> getPartitionMetadataFromZk(
- PhysicalTablePath physicalTablePath);
+ List<PartitionMetadata> getPartitionsMetadataFromZK(
+ Collection<PhysicalTablePath> partitionPaths);
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
index 5479b523a..79a22621b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataCache.java
@@ -26,7 +26,6 @@ import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.coordinator.MetadataManager;
import org.apache.fluss.server.tablet.TabletServer;
-import org.apache.fluss.server.zk.ZooKeeperClient;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
@@ -65,12 +64,10 @@ public class TabletServerMetadataCache implements ServerMetadataCache {
private volatile ServerMetadataSnapshot serverMetadataSnapshot;
private final MetadataManager metadataManager;
- private final ZooKeeperClient zkClient;
- public TabletServerMetadataCache(MetadataManager metadataManager, ZooKeeperClient zkClient) {
+ public TabletServerMetadataCache(MetadataManager metadataManager) {
this.serverMetadataSnapshot = ServerMetadataSnapshot.empty();
this.metadataManager = metadataManager;
- this.zkClient = zkClient;
}
@Override
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java
index b52269ea2..72a2f665f 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/TabletServerMetadataProvider.java
@@ -18,13 +18,13 @@
package org.apache.fluss.server.metadata;
import org.apache.fluss.metadata.PhysicalTablePath;
-import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.coordinator.MetadataManager;
import org.apache.fluss.server.zk.ZooKeeperClient;
+import java.util.Collection;
+import java.util.List;
import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
/**
* Tablet server-side implementation of the MetadataProvider interface.
@@ -46,28 +46,23 @@ import java.util.concurrent.CompletableFuture;
* metadata for partition management, leader election participation, and client request handling,
* while maintaining consistency with the coordinator's view of metadata.
*/
-public class TabletServerMetadataProvider implements MetadataProvider {
-
- private final ZooKeeperClient zkClient;
+public class TabletServerMetadataProvider extends ZkBasedMetadataProvider {
private final TabletServerMetadataCache metadataCache;
- private final MetadataManager metadataManager;
-
/**
* Creates a new TabletServerMetadataProvider.
*
* @param zkClient the ZooKeeper client for accessing distributed metadata
- * @param metadataCache the local metadata cache for efficient metadata access
* @param metadataManager the metadata manager for table information
+ * @param metadataCache the local metadata cache for efficient metadata access
*/
public TabletServerMetadataProvider(
ZooKeeperClient zkClient,
- TabletServerMetadataCache metadataCache,
- MetadataManager metadataManager) {
- this.zkClient = zkClient;
+ MetadataManager metadataManager,
+ TabletServerMetadataCache metadataCache) {
+ super(zkClient, metadataManager);
this.metadataCache = metadataCache;
- this.metadataManager = metadataManager;
}
@Override
@@ -76,18 +71,9 @@ public class TabletServerMetadataProvider implements MetadataProvider {
}
@Override
- public CompletableFuture<TableMetadata> getTableMetadataFromZk(TablePath tablePath) {
- TableInfo tableInfo = metadataManager.getTable(tablePath);
- return ZooKeeperClient.getTableMetadataFromZkAsync(
- zkClient, tablePath, tableInfo.getTableId(), tableInfo.isPartitioned())
- .thenApply(
- bucketMetadataList -> {
- TableMetadata tableMetadata =
- new TableMetadata(tableInfo, bucketMetadataList);
- // Update local cache after successfully fetching from ZooKeeper
- metadataCache.updateTableMetadata(tableMetadata);
- return tableMetadata;
- });
+ public Optional<PartitionMetadata> getPartitionMetadataFromCache(
+ PhysicalTablePath physicalTablePath) {
+ return metadataCache.getPartitionMetadata(physicalTablePath);
}
@Override
@@ -96,20 +82,19 @@ public class TabletServerMetadataProvider implements MetadataProvider {
}
@Override
- public Optional<PartitionMetadata> getPartitionMetadataFromCache(
- PhysicalTablePath physicalTablePath) {
- return metadataCache.getPartitionMetadata(physicalTablePath);
+ public List<TableMetadata> getTablesMetadataFromZK(Collection<TablePath> tablePaths) {
+ List<TableMetadata> result = super.getTablesMetadataFromZK(tablePaths);
+ // Update local cache after successfully fetching from ZooKeeper
+ result.forEach(metadataCache::updateTableMetadata);
+ return result;
}
@Override
- public CompletableFuture<PartitionMetadata> getPartitionMetadataFromZk(
- PhysicalTablePath physicalTablePath) {
- return ZooKeeperClient.getPartitionMetadataFromZkAsync(physicalTablePath, zkClient)
- .thenApply(
- partitionMetadata -> {
- // Update local cache after successfully fetching from ZooKeeper
- metadataCache.updatePartitionMetadata(partitionMetadata);
- return partitionMetadata;
- });
+ public List<PartitionMetadata> getPartitionsMetadataFromZK(
+ Collection<PhysicalTablePath> partitionPaths) {
+ List<PartitionMetadata> result = super.getPartitionsMetadataFromZK(partitionPaths);
+ // Update local cache after successfully fetching from ZooKeeper
+ result.forEach(metadataCache::updatePartitionMetadata);
+ return result;
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/ZkBasedMetadataProvider.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ZkBasedMetadataProvider.java
new file mode 100644
index 000000000..e83241b65
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/ZkBasedMetadataProvider.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.metadata;
+
+import org.apache.fluss.exception.PartitionNotExistException;
+import org.apache.fluss.exception.TableNotExistException;
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePartition;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * An abstract implementation of {@link MetadataProvider} that provides common functionality for
+ * fetching table and partition metadata from ZooKeeper.
+ */
+public abstract class ZkBasedMetadataProvider implements MetadataProvider {
+
+ private final ZooKeeperClient zkClient;
+ private final MetadataManager metadataManager;
+
+ protected ZkBasedMetadataProvider(ZooKeeperClient zkClient, MetadataManager metadataManager) {
+ this.zkClient = zkClient;
+ this.metadataManager = metadataManager;
+ }
+
+ @Override
+ public List<TableMetadata> getTablesMetadataFromZK(Collection<TablePath> tablePaths) {
+ try {
+ Map<TablePath, TableInfo> path2Info = metadataManager.getTables(tablePaths);
+ Map<Long, TableInfo> id2Info =
+ path2Info.values().stream()
+ .collect(Collectors.toMap(TableInfo::getTableId, info -> info));
+ // Fetch all table metadata from ZooKeeper
+ Map<Long, List<BucketMetadata>> metadataForTables =
+ zkClient.getBucketMetadataForTables(id2Info.keySet());
+ // we should iterate on the input tablePaths to guarantee the result contains all tables
+ List<TableMetadata> result = new ArrayList<>();
+ for (TablePath path : tablePaths) {
+ TableInfo tableInfo = path2Info.get(path);
+ if (tableInfo == null) {
+ throw new TableNotExistException("Table '" + path + "' does not exist.");
+ }
+ // the metadata maybe null when the table assignment not generated yet.
+ List<BucketMetadata> metadata = metadataForTables.get(tableInfo.getTableId());
+ TableMetadata tableMetadata = new TableMetadata(tableInfo, metadata);
+ result.add(tableMetadata);
+ }
+ return result;
+ } catch (Exception e) {
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new RuntimeException("Failed to fetch table metadata from ZooKeeper", e);
+ }
+ }
+
+ @Override
+ public List<PartitionMetadata> getPartitionsMetadataFromZK(
+ Collection<PhysicalTablePath> partitionPaths) {
+ try {
+ Map<Long, String> partitionId2PartitionName = new HashMap<>();
+ Map<Long, Long> partitionId2TableId = new HashMap<>();
+ Map<PhysicalTablePath, TablePartition> partitionIds =
+ zkClient.getPartitionIds(partitionPaths);
+ for (PhysicalTablePath partitionPath : partitionPaths) {
+ TablePartition partition = partitionIds.get(partitionPath);
+ if (partition == null) {
+ throw new PartitionNotExistException(
+ "Table partition '" + partitionPath + "' does not exist.");
+ }
+ partitionId2PartitionName.put(
+ partition.getPartitionId(), checkNotNull(partitionPath.getPartitionName()));
+ partitionId2TableId.put(partition.getPartitionId(), partition.getTableId());
+ }
+ List<PartitionMetadata> result = new ArrayList<>();
+ // Fetch all table metadata from ZooKeeper
+ zkClient.getBucketMetadataForPartitions(partitionId2TableId.keySet())
+ .forEach(
+ (partitionId, bucketMetadataList) -> {
+ long tableId = partitionId2TableId.get(partitionId);
+ String partitionName = partitionId2PartitionName.get(partitionId);
+ PartitionMetadata partitionMetadata =
+ new PartitionMetadata(
+ tableId,
+ partitionName,
+ partitionId,
+ bucketMetadataList);
+ result.add(partitionMetadata);
+ });
+ return result;
+ } catch (Exception e) {
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new RuntimeException("Failed to fetch partition metadata from ZooKeeper", e);
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index 63121dbc3..3925cb96c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -195,7 +195,7 @@ public class TabletServer extends ServerBase {
this.zkClient = ZooKeeperUtils.startZookeeperClient(conf, this);
MetadataManager metadataManager = new MetadataManager(zkClient, conf);
- this.metadataCache = new TabletServerMetadataCache(metadataManager, zkClient);
+ this.metadataCache = new TabletServerMetadataCache(metadataManager);
this.scheduler = new FlussScheduler(conf.get(BACKGROUND_THREADS));
scheduler.startup();
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 4114b1ab4..a182b25a3 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
@@ -138,7 +138,7 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
this.replicaManager = replicaManager;
this.metadataCache = metadataCache;
this.metadataFunctionProvider =
- new TabletServerMetadataProvider(zkClient, metadataCache, metadataManager);
+ new TabletServerMetadataProvider(zkClient, metadataManager, metadataCache);
}
@Override
@@ -286,16 +286,15 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
@Override
public CompletableFuture<MetadataResponse> metadata(MetadataRequest request) {
- CompletableFuture<MetadataResponse> response = new CompletableFuture<>();
- processMetadataRequest(
- request,
- currentListenerName(),
- currentSession(),
- authorizer,
- metadataCache,
- metadataFunctionProvider,
- response);
- return response;
+ MetadataResponse metadataResponse =
+ processMetadataRequest(
+ request,
+ currentListenerName(),
+ currentSession(),
+ authorizer,
+ metadataCache,
+ metadataFunctionProvider);
+ return CompletableFuture.completedFuture(metadataResponse);
}
@Override
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 872b06d7b..0bac972f3 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
@@ -20,9 +20,6 @@ package org.apache.fluss.server.zk;
import org.apache.fluss.annotation.Internal;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
-import org.apache.fluss.exception.FlussRuntimeException;
-import org.apache.fluss.exception.PartitionNotExistException;
-import org.apache.fluss.exception.TableNotExistException;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
import org.apache.fluss.metadata.Schema;
@@ -36,14 +33,10 @@ import org.apache.fluss.security.acl.ResourceType;
import org.apache.fluss.server.authorizer.DefaultAuthorizer.VersionedAcls;
import org.apache.fluss.server.entity.RegisterTableBucketLeadAndIsrInfo;
import org.apache.fluss.server.metadata.BucketMetadata;
-import org.apache.fluss.server.metadata.PartitionMetadata;
import org.apache.fluss.server.zk.ZkAsyncRequest.ZkGetChildrenRequest;
import org.apache.fluss.server.zk.ZkAsyncRequest.ZkGetDataRequest;
import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetChildrenResponse;
import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetDataResponse;
-import org.apache.fluss.server.zk.ZooKeeperMetadataUtils.AssignmentInfo;
-import org.apache.fluss.server.zk.ZooKeeperMetadataUtils.PartitionPathInfo;
-import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.BucketSnapshot;
import org.apache.fluss.server.zk.data.CoordinatorAddress;
import org.apache.fluss.server.zk.data.DatabaseRegistration;
@@ -88,7 +81,6 @@ import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.CreateMode;
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.data.Stat;
import org.apache.fluss.utils.ExceptionUtils;
-import org.apache.fluss.utils.concurrent.FutureUtils;
import org.apache.fluss.utils.types.Tuple2;
import org.slf4j.Logger;
@@ -116,6 +108,7 @@ import java.util.concurrent.Semaphore;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static java.util.stream.Collectors.toMap;
import static org.apache.fluss.metadata.ResolvedPartitionSpec.fromPartitionName;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
@@ -129,6 +122,7 @@ public class ZooKeeperClient implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperClient.class);
public static final int UNKNOWN_VERSION = -2;
private static final int MAX_BATCH_SIZE = 1024;
+ private static final int DEFAULT_SCHEMA_ID = 1;
private final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
@@ -162,110 +156,6 @@ public class ZooKeeperClient implements AutoCloseable {
}
}
- /**
- * Send a pipelined sequence of requests and return a CompletableFuture for all their responses.
- *
- * <p>The watch flag on each outgoing request will be set if we've already registered a handler
- * for the path associated with the request.
- *
- * @param requests a sequence of requests to send
- * @param respCreator function to create response objects from curator events
- * @return CompletableFuture containing the responses for the requests
- */
- private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
- CompletableFuture<List<Resp>> handleRequestInBackgroundAsync(
- List<Req> requests, Function<CuratorEvent, Resp> respCreator) {
- if (requests == null || requests.isEmpty()) {
- return CompletableFuture.completedFuture(Collections.emptyList());
- }
-
- CompletableFuture<List<Resp>> future = new CompletableFuture<>();
- BackgroundCallback callback = getZkBackgroundCallback(requests, respCreator, future);
-
- try {
- for (Req request : requests) {
- try {
- inFlightRequests.acquire();
- if (request instanceof ZkGetDataRequest) {
- zkClient.getData().inBackground(callback).forPath(request.getPath());
-
- } else if (request instanceof ZkGetChildrenRequest) {
- zkClient.getChildren().inBackground(callback).forPath(request.getPath());
-
- } else {
- inFlightRequests.release();
- throw new IllegalArgumentException(
- "Unsupported request type: " + request.getClass());
- }
- } catch (Exception e) {
- inFlightRequests.release();
- throw e;
- }
- }
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
-
- return future;
- }
-
- @Nonnull
- private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
- BackgroundCallback getZkBackgroundCallback(
- List<Req> requests,
- Function<CuratorEvent, Resp> respCreator,
- CompletableFuture<List<Resp>> future) {
- ArrayBlockingQueue<Resp> responseQueue = new ArrayBlockingQueue<>(requests.size());
- CountDownLatch countDownLatch = new CountDownLatch(requests.size());
-
- // Complete future when all responses received
- return (client, event) -> {
- try {
- Resp response = respCreator.apply(event);
- responseQueue.add(response);
- inFlightRequests.release();
- countDownLatch.countDown();
-
- // Complete future when all responses received
- if (countDownLatch.getCount() == 0) {
- future.complete(new ArrayList<>(responseQueue));
- }
- } catch (Exception e) {
- inFlightRequests.release();
- future.completeExceptionally(e);
- }
- };
- }
-
- /**
- * Send a pipelined sequence of requests and wait for all of their responses synchronously in
- * background.
- *
- * <p>The watch flag on each outgoing request will be set if we've already registered a handler
- * for the path associated with the request.
- *
- * @param requests a sequence of requests to send and wait on.
- * @return the responses for the requests. If all requests have the same type, the responses
- * will have the respective response type.
- */
- private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
- List<Resp> handleRequestInBackground(
- List<Req> requests, Function<CuratorEvent, Resp> respCreator) throws Exception {
- try {
- return handleRequestInBackgroundAsync(requests, respCreator).get();
- } catch (ExecutionException e) {
- Throwable cause = ExceptionUtils.stripExecutionException(e);
- if (cause instanceof Exception) {
- throw (Exception) cause;
- } else {
- throw new Exception("Async request handling failed", cause);
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new Exception("Request handling was interrupted", e);
- }
- }
-
// --------------------------------------------------------------------------------------------
// Coordinator server
// --------------------------------------------------------------------------------------------
@@ -318,10 +208,10 @@ public class ZooKeeperClient implements AutoCloseable {
Map<String, Integer> path2IdMap =
Arrays.stream(tabletServerIds)
.boxed()
- .collect(Collectors.toMap(ServerIdZNode::path, id -> id));
+ .collect(toMap(ServerIdZNode::path, id -> id));
List<ZkGetDataResponse> responses = getDataInBackground(path2IdMap.keySet());
- return ZooKeeperMetadataUtils.processMultipleDataResponses(
+ return processGetDataResponses(
responses,
response -> path2IdMap.get(response.getPath()),
ServerIdZNode::decode,
@@ -351,68 +241,43 @@ public class ZooKeeperClient implements AutoCloseable {
/** Get the table assignment in ZK. */
public Optional<TableAssignment> getTableAssignment(long tableId) throws Exception {
- return FutureUtils.asyncToSync(() -> getTableAssignmentAsync(tableId), "table assignment");
- }
-
- /** Get the table assignment in ZK asynchronously. */
- public CompletableFuture<TableAssignment> getTableAssignmentAsync(long tableId) {
- String path = TableIdZNode.path(tableId);
-
- return getDataInBackgroundAsync(Collections.singleton(path))
- .thenApply(
- responses ->
- ZooKeeperMetadataUtils.processSingleDataResponse(
- responses,
- data -> data.length == 0 ? null : TableIdZNode.decode(data),
- path));
+ Optional<byte[]> bytes = getOrEmpty(TableIdZNode.path(tableId));
+ return bytes.map(
+ data ->
+ // we'll put a laketable node under TableIdZNode,
+ // so it won't be Optional#empty
+ // but will with a zero-length array
+ data.length == 0 ? null : TableIdZNode.decode(data));
}
/** Get the tables assignments in ZK. */
- public Map<Long, TableAssignment> getTablesAssignments(List<Long> tableIds) throws Exception {
+ public Map<Long, TableAssignment> getTablesAssignments(Collection<Long> tableIds)
+ throws Exception {
Map<String, Long> path2TableIdMap =
- tableIds.stream().collect(Collectors.toMap(TableIdZNode::path, id -> id));
+ tableIds.stream().collect(toMap(TableIdZNode::path, id -> id));
List<ZkGetDataResponse> responses = getDataInBackground(path2TableIdMap.keySet());
- return ZooKeeperMetadataUtils.processMultipleDataResponsesWithCondition(
+ return processGetDataResponses(
responses,
response -> path2TableIdMap.get(response.getPath()),
TableIdZNode::decode,
- response -> response.getData().length > 0,
"table assignment");
}
/** Get the partition assignment in ZK. */
public Optional<PartitionAssignment> getPartitionAssignment(long partitionId) throws Exception {
- return FutureUtils.asyncToSync(
- () -> getPartitionAssignmentAsync(partitionId),
- assignment ->
- (PartitionAssignment)
- assignment, // PartitionAssignment extends TableAssignment
- "partition assignment");
- }
-
- /** Get the partition assignment in ZK asynchronously. */
- private CompletableFuture<TableAssignment> getPartitionAssignmentAsync(long partitionId) {
- String path = PartitionIdZNode.path(partitionId);
-
- return getDataInBackgroundAsync(Collections.singleton(path))
- .thenApply(
- responses ->
- ZooKeeperMetadataUtils.processSingleDataResponse(
- responses,
- PartitionIdZNode::decode, // PartitionAssignment extends
- // TableAssignment
- path));
+ Optional<byte[]> bytes = getOrEmpty(PartitionIdZNode.path(partitionId));
+ return bytes.map(PartitionIdZNode::decode);
}
/** Get the partitions assignments in ZK. */
- public Map<Long, PartitionAssignment> getPartitionsAssignments(List<Long> partitionIds)
+ public Map<Long, PartitionAssignment> getPartitionsAssignments(Collection<Long> partitionIds)
throws Exception {
Map<String, Long> path2PartitionIdMap =
- partitionIds.stream().collect(Collectors.toMap(PartitionIdZNode::path, id -> id));
+ partitionIds.stream().collect(toMap(PartitionIdZNode::path, id -> id));
List<ZkGetDataResponse> responses = getDataInBackground(path2PartitionIdMap.keySet());
- return ZooKeeperMetadataUtils.processMultipleDataResponses(
+ return processGetDataResponses(
responses,
response -> path2PartitionIdMap.get(response.getPath()),
PartitionIdZNode::decode,
@@ -510,36 +375,17 @@ public class ZooKeeperClient implements AutoCloseable {
return bytes.map(LeaderAndIsrZNode::decode);
}
- /** Get the buckets LeaderAndIsr in ZK asynchronously. */
- private CompletableFuture<Map<TableBucket, Optional<LeaderAndIsr>>> getLeaderAndIsrsAsync(
- Collection<TableBucket> tableBuckets) {
- if (tableBuckets.isEmpty()) {
- return CompletableFuture.completedFuture(Collections.emptyMap());
- }
-
- Map<String, TableBucket> path2TableBucketMap =
- tableBuckets.stream()
- .collect(Collectors.toMap(LeaderAndIsrZNode::path, bucket -> bucket));
-
- return getDataInBackgroundAsync(path2TableBucketMap.keySet())
- .thenApply(
- responses ->
- ZooKeeperMetadataUtils.processMultipleDataResponsesWithOptional(
- responses,
- response -> path2TableBucketMap.get(response.getPath()),
- LeaderAndIsrZNode::decode,
- "leader and isr"));
- }
-
- /** Get the buckets LeaderAndIsr in ZK. */
+ /**
+ * Get the LeaderAndIsr for each buckets in a batch async way. The returned map only contains
+ * buckets those have leader and isr assigned (zk node created).
+ */
public Map<TableBucket, LeaderAndIsr> getLeaderAndIsrs(Collection<TableBucket> tableBuckets)
throws Exception {
Map<String, TableBucket> path2TableBucketMap =
- tableBuckets.stream()
- .collect(Collectors.toMap(LeaderAndIsrZNode::path, bucket -> bucket));
+ tableBuckets.stream().collect(toMap(LeaderAndIsrZNode::path, bucket -> bucket));
List<ZkGetDataResponse> responses = getDataInBackground(path2TableBucketMap.keySet());
- return ZooKeeperMetadataUtils.processMultipleDataResponses(
+ return processGetDataResponses(
responses,
response -> path2TableBucketMap.get(response.getPath()),
LeaderAndIsrZNode::decode,
@@ -678,23 +524,29 @@ public class ZooKeeperClient implements AutoCloseable {
/** Get the tables in ZK. */
public Map<TablePath, TableRegistration> getTables(Collection<TablePath> tablePaths)
throws Exception {
- return FutureUtils.asyncToSyncDirect(() -> getTablesAsync(tablePaths), "tables");
- }
-
- /** Get the tables in ZK asynchronously. */
- public CompletableFuture<Map<TablePath, TableRegistration>> getTablesAsync(
- Collection<TablePath> tablePaths) {
Map<String, TablePath> path2TablePathMap =
tablePaths.stream().collect(Collectors.toMap(TableZNode::path, path -> path));
- return getDataInBackgroundAsync(path2TablePathMap.keySet())
- .thenApply(
- responses ->
- ZooKeeperMetadataUtils.processMultipleDataResponses(
- responses,
- response -> path2TablePathMap.get(response.getPath()),
- TableZNode::decode,
- "table registration"));
+ List<ZkGetDataResponse> responses = getDataInBackground(path2TablePathMap.keySet());
+ return processGetDataResponses(
+ responses,
+ response -> path2TablePathMap.get(response.getPath()),
+ TableZNode::decode,
+ "tables registration");
+ }
+
+ /** Get the v1 schema for given tables in ZK. */
+ public Map<TablePath, SchemaInfo> getV1Schemas(Collection<TablePath> tablePaths)
+ throws Exception {
+ Map<String, TablePath> path2TablePathMap =
+ tablePaths.stream()
+ .collect(toMap(p -> SchemaZNode.path(p, DEFAULT_SCHEMA_ID), path -> path));
+ List<ZkGetDataResponse> responses = getDataInBackground(path2TablePathMap.keySet());
+ return processGetDataResponses(
+ responses,
+ resp -> path2TablePathMap.get(resp.getPath()),
+ b -> new SchemaInfo(SchemaZNode.decode(b), DEFAULT_SCHEMA_ID),
+ "schema");
}
/** Update the table in ZK. */
@@ -731,13 +583,13 @@ public class ZooKeeperClient implements AutoCloseable {
}
/** Get the partitions of tables in ZK. */
- public Map<TablePath, List<String>> getPartitionsForTables(List<TablePath> tablePaths)
+ public Map<TablePath, List<String>> getPartitionsForTables(Collection<TablePath> tablePaths)
throws Exception {
Map<String, TablePath> path2TablePathMap =
- tablePaths.stream().collect(Collectors.toMap(PartitionsZNode::path, path -> path));
+ tablePaths.stream().collect(toMap(PartitionsZNode::path, path -> path));
List<ZkGetChildrenResponse> responses = getChildrenInBackground(path2TablePathMap.keySet());
- return ZooKeeperMetadataUtils.processMultipleChildrenResponses(
+ return processGetChildrenResponses(
responses,
response -> path2TablePathMap.get(response.getPath()),
"partitions for tables");
@@ -817,33 +669,66 @@ public class ZooKeeperClient implements AutoCloseable {
/** Get the id and name for the partitions of a table in ZK. */
public Map<Long, String> getPartitionIdAndNames(TablePath tablePath) throws Exception {
- Map<Long, String> partitionIdAndNames = new HashMap<>();
- for (String partitionName : getPartitions(tablePath)) {
- Optional<TablePartition> optPartition = getPartition(tablePath, partitionName);
- optPartition.ifPresent(
- partition ->
- partitionIdAndNames.put(partition.getPartitionId(), partitionName));
+ Map<Long, String> result = new HashMap<>();
+ getPartitionIdAndPaths(Collections.singletonList(tablePath))
+ .forEach(
+ (k, v) -> {
+ result.put(k, v.getPartitionName());
+ });
+ return result;
+ }
+
+ /** Get the id and name for the partitions of tables in ZK. */
+ public Map<Long, PhysicalTablePath> getPartitionIdAndPaths(Collection<TablePath> tablePath)
+ throws Exception {
+ // batch get partitions names for tables
+ Map<TablePath, List<String>> partitionsForTables = getPartitionsForTables(tablePath);
+ List<PhysicalTablePath> partitionPaths = new ArrayList<>();
+ for (Map.Entry<TablePath, List<String>> entry : partitionsForTables.entrySet()) {
+ for (String partitionName : entry.getValue()) {
+ partitionPaths.add(PhysicalTablePath.of(entry.getKey(), partitionName));
+ }
}
- return partitionIdAndNames;
+
+ // batch get partitions ids
+ Map<PhysicalTablePath, TablePartition> partitionIds = getPartitionIds(partitionPaths);
+ Map<Long, PhysicalTablePath> partitionIdAndPaths = new HashMap<>();
+ partitionIds.forEach(
+ (partitionPath, tablePartition) -> {
+ if (tablePartition != null) {
+ partitionIdAndPaths.put(tablePartition.getPartitionId(), partitionPath);
+ }
+ });
+
+ return partitionIdAndPaths;
}
/** Get a partition of a table in ZK. */
public Optional<TablePartition> getPartition(TablePath tablePath, String partitionName)
throws Exception {
- return FutureUtils.asyncToSync(
- () -> getPartitionAsync(tablePath, partitionName), "partition");
- }
-
- /** Get a partition of a table in ZK asynchronously. */
- public CompletableFuture<TablePartition> getPartitionAsync(
- TablePath tablePath, String partitionName) {
String path = PartitionZNode.path(tablePath, partitionName);
-
- return getDataInBackgroundAsync(Collections.singleton(path))
- .thenApply(
- responses ->
- ZooKeeperMetadataUtils.processSingleDataResponse(
- responses, PartitionZNode::decode, path));
+ return getOrEmpty(path).map(PartitionZNode::decode);
+ }
+
+ /** Get partition id and table id for each partition in a batch async way. */
+ public Map<PhysicalTablePath, TablePartition> getPartitionIds(
+ Collection<PhysicalTablePath> partitionPaths) throws Exception {
+ Map<String, PhysicalTablePath> path2PartitionPathMap =
+ partitionPaths.stream()
+ .collect(
+ toMap(
+ p ->
+ PartitionZNode.path(
+ p.getTablePath(),
+ checkNotNull(p.getPartitionName())),
+ path -> path));
+
+ List<ZkGetDataResponse> responses = getDataInBackground(path2PartitionPathMap.keySet());
+ return processGetDataResponses(
+ responses,
+ response -> path2PartitionPathMap.get(response.getPath()),
+ PartitionZNode::decode,
+ "partition");
}
/** Get partition num of a table in ZK. */
@@ -1265,59 +1150,6 @@ public class ZooKeeperClient implements AutoCloseable {
}
}
- /**
- * Gets the child nodes at given zk node paths in background asynchronously.
- *
- * @param paths the paths to list children
- * @return CompletableFuture containing list of async responses for each path
- */
- public CompletableFuture<List<ZkGetChildrenResponse>> getChildrenInBackgroundAsync(
- Collection<String> paths) {
- List<ZkGetChildrenRequest> requests =
- paths.stream().map(ZkGetChildrenRequest::new).collect(Collectors.toList());
- return handleRequestInBackgroundAsync(requests, ZkGetChildrenResponse::create);
- }
-
- /**
- * Gets the child nodes at given zk node paths in background.
- *
- * @param paths the paths to list children
- * @return list of async responses for each path
- * @throws Exception if there is an error during the operation
- */
- public List<ZkGetChildrenResponse> getChildrenInBackground(Collection<String> paths)
- throws Exception {
- List<ZkGetChildrenRequest> requests =
- paths.stream().map(ZkGetChildrenRequest::new).collect(Collectors.toList());
- return handleRequestInBackground(requests, ZkGetChildrenResponse::create);
- }
-
- /**
- * Gets the data of given zk node paths in background asynchronously.
- *
- * @param paths the paths to fetch data
- * @return CompletableFuture containing list of async responses for each path
- */
- public CompletableFuture<List<ZkGetDataResponse>> getDataInBackgroundAsync(
- Collection<String> paths) {
- List<ZkGetDataRequest> requests =
- paths.stream().map(ZkGetDataRequest::new).collect(Collectors.toList());
- return handleRequestInBackgroundAsync(requests, ZkGetDataResponse::create);
- }
-
- /**
- * Gets the data of given zk node paths in background.
- *
- * @param paths the paths to fetch data
- * @return list of async responses for each path
- * @throws Exception if there is an error during the operation
- */
- public List<ZkGetDataResponse> getDataInBackground(Collection<String> paths) throws Exception {
- List<ZkGetDataRequest> requests =
- paths.stream().map(ZkGetDataRequest::new).collect(Collectors.toList());
- return handleRequestInBackground(requests, ZkGetDataResponse::create);
- }
-
/** Gets the data and stat of a given zk node path. */
public Optional<Stat> getStat(String path) throws Exception {
try {
@@ -1345,434 +1177,296 @@ public class ZooKeeperClient implements AutoCloseable {
// --------------------------------------------------------------------------------------------
/**
- * Asynchronously retrieves table metadata from ZooKeeper for the specified table.
- *
- * <p>This method fetches the table assignment information from ZooKeeper and converts it into a
- * list of bucket metadata. The operation is performed asynchronously to avoid blocking the
- * calling thread.
- *
- * @param tablePath the path of the table to retrieve metadata for
- * @param tableId the unique identifier of the table
- * @param isPartitioned whether the table is partitioned (affects warning behavior for missing
- * assignments)
- * @return a CompletableFuture containing a list of BucketMetadata objects representing all
- * buckets for this table. Returns an empty list if no table assignment is found.
- * @throws TableNotExistException if the table does not exist in ZooKeeper
- * @throws FlussRuntimeException if there's an error retrieving or processing the metadata
+ * Get bucket metadata for multiple tables in batch async way. The returned map only contains
+ * tables those have assignments. If a table has no assignment yet (just been created), it will
+ * not be included in the result map.
*/
- public CompletableFuture<List<BucketMetadata>> getTableMetadataFromZkAsync(
- TablePath tablePath, long tableId, boolean isPartitioned) {
-
- PhysicalTablePath physicalTablePath = PhysicalTablePath.of(tablePath);
-
- return getAssignmentInfoAsync(tableId, physicalTablePath)
- .thenCompose(
- assignmentInfo -> {
- if (assignmentInfo.tableAssignment == null) {
- if (isPartitioned) {
- LOG.warn(
- "No table assignment node found for table {}", tableId);
- }
- return CompletableFuture.completedFuture(
- new ArrayList<BucketMetadata>());
- }
-
- TableAssignment tableAssignment = assignmentInfo.tableAssignment;
- if (tableAssignment == null) {
- return CompletableFuture.completedFuture(
- new ArrayList<BucketMetadata>());
- }
- tableAssignment = checkNotNull(tableAssignment);
-
- Map<Integer, BucketAssignment> bucketAssignments =
- tableAssignment.getBucketAssignments();
- if (bucketAssignments.isEmpty()) {
- return CompletableFuture.completedFuture(
- new ArrayList<BucketMetadata>());
- }
-
- // Create TableBucket collection for batch processing
- List<TableBucket> tableBuckets = new ArrayList<>();
- Map<TableBucket, Map.Entry<Integer, BucketAssignment>>
- bucketToAssignmentMap = new HashMap<>();
-
- for (Map.Entry<Integer, BucketAssignment> assignment :
- bucketAssignments.entrySet()) {
- int bucketId = assignment.getKey();
- TableBucket tableBucket = new TableBucket(tableId, null, bucketId);
- tableBuckets.add(tableBucket);
- bucketToAssignmentMap.put(tableBucket, assignment);
- }
-
- // Use batch async method to get all LeaderAndIsr information in one ZK
- // call
- return getLeaderAndIsrsAsync(tableBuckets)
- .thenApply(
- leaderAndIsrMap ->
- ZooKeeperMetadataUtils
- .assembleBucketMetadataList(
- tableBuckets,
- bucketToAssignmentMap,
- leaderAndIsrMap,
- null));
- })
- .exceptionally(
- throwable ->
- ZooKeeperMetadataUtils.handleTableMetadataException(
- throwable, tablePath, tableId));
- }
-
- public static CompletableFuture<List<BucketMetadata>> getTableMetadataFromZkAsync(
- ZooKeeperClient zkClient, TablePath tablePath, long tableId, boolean isPartitioned) {
- return zkClient.getTableMetadataFromZkAsync(tablePath, tableId, isPartitioned);
+ public Map<Long, List<BucketMetadata>> getBucketMetadataForTables(Collection<Long> tableIds)
+ throws Exception {
+ // The replica assignments map may not contain all tableIds
+ Map<Long, List<BucketMetadata>> result = new HashMap<>();
+ // initialize result with all table ids.
+ tableIds.forEach(id -> result.put(id, new ArrayList<>()));
+ Map<Long, TableAssignment> tablesAssignments = getTablesAssignments(tableIds);
+ List<TableBucket> buckets = new ArrayList<>();
+ tablesAssignments.forEach(
+ (tableId, assignment) -> {
+ for (Integer bucketId : assignment.getBuckets()) {
+ buckets.add(new TableBucket(tableId, null, bucketId));
+ }
+ });
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrs = getLeaderAndIsrs(buckets);
+ // The LeaderAndIsr map may not contain all buckets, so we iterate on assignment buckets
+ for (TableBucket bucket : buckets) {
+ int bucketId = bucket.getBucket();
+ long tableId = bucket.getTableId();
+ // this might be null if the leader and isr not known yet.
+ BucketMetadata bucketMetadata =
+ createBucketMetadata(
+ leaderAndIsrs,
+ bucket,
+ bucketId,
+ checkNotNull(tablesAssignments.get(tableId)));
+ result.get(tableId).add(bucketMetadata);
+ }
+ return result;
+ }
+
+ /**
+ * Get bucket metadata for multiple partitions in batch async way. The returned map only
+ * contains partitions those have assignments. If a partition has no assignment yet (just been
+ * created), it will not be included in the result map.
+ */
+ public Map<Long, List<BucketMetadata>> getBucketMetadataForPartitions(
+ Collection<Long> partitionIds) throws Exception {
+ // The replica assignments map may not contain all partitionIds
+ Map<Long, PartitionAssignment> partitionsAssignments =
+ getPartitionsAssignments(partitionIds);
+ List<TableBucket> buckets = new ArrayList<>();
+ partitionsAssignments.forEach(
+ (partitionId, assignment) -> {
+ for (Integer bucketId : assignment.getBuckets()) {
+ buckets.add(
+ new TableBucket(assignment.getTableId(), partitionId, bucketId));
+ }
+ });
+ // The LeaderAndIsr map may not contain all buckets
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrs = getLeaderAndIsrs(buckets);
+ Map<Long, List<BucketMetadata>> result = new HashMap<>();
+ for (TableBucket bucket : buckets) {
+ int bucketId = bucket.getBucket();
+ long partitionId = checkNotNull(bucket.getPartitionId());
+ BucketMetadata bucketMetadata =
+ createBucketMetadata(
+ leaderAndIsrs,
+ bucket,
+ bucketId,
+ checkNotNull(partitionsAssignments.get(partitionId)));
+ result.computeIfAbsent(partitionId, k -> new ArrayList<>()).add(bucketMetadata);
+ }
+ return result;
}
+ private BucketMetadata createBucketMetadata(
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrs,
+ TableBucket bucket,
+ int bucketId,
+ TableAssignment assignment) {
+ // this might be null if the leader and isr not known yet.
+ LeaderAndIsr leaderAndIsr = leaderAndIsrs.get(bucket);
+ Integer leader = leaderAndIsr != null ? leaderAndIsr.leader() : null;
+ Integer leaderEpoch = leaderAndIsr != null ? leaderAndIsr.leaderEpoch() : null;
+ List<Integer> replicas = assignment.getBucketAssignments().get(bucketId).getReplicas();
+ return new BucketMetadata(bucketId, leader, leaderEpoch, replicas);
+ }
+
+ /** Close the underlying ZooKeeperClient. */
+ @Override
+ public void close() {
+ LOG.info("Closing...");
+ if (curatorFrameworkWrapper != null) {
+ curatorFrameworkWrapper.close();
+ }
+ }
+
+ // -------------------------------------------------------------------------------------------
+ // ZK batch async utils
+ // -------------------------------------------------------------------------------------------
+
/**
- * Asynchronously retrieves partition metadata from ZooKeeper for the specified partition.
+ * Send a pipelined sequence of requests and return a CompletableFuture for all their responses.
*
- * <p>This method fetches the partition assignment information from ZooKeeper and converts it
- * into partition metadata including all bucket metadata. The operation is performed
- * asynchronously to avoid blocking the calling thread.
+ * <p>The watch flag on each outgoing request will be set if we've already registered a handler
+ * for the path associated with the request.
*
- * @param partitionPath the physical path of the partition to retrieve metadata for
- * @return a CompletableFuture containing a PartitionMetadata object representing the partition
- * and all its buckets. Returns a PartitionMetadata with empty bucket list if no partition
- * assignment is found.
- * @throws PartitionNotExistException if the partition does not exist in ZooKeeper
- * @throws FlussRuntimeException if there's an error retrieving or processing the metadata
- * @throws IllegalArgumentException if the partition ID is null (should never happen for valid
- * partitions)
+ * @param requests a sequence of requests to send
+ * @param respCreator function to create response objects from curator events
+ * @return CompletableFuture containing the responses for the requests
*/
- public CompletableFuture<PartitionMetadata> getPartitionMetadataFromZkAsync(
- PhysicalTablePath partitionPath) {
-
- return getAssignmentInfoAsync(null, partitionPath)
- .thenCompose(
- assignmentInfo -> {
- checkNotNull(
- assignmentInfo.partitionId,
- "partition id must be not null for " + partitionPath);
-
- if (assignmentInfo.tableAssignment == null) {
- LOG.warn(
- "No partition assignment node found for partition {}",
- partitionPath);
-
- PartitionMetadata emptyMetadata =
- new PartitionMetadata(
- assignmentInfo.tableId,
- partitionPath.getPartitionName(),
- assignmentInfo.partitionId != null
- ? assignmentInfo.partitionId
- : -1L,
- new ArrayList<>());
-
- return CompletableFuture.completedFuture(emptyMetadata);
- }
-
- TableAssignment tableAssignment =
- checkNotNull(assignmentInfo.tableAssignment);
- Map<Integer, BucketAssignment> bucketAssignments =
- tableAssignment.getBucketAssignments();
- if (bucketAssignments.isEmpty()) {
- return CompletableFuture.completedFuture(
- new PartitionMetadata(
- assignmentInfo.tableId,
- partitionPath.getPartitionName(),
- assignmentInfo.partitionId != null
- ? assignmentInfo.partitionId
- : -1L,
- new ArrayList<>()));
- }
-
- // Create TableBucket collection for batch processing
- List<TableBucket> tableBuckets = new ArrayList<>();
- Map<TableBucket, Map.Entry<Integer, BucketAssignment>>
- bucketToAssignmentMap = new HashMap<>();
-
- for (Map.Entry<Integer, BucketAssignment> assignment :
- bucketAssignments.entrySet()) {
- int bucketId = assignment.getKey();
- TableBucket tableBucket =
- new TableBucket(
- assignmentInfo.tableId,
- assignmentInfo.partitionId,
- bucketId);
- tableBuckets.add(tableBucket);
- bucketToAssignmentMap.put(tableBucket, assignment);
- }
-
- // Use batch async method to get all LeaderAndIsr information in one ZK
- // call
- return getLeaderAndIsrsAsync(tableBuckets)
- .thenApply(
- leaderAndIsrMap -> {
- List<BucketMetadata> bucketMetadataList =
- ZooKeeperMetadataUtils
- .assembleBucketMetadataList(
- tableBuckets,
- bucketToAssignmentMap,
- leaderAndIsrMap,
- null);
- return new PartitionMetadata(
- assignmentInfo.tableId,
- partitionPath.getPartitionName(),
- assignmentInfo.partitionId != null
- ? assignmentInfo.partitionId
- : -1L,
- bucketMetadataList);
- });
- })
- .exceptionally(
- throwable ->
- ZooKeeperMetadataUtils.handlePartitionException(
- throwable,
- "Failed to get metadata for partition {}",
- partitionPath,
- String.format(
- "Failed to get metadata for partition %s",
- partitionPath)));
- }
-
- public static CompletableFuture<PartitionMetadata> getPartitionMetadataFromZkAsync(
- PhysicalTablePath partitionPath, ZooKeeperClient zkClient) {
- return zkClient.getPartitionMetadataFromZkAsync(partitionPath);
+ private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
+ CompletableFuture<List<Resp>> handleRequestInBackgroundAsync(
+ List<Req> requests, Function<CuratorEvent, Resp> respCreator) {
+ if (requests == null || requests.isEmpty()) {
+ return CompletableFuture.completedFuture(Collections.emptyList());
+ }
+
+ CompletableFuture<List<Resp>> future = new CompletableFuture<>();
+ BackgroundCallback callback = createBackgroundCallback(requests, respCreator, future);
+
+ try {
+ for (Req request : requests) {
+ try {
+ inFlightRequests.acquire();
+ if (request instanceof ZkGetDataRequest) {
+ zkClient.getData().inBackground(callback).forPath(request.getPath());
+
+ } else if (request instanceof ZkGetChildrenRequest) {
+ zkClient.getChildren().inBackground(callback).forPath(request.getPath());
+
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported request type: " + request.getClass());
+ }
+ } catch (Exception e) {
+ inFlightRequests.release();
+ throw e;
+ }
+ }
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+
+ return future;
+ }
+
+ @Nonnull
+ private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
+ BackgroundCallback createBackgroundCallback(
+ List<Req> requests,
+ Function<CuratorEvent, Resp> respCreator,
+ CompletableFuture<List<Resp>> future) {
+ ArrayBlockingQueue<Resp> responseQueue = new ArrayBlockingQueue<>(requests.size());
+ CountDownLatch countDownLatch = new CountDownLatch(requests.size());
+
+ // Complete future when all responses received
+ return (client, event) -> {
+ try {
+ Resp response = respCreator.apply(event);
+ responseQueue.add(response);
+ countDownLatch.countDown();
+
+ // Complete future when all responses received
+ if (countDownLatch.getCount() == 0) {
+ future.complete(new ArrayList<>(responseQueue));
+ }
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ } finally {
+ inFlightRequests.release();
+ }
+ };
}
/**
- * Asynchronously retrieves partition metadata from ZooKeeper for multiple partitions in batch.
+ * Gets the child nodes at given zk node paths in background.
*
- * <p>This method efficiently fetches partition metadata for multiple partitions across multiple
- * tables in a single batched operation. It performs the following steps:
+ * @param paths the paths to list children
+ * @return list of async responses for each path
+ * @throws Exception if there is an error during the operation
+ */
+ private List<ZkGetChildrenResponse> getChildrenInBackground(Collection<String> paths)
+ throws Exception {
+ List<ZkGetChildrenRequest> requests =
+ paths.stream().map(ZkGetChildrenRequest::new).collect(Collectors.toList());
+ return handleRequestInBackground(requests, ZkGetChildrenResponse::create);
+ }
+
+ /**
+ * Gets the data of given zk node paths in background.
*
- * <ul>
- * <li>Maps table paths to their partition information (name and ID mappings)
- * <li>Validates that all requested partition IDs exist
- * <li>Builds partition path information with resolved table IDs
- * <li>Batch retrieves partition assignments and bucket metadata
- * </ul>
+ * @param paths the paths to fetch data
+ * @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 {
+ List<ZkGetDataRequest> requests =
+ paths.stream().map(ZkGetDataRequest::new).collect(Collectors.toList());
+ return handleRequestInBackground(requests, ZkGetDataResponse::create);
+ }
+
+ /**
+ * Send a pipelined sequence of requests and wait for all of their responses synchronously in
+ * background.
*
- * <p>The batch operation is optimized to minimize ZooKeeper calls by grouping related
- * operations together, making it much more efficient than calling {@link
- * #getPartitionMetadataFromZkAsync(PhysicalTablePath)} individually for each partition.
+ * <p>The watch flag on each outgoing request will be set if we've already registered a handler
+ * for the path associated with the request.
*
- * @param tablePaths the collection of table paths that may contain the requested partitions.
- * This is used to scope the search and improve efficiency. Cannot be empty if
- * partitionIdSet is not empty.
- * @param partitionIdSet the set of partition IDs to retrieve metadata for. Cannot be empty if
- * tablePaths is not empty.
- * @return a CompletableFuture containing a list of PartitionMetadata objects for all found
- * partitions. The list may be smaller than the input partition ID set if some partitions
- * have no assignment data. Returns an empty list if either input parameter is empty.
- * @throws PartitionNotExistException if any of the requested partition IDs do not exist in any
- * of the provided tables
- * @throws FlussRuntimeException if there's an error during ZooKeeper operations or metadata
- * processing
+ * @param requests a sequence of requests to send and wait on.
+ * @return the responses for the requests. If all requests have the same type, the responses
+ * will have the respective response type.
*/
- public CompletableFuture<List<PartitionMetadata>> batchGetPartitionMetadataFromZkAsync(
- Collection<TablePath> tablePaths, Set<Long> partitionIdSet) {
- if (tablePaths.isEmpty() || partitionIdSet.isEmpty()) {
- return CompletableFuture.completedFuture(Collections.<PartitionMetadata>emptyList());
+ private <Resp extends ZkAsyncResponse, Req extends ZkAsyncRequest>
+ List<Resp> handleRequestInBackground(
+ List<Req> requests, Function<CuratorEvent, Resp> respCreator) throws Exception {
+ try {
+ return handleRequestInBackgroundAsync(requests, respCreator).get();
+ } catch (ExecutionException e) {
+ Throwable cause = ExceptionUtils.stripExecutionException(e);
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ } else {
+ throw new Exception("Async request handling failed", cause);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new Exception("Request handling was interrupted", e);
}
+ }
- return CompletableFuture.supplyAsync(
- () -> {
- try {
- return getPartitionNameAndIdsForTables(new ArrayList<>(tablePaths));
- } catch (Exception e) {
- throw new RuntimeException(
- "Failed to get partition name and ids for tables", e);
- }
- })
- .thenCompose(
- tableToPartitionInfo -> {
- // Inline collectValidPartitionsAndTables logic
- Set<Long> remainingPartitionIds = new HashSet<>(partitionIdSet);
- Set<TablePath> tablesNeedingIds = new HashSet<>();
- Map<TablePath, List<Tuple2<String, Long>>> tablePartitionMapping =
- new HashMap<>();
-
- for (Map.Entry<TablePath, Map<String, Long>> entry :
- tableToPartitionInfo.entrySet()) {
- TablePath tablePath = entry.getKey();
- Map<String, Long> partitionInfo = entry.getValue();
-
- List<Tuple2<String, Long>> validPartitions =
- ZooKeeperMetadataUtils.findValidPartitionsForTable(
- partitionInfo,
- partitionIdSet,
- remainingPartitionIds);
-
- if (!validPartitions.isEmpty()) {
- tablesNeedingIds.add(tablePath);
- tablePartitionMapping.put(tablePath, validPartitions);
- }
- }
-
- if (!remainingPartitionIds.isEmpty()) {
- throw new PartitionNotExistException(
- "Partition not exist for partition ids: "
- + remainingPartitionIds);
- }
-
- if (tablePartitionMapping.isEmpty()) {
- return CompletableFuture.completedFuture(
- Collections.<PartitionPathInfo>emptyList());
- }
-
- return getTablesAsync(tablesNeedingIds)
- .thenApply(
- tableRegistrations ->
- ZooKeeperMetadataUtils
- .buildPartitionPathInfoList(
- tablePartitionMapping,
- tableRegistrations));
- })
- .thenCompose(
- partitionPathInfos -> {
- if (partitionPathInfos.isEmpty()) {
- return CompletableFuture.completedFuture(
- Collections.<PartitionMetadata>emptyList());
- }
-
- List<Long> partitionIds =
- partitionPathInfos.stream()
- .map(info -> info.partitionId)
- .collect(Collectors.toList());
-
- CompletableFuture<Map<Long, PartitionAssignment>>
- partitionAssignmentsFuture;
- if (partitionIds.isEmpty()) {
- partitionAssignmentsFuture =
- CompletableFuture.completedFuture(Collections.emptyMap());
- } else {
- Map<String, Long> path2PartitionIdMap =
- partitionIds.stream()
- .collect(
- Collectors.toMap(
- PartitionIdZNode::path, id -> id));
-
- partitionAssignmentsFuture =
- getDataInBackgroundAsync(path2PartitionIdMap.keySet())
- .thenApply(
- responses ->
- ZooKeeperMetadataUtils
- .processMultipleDataResponses(
- responses,
- response ->
- path2PartitionIdMap
- .get(
- response
- .getPath()),
- PartitionIdZNode
- ::decode,
- "partition assignment"));
- }
-
- return partitionAssignmentsFuture.thenCompose(
- partitionAssignments -> {
- List<AssignmentInfo> assignmentInfos = new ArrayList<>();
- for (PartitionPathInfo pathInfo : partitionPathInfos) {
- PartitionAssignment partitionAssignment =
- partitionAssignments.get(pathInfo.partitionId);
- AssignmentInfo assignmentInfo =
- new AssignmentInfo(
- pathInfo.tableId,
- partitionAssignment,
- pathInfo.partitionId);
- assignmentInfos.add(assignmentInfo);
- }
-
- List<TableBucket> allTableBuckets = new ArrayList<>();
- Map<TableBucket, Integer> bucketToIndexMap =
- new HashMap<>();
-
- for (int i = 0; i < assignmentInfos.size(); i++) {
- AssignmentInfo assignmentInfo = assignmentInfos.get(i);
- PartitionPathInfo pathInfo = partitionPathInfos.get(i);
-
- if (assignmentInfo.tableAssignment != null) {
- for (Integer bucketId :
- assignmentInfo.tableAssignment
- .getBuckets()) {
- TableBucket tableBucket =
- new TableBucket(
- pathInfo.tableId,
- pathInfo.partitionId,
- bucketId);
- allTableBuckets.add(tableBucket);
- bucketToIndexMap.put(tableBucket, i);
- }
- }
- }
-
- return getLeaderAndIsrsAsync(allTableBuckets)
- .thenApply(
- leaderAndIsrMap ->
- ZooKeeperMetadataUtils
- .assemblePartitionMetadata(
- partitionPathInfos,
- assignmentInfos,
- bucketToIndexMap,
- leaderAndIsrMap));
- });
- })
- .exceptionally(
- throwable ->
- ZooKeeperMetadataUtils.handlePartitionException(
- throwable,
- "Failed to batch get partition metadata for partition ids: {}",
- partitionIdSet,
- "Failed to batch get partition metadata for partition ids: "
- + partitionIdSet));
- }
-
- private CompletableFuture<AssignmentInfo> getAssignmentInfoAsync(
- @Nullable Long tableId, PhysicalTablePath physicalTablePath) {
-
- if (physicalTablePath.getPartitionName() != null) {
- return getPartitionAsync(
- physicalTablePath.getTablePath(), physicalTablePath.getPartitionName())
- .thenApply(
- tablePartition -> {
- if (tablePartition == null) {
- throw new PartitionNotExistException(
- "Table partition '"
- + physicalTablePath
- + "' does not exist.");
- }
- return tablePartition;
- })
- .thenCompose(
- tablePartition -> {
- long partitionId = tablePartition.getPartitionId();
- long partitionTableId = tablePartition.getTableId();
-
- return getPartitionAssignmentAsync(partitionId)
- .thenApply(
- tableAssignment ->
- new AssignmentInfo(
- partitionTableId,
- tableAssignment,
- partitionId));
- });
- } else {
- checkNotNull(tableId, "tableId must be not null");
- long realTableId = tableId != null ? tableId : -1L;
-
- return getTableAssignmentAsync(realTableId)
- .thenApply(
- tableAssignment ->
- new AssignmentInfo(realTableId, tableAssignment, null));
+ /**
+ * Template method to process multiple ZooKeeper data responses with decoder.
+ *
+ * @param responses list of ZkGetDataResponse from ZooKeeper
+ * @param keyExtractor function to extract key from response
+ * @param decoder function to decode byte array to target type
+ * @param operationName name of the operation for error messages
+ * @param <K> the type of the result map key
+ * @param <V> the type of the result map value
+ * @return Map containing decoded results for successful responses
+ */
+ public static <K, V> Map<K, V> processGetDataResponses(
+ List<ZkGetDataResponse> responses,
+ Function<ZkGetDataResponse, K> keyExtractor,
+ Function<byte[], V> decoder,
+ String operationName) {
+ Map<K, V> result = new HashMap<>();
+ for (ZkGetDataResponse response : responses) {
+ if (response.getResultCode() == KeeperException.Code.OK) {
+ byte[] data = response.getData();
+ if (data != null && data.length > 0) {
+ K key = keyExtractor.apply(response);
+ V value = decoder.apply(response.getData());
+ result.put(key, value);
+ } else {
+ LOG.warn("Data is empty for path {}", response.getPath());
+ }
+ } else {
+ LOG.warn(
+ "Failed to get {} for path {}: {}",
+ operationName,
+ response.getPath(),
+ response.getResultCode());
+ }
}
+ return result;
}
- /** Close the underlying ZooKeeperClient. */
- @Override
- public void close() {
- LOG.info("Closing...");
- if (curatorFrameworkWrapper != null) {
- curatorFrameworkWrapper.close();
+ /**
+ * Template method to process multiple ZooKeeper children responses with decoder.
+ *
+ * @param responses list of ZkGetChildrenResponse from ZooKeeper
+ * @param keyExtractor function to extract key from response
+ * @param operationName name of the operation for error messages
+ * @param <K> the type of the result map key
+ * @return Map containing children lists for successful responses
+ */
+ public static <K> Map<K, List<String>> processGetChildrenResponses(
+ List<ZkGetChildrenResponse> responses,
+ Function<ZkGetChildrenResponse, K> keyExtractor,
+ String operationName) {
+ Map<K, List<String>> result = new HashMap<>();
+ for (ZkGetChildrenResponse response : responses) {
+ if (response.getResultCode() == KeeperException.Code.OK) {
+ K key = keyExtractor.apply(response);
+ result.put(key, response.getChildren());
+ } else {
+ LOG.warn(
+ "Failed to get {} for path {}: {}",
+ operationName,
+ response.getPath(),
+ response.getResultCode());
+ }
}
+ return result;
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperMetadataUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperMetadataUtils.java
deleted file mode 100644
index 803db46b2..000000000
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperMetadataUtils.java
+++ /dev/null
@@ -1,478 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.fluss.server.zk;
-
-import org.apache.fluss.exception.FlussRuntimeException;
-import org.apache.fluss.exception.PartitionNotExistException;
-import org.apache.fluss.exception.TableNotExistException;
-import org.apache.fluss.metadata.PhysicalTablePath;
-import org.apache.fluss.metadata.TableBucket;
-import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.server.metadata.BucketMetadata;
-import org.apache.fluss.server.metadata.PartitionMetadata;
-import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetChildrenResponse;
-import org.apache.fluss.server.zk.ZkAsyncResponse.ZkGetDataResponse;
-import org.apache.fluss.server.zk.data.BucketAssignment;
-import org.apache.fluss.server.zk.data.LeaderAndIsr;
-import org.apache.fluss.server.zk.data.TableAssignment;
-import org.apache.fluss.server.zk.data.TableRegistration;
-import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
-import org.apache.fluss.utils.ExceptionUtils;
-import org.apache.fluss.utils.types.Tuple2;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.function.Function;
-
-/** Utility class containing helper methods for ZooKeeper metadata operations. */
-public class ZooKeeperMetadataUtils {
-
- private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperMetadataUtils.class);
-
- /**
- * Template method to process multiple ZooKeeper data responses with decoder.
- *
- * @param responses list of ZkGetDataResponse from ZooKeeper
- * @param keyExtractor function to extract key from response
- * @param decoder function to decode byte array to target type
- * @param operationName name of the operation for error messages
- * @param <K> the type of the result map key
- * @param <V> the type of the result map value
- * @return Map containing decoded results for successful responses
- */
- public static <K, V> Map<K, V> processMultipleDataResponses(
- List<ZkGetDataResponse> responses,
- Function<ZkGetDataResponse, K> keyExtractor,
- Function<byte[], V> decoder,
- String operationName) {
- Map<K, V> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- K key = keyExtractor.apply(response);
- V value = decoder.apply(response.getData());
- result.put(key, value);
- } else {
- LOG.warn(
- "Failed to get {} for path {}: {}",
- operationName,
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
- }
-
- /**
- * Template method to process multiple ZooKeeper data responses with decoder and condition.
- *
- * @param responses list of ZkGetDataResponse from ZooKeeper
- * @param keyExtractor function to extract key from response
- * @param decoder function to decode byte array to target type
- * @param condition additional condition to check before processing (beyond OK status)
- * @param operationName name of the operation for error messages
- * @param <K> the type of the result map key
- * @param <V> the type of the result map value
- * @return Map containing decoded results for successful responses
- */
- public static <K, V> Map<K, V> processMultipleDataResponsesWithCondition(
- List<ZkGetDataResponse> responses,
- Function<ZkGetDataResponse, K> keyExtractor,
- Function<byte[], V> decoder,
- Function<ZkGetDataResponse, Boolean> condition,
- String operationName) {
- Map<K, V> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK && condition.apply(response)) {
- K key = keyExtractor.apply(response);
- V value = decoder.apply(response.getData());
- result.put(key, value);
- } else {
- String reason =
- response.getResultCode() != KeeperException.Code.OK
- ? response.getResultCode().toString()
- : "condition check failed";
- LOG.warn(
- "Failed to get {} for path {}: {}",
- operationName,
- response.getPath(),
- reason);
- }
- }
- return result;
- }
-
- /**
- * Template method to process multiple ZooKeeper data responses with decoder and Optional
- * result. This method handles NONODE as a valid case returning Optional.empty().
- *
- * @param responses list of ZkGetDataResponse from ZooKeeper
- * @param keyExtractor function to extract key from response
- * @param decoder function to decode byte array to target type
- * @param operationName name of the operation for error messages
- * @param <K> the type of the result map key
- * @param <V> the type of the result map value
- * @return Map containing Optional results for all responses (including NONODE cases)
- */
- public static <K, V> Map<K, Optional<V>> processMultipleDataResponsesWithOptional(
- List<ZkGetDataResponse> responses,
- Function<ZkGetDataResponse, K> keyExtractor,
- Function<byte[], V> decoder,
- String operationName) {
- Map<K, Optional<V>> result = new HashMap<>();
- for (ZkGetDataResponse response : responses) {
- K key = keyExtractor.apply(response);
- if (response.getResultCode() == KeeperException.Code.OK) {
- V value = decoder.apply(response.getData());
- result.put(key, Optional.of(value));
- } else if (response.getResultCode() == KeeperException.Code.NONODE) {
- result.put(key, Optional.empty());
- } else {
- LOG.warn(
- "Failed to get {} for path {}: {}",
- operationName,
- response.getPath(),
- response.getResultCode());
- result.put(key, Optional.empty());
- }
- }
- return result;
- }
-
- /**
- * Template method to process multiple ZooKeeper children responses with decoder.
- *
- * @param responses list of ZkGetChildrenResponse from ZooKeeper
- * @param keyExtractor function to extract key from response
- * @param operationName name of the operation for error messages
- * @param <K> the type of the result map key
- * @return Map containing children lists for successful responses
- */
- public static <K> Map<K, List<String>> processMultipleChildrenResponses(
- List<ZkGetChildrenResponse> responses,
- Function<ZkGetChildrenResponse, K> keyExtractor,
- String operationName) {
- Map<K, List<String>> result = new HashMap<>();
- for (ZkGetChildrenResponse response : responses) {
- if (response.getResultCode() == KeeperException.Code.OK) {
- K key = keyExtractor.apply(response);
- result.put(key, response.getChildren());
- } else {
- LOG.warn(
- "Failed to get {} for path {}: {}",
- operationName,
- response.getPath(),
- response.getResultCode());
- }
- }
- return result;
- }
-
- /**
- * Template method to process single ZooKeeper data response with decoder.
- *
- * @param responses list of ZkGetDataResponse from ZooKeeper
- * @param decoder function to decode byte array to target type
- * @param path ZooKeeper path for error messages
- * @param <T> the type of the decoded result
- * @return decoded result, or null if no data or NONODE
- */
- public static <T> T processSingleDataResponse(
- List<ZkGetDataResponse> responses, Function<byte[], T> decoder, String path) {
- if (responses.isEmpty()) {
- return null;
- }
-
- ZkGetDataResponse response = responses.get(0);
- if (response.getResultCode() == KeeperException.Code.OK) {
- return decoder.apply(response.getData());
- } else if (response.getResultCode() == KeeperException.Code.NONODE) {
- return null;
- } else {
- throw new RuntimeException(KeeperException.create(response.getResultCode(), path));
- }
- }
-
- /**
- * Handle exceptions that occur during table metadata retrieval.
- *
- * @param throwable the exception that occurred
- * @param tablePath the table path for error context
- * @param tableId the table ID for error context
- * @return never returns normally, always throws an exception
- * @throws TableNotExistException if the original cause was table not existing
- * @throws FlussRuntimeException for other errors
- */
- public static List<BucketMetadata> handleTableMetadataException(
- Throwable throwable, TablePath tablePath, long tableId) {
-
- Throwable cause = ExceptionUtils.stripCompletionException(throwable);
- cause = ExceptionUtils.stripException(cause, RuntimeException.class);
-
- if (cause instanceof TableNotExistException) {
- throw (TableNotExistException) cause;
- }
-
- LOG.error("Failed to get metadata for table {}, id {}", tablePath, tableId, cause);
- throw new FlussRuntimeException(
- String.format("Failed to get metadata for table %s, id %d", tablePath, tableId),
- cause);
- }
-
- /**
- * Assemble a list of BucketMetadata from table buckets, assignments, and leader information.
- *
- * @param tableBuckets the list of table buckets to process
- * @param bucketToAssignmentMap mapping from table buckets to their assignment entries
- * @param leaderAndIsrMap mapping from table buckets to their leader and ISR information
- * @param leaderEpoch the leader epoch to use for all buckets (can be null)
- * @return a list of BucketMetadata objects
- */
- public static List<BucketMetadata> assembleBucketMetadataList(
- List<TableBucket> tableBuckets,
- Map<TableBucket, Map.Entry<Integer, BucketAssignment>> bucketToAssignmentMap,
- Map<TableBucket, Optional<LeaderAndIsr>> leaderAndIsrMap,
- @Nullable Integer leaderEpoch) {
-
- List<BucketMetadata> result = new ArrayList<>();
-
- for (TableBucket tableBucket : tableBuckets) {
- Map.Entry<Integer, BucketAssignment> assignment =
- bucketToAssignmentMap.get(tableBucket);
- int bucketId = assignment.getKey();
- List<Integer> replicas = assignment.getValue().getReplicas();
-
- Optional<LeaderAndIsr> optLeaderAndIsr = leaderAndIsrMap.get(tableBucket);
- Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
-
- result.add(new BucketMetadata(bucketId, leader, leaderEpoch, replicas));
- }
-
- return result;
- }
-
- /**
- * Common exception handler for partition-related operations.
- *
- * @param throwable the exception that occurred
- * @param logMessageTemplate the template for log message (with {} placeholder)
- * @param logContext the context object to be logged
- * @param errorMessage the error message for FlussRuntimeException
- * @param <T> the return type (never actually returns)
- * @return never returns normally, always throws an exception
- * @throws PartitionNotExistException if the original cause was partition not existing
- * @throws FlussRuntimeException for other errors
- */
- public static <T> T handlePartitionException(
- Throwable throwable,
- String logMessageTemplate,
- Object logContext,
- String errorMessage) {
-
- Throwable cause = ExceptionUtils.stripCompletionException(throwable);
- cause = ExceptionUtils.stripException(cause, RuntimeException.class);
-
- if (cause instanceof PartitionNotExistException) {
- throw (PartitionNotExistException) cause;
- }
-
- LOG.error(logMessageTemplate, logContext, cause);
- throw new FlussRuntimeException(errorMessage, cause);
- }
-
- /**
- * Find valid partitions for a specific table.
- *
- * @param partitionInfo partition information for the table
- * @param partitionIdSet the set of partition IDs to look for
- * @param remainingPartitionIds mutable set to track unmatched partition IDs
- * @return list of valid partitions (name, id) tuples
- */
- public static List<Tuple2<String, Long>> findValidPartitionsForTable(
- Map<String, Long> partitionInfo,
- Set<Long> partitionIdSet,
- Set<Long> remainingPartitionIds) {
-
- List<Tuple2<String, Long>> validPartitions = new ArrayList<>();
-
- for (Map.Entry<String, Long> partitionEntry : partitionInfo.entrySet()) {
- Long partitionId = partitionEntry.getValue();
- if (partitionIdSet.contains(partitionId)) {
- String partitionName = partitionEntry.getKey();
- validPartitions.add(Tuple2.of(partitionName, partitionId));
- remainingPartitionIds.remove(partitionId);
- }
- }
-
- return validPartitions;
- }
-
- /**
- * Build the list of PartitionPathInfo from table registrations and partition mappings.
- *
- * @param tablePartitionMapping mapping of tables to their valid partitions
- * @param tableRegistrations table registration information containing table IDs
- * @return list of PartitionPathInfo objects
- */
- public static List<PartitionPathInfo> buildPartitionPathInfoList(
- Map<TablePath, List<Tuple2<String, Long>>> tablePartitionMapping,
- Map<TablePath, TableRegistration> tableRegistrations) {
-
- List<PartitionPathInfo> targetPartitionInfos = new ArrayList<>();
-
- for (Map.Entry<TablePath, List<Tuple2<String, Long>>> entry :
- tablePartitionMapping.entrySet()) {
- TablePath tablePath = entry.getKey();
- TableRegistration tableRegistration = tableRegistrations.get(tablePath);
-
- if (tableRegistration != null) {
- long tableId = tableRegistration.tableId;
- addPartitionPathInfosForTable(
- targetPartitionInfos, tablePath, tableId, entry.getValue());
- }
- }
-
- return targetPartitionInfos;
- }
-
- /**
- * Add PartitionPathInfo objects for all partitions of a specific table.
- *
- * @param targetPartitionInfos target list to add to
- * @param tablePath the table path
- * @param tableId the table ID
- * @param partitionInfos list of partition (name, id) tuples
- */
- public static void addPartitionPathInfosForTable(
- List<PartitionPathInfo> targetPartitionInfos,
- TablePath tablePath,
- long tableId,
- List<Tuple2<String, Long>> partitionInfos) {
-
- for (Tuple2<String, Long> partitionInfo : partitionInfos) {
- String partitionName = partitionInfo.f0;
- Long partitionId = partitionInfo.f1;
- PhysicalTablePath partitionPath = PhysicalTablePath.of(tablePath, partitionName);
- targetPartitionInfos.add(new PartitionPathInfo(partitionPath, partitionId, tableId));
- }
- }
-
- /**
- * Assemble partition metadata from the collected bucket information.
- *
- * @param partitionPathInfos the partition path information
- * @param assignmentInfos the assignment information for each partition
- * @param bucketToIndexMap mapping from table buckets to partition index
- * @param leaderAndIsrMap mapping from table buckets to leader information
- * @return list of assembled partition metadata
- */
- public static List<PartitionMetadata> assemblePartitionMetadata(
- List<PartitionPathInfo> partitionPathInfos,
- List<AssignmentInfo> assignmentInfos,
- Map<TableBucket, Integer> bucketToIndexMap,
- Map<TableBucket, Optional<LeaderAndIsr>> leaderAndIsrMap) {
-
- Map<Integer, List<BucketMetadata>> indexToBuckets = new HashMap<>();
-
- for (Map.Entry<TableBucket, Optional<LeaderAndIsr>> entry : leaderAndIsrMap.entrySet()) {
- TableBucket tableBucket = entry.getKey();
- Optional<LeaderAndIsr> optLeaderAndIsr = entry.getValue();
- Integer assignmentIndex = bucketToIndexMap.get(tableBucket);
-
- if (assignmentIndex != null) {
- AssignmentInfo assignmentInfo = assignmentInfos.get(assignmentIndex);
- if (assignmentInfo.tableAssignment != null) {
- BucketAssignment bucketAssignment =
- assignmentInfo
- .tableAssignment
- .getBucketAssignments()
- .get(tableBucket.getBucket());
-
- if (bucketAssignment != null) {
- Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
- BucketMetadata bucketMetadata =
- new BucketMetadata(
- tableBucket.getBucket(),
- leader,
- // leaderEpoch
- null,
- bucketAssignment.getReplicas());
-
- indexToBuckets
- .computeIfAbsent(assignmentIndex, k -> new ArrayList<>())
- .add(bucketMetadata);
- }
- }
- }
- }
-
- List<PartitionMetadata> result = new ArrayList<>();
- for (int i = 0; i < partitionPathInfos.size(); i++) {
- PartitionPathInfo pathInfo = partitionPathInfos.get(i);
- List<BucketMetadata> bucketMetadataList =
- indexToBuckets.getOrDefault(i, new ArrayList<>());
-
- result.add(
- new PartitionMetadata(
- pathInfo.tableId,
- pathInfo.partitionPath.getPartitionName(),
- pathInfo.partitionId,
- bucketMetadataList));
- }
-
- return result;
- }
-
- /** Data structure to hold partition path and its corresponding partition ID. */
- public static class PartitionPathInfo {
- public final PhysicalTablePath partitionPath;
- public final long partitionId;
- public final long tableId;
-
- public PartitionPathInfo(PhysicalTablePath partitionPath, long partitionId, long tableId) {
- this.partitionPath = partitionPath;
- this.partitionId = partitionId;
- this.tableId = tableId;
- }
- }
-
- /** Data structure to hold assignment information. */
- public static class AssignmentInfo {
- public final long tableId;
- // null then the bucket doesn't belong to a partition. Otherwise, not null
- public final @Nullable Long partitionId;
- public final @Nullable TableAssignment tableAssignment;
-
- public AssignmentInfo(
- long tableId,
- @Nullable TableAssignment tableAssignment,
- @Nullable Long partitionId) {
- this.tableId = tableId;
- this.tableAssignment = tableAssignment;
- this.partitionId = partitionId;
- }
- }
-}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java
index 47f445926..61812d2a7 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/TabletServerMetadataCacheTest.java
@@ -88,8 +88,7 @@ public class TabletServerMetadataCacheTest {
serverMetadataCache =
new TabletServerMetadataCache(
new TestingMetadataManager(
- Arrays.asList(DATA1_TABLE_INFO, partitionTableInfo)),
- null);
+ Arrays.asList(DATA1_TABLE_INFO, partitionTableInfo)));
coordinatorServer =
new ServerInfo(
0,
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java
new file mode 100644
index 000000000..69a25dc55
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.metadata;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.PartitionNotExistException;
+import org.apache.fluss.metadata.DatabaseDescriptor;
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.TestData;
+import org.apache.fluss.server.coordinator.MetadataManager;
+import org.apache.fluss.server.zk.NOPErrorHandler;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.ZooKeeperExtension;
+import org.apache.fluss.server.zk.data.BucketAssignment;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+import org.apache.fluss.server.zk.data.PartitionAssignment;
+import org.apache.fluss.server.zk.data.TableAssignment;
+import org.apache.fluss.server.zk.data.TableRegistration;
+import org.apache.fluss.testutils.common.AllCallbackWrapper;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ZkBasedMetadataProviderTest {
+ @RegisterExtension
+ public static final AllCallbackWrapper<ZooKeeperExtension> ZOO_KEEPER_EXTENSION_WRAPPER =
+ new AllCallbackWrapper<>(new ZooKeeperExtension());
+
+ private static ZooKeeperClient zookeeperClient;
+ private static MetadataManager metadataManager;
+ private static ZkBasedMetadataProvider metadataProvider;
+
+ @BeforeAll
+ static void beforeAll() {
+ zookeeperClient =
+ ZOO_KEEPER_EXTENSION_WRAPPER
+ .getCustomExtension()
+ .getZooKeeperClient(NOPErrorHandler.INSTANCE);
+ metadataManager = new MetadataManager(zookeeperClient, new Configuration());
+ TabletServerMetadataCache metadataCache = new TabletServerMetadataCache(metadataManager);
+ metadataProvider =
+ new TabletServerMetadataProvider(zookeeperClient, metadataManager, metadataCache);
+ }
+
+ @AfterEach
+ void afterEach() {
+ ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupRoot();
+ }
+
+ @AfterAll
+ static void afterAll() {
+ zookeeperClient.close();
+ }
+
+ @Test
+ void testGetTableMetadataFromZk() throws Exception {
+ // Create Table
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ TableDescriptor desc = TestData.DATA1_TABLE_DESCRIPTOR.withReplicationFactor(1);
+ TableAssignment tableAssignment =
+ TableAssignment.builder()
+ .add(0, BucketAssignment.of(1, 2, 3))
+ .add(1, BucketAssignment.of(2, 3, 4))
+ .build();
+ metadataManager.createDatabase("test_db", DatabaseDescriptor.EMPTY, true);
+ long tableId = metadataManager.createTable(tablePath, desc, tableAssignment, false);
+
+ // Create leader and isr for buckets
+ TableBucket tableBucket0 = new TableBucket(tableId, 0);
+ TableBucket tableBucket1 = new TableBucket(tableId, 1);
+
+ LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3, 4), 200, 2000);
+
+ zookeeperClient.registerLeaderAndIsr(tableBucket0, leaderAndIsr0);
+ zookeeperClient.registerLeaderAndIsr(tableBucket1, leaderAndIsr1);
+
+ List<TableMetadata> tablesMetadataFromZK =
+ metadataProvider.getTablesMetadataFromZK(
+ Collections.singletonList(TablePath.of("test_db", "test_table")));
+ assertThat(tablesMetadataFromZK).hasSize(1);
+
+ TableMetadata tableMetadata = tablesMetadataFromZK.get(0);
+ TableInfo tableInfo = tableMetadata.getTableInfo();
+ assertThat(tableInfo.getTableId()).isEqualTo(tableId);
+ assertThat(tableInfo.getTablePath()).isEqualTo(tablePath);
+ assertThat(tableInfo.getSchema()).isEqualTo(desc.getSchema());
+
+ List<BucketMetadata> bucketMetadataList = tableMetadata.getBucketMetadataList();
+ assertThat(bucketMetadataList).hasSize(2);
+
+ // Verify bucket metadata
+ Map<Integer, BucketMetadata> bucketMap =
+ bucketMetadataList.stream()
+ .collect(Collectors.toMap(BucketMetadata::getBucketId, b -> b));
+
+ BucketMetadata bucket0Metadata = bucketMap.get(0);
+ assertThat(extractLeaderFromBucketMetadata(bucket0Metadata)).isEqualTo(1);
+ assertThat(bucket0Metadata.getReplicas()).containsExactly(1, 2, 3);
+
+ BucketMetadata bucket1Metadata = bucketMap.get(1);
+ assertThat(extractLeaderFromBucketMetadata(bucket1Metadata)).isEqualTo(2);
+ assertThat(bucket1Metadata.getReplicas()).containsExactly(2, 3, 4);
+ }
+
+ @Test
+ void testGetPartitionMetadataFromZk() throws Exception {
+ // Prepare test data
+ TablePath tablePath = TablePath.of("test_db", "test_partition_table");
+ long tableId = 2001L;
+ long partitionId = 1L;
+ String partitionName = "p1";
+
+ long currentMillis = System.currentTimeMillis();
+ TableRegistration tableReg =
+ createTestTableRegistration(tableId, "partition table", currentMillis);
+ zookeeperClient.registerTable(tablePath, tableReg);
+
+ // Create partition assignment
+ Map<Integer, BucketAssignment> bucketAssignments = new HashMap<>();
+ bucketAssignments.put(0, BucketAssignment.of(1, 2));
+ bucketAssignments.put(1, BucketAssignment.of(2, 3));
+ PartitionAssignment partitionAssignment =
+ new PartitionAssignment(tableId, bucketAssignments);
+
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId, partitionName, partitionAssignment, tablePath, tableId);
+
+ // Create leader and isr for partition buckets
+ TableBucket partitionBucket0 = new TableBucket(tableId, partitionId, 0);
+ TableBucket partitionBucket1 = new TableBucket(tableId, partitionId, 1);
+
+ LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000);
+
+ zookeeperClient.registerLeaderAndIsr(partitionBucket0, leaderAndIsr0);
+ zookeeperClient.registerLeaderAndIsr(partitionBucket1, leaderAndIsr1);
+
+ // Test getPartitionMetadataFromZkAsync
+ PhysicalTablePath partitionPath = PhysicalTablePath.of(tablePath, partitionName);
+
+ List<PartitionMetadata> partitionsMetadataFromZK =
+ metadataProvider.getPartitionsMetadataFromZK(Collections.singleton(partitionPath));
+
+ assertThat(partitionsMetadataFromZK).hasSize(1);
+ PartitionMetadata partitionMetadata = partitionsMetadataFromZK.get(0);
+
+ assertThat(partitionMetadata.getTableId()).isEqualTo(tableId);
+ assertThat(partitionMetadata.getPartitionName()).isEqualTo(partitionName);
+ assertThat(partitionMetadata.getPartitionId()).isEqualTo(partitionId);
+
+ List<BucketMetadata> bucketMetadataList = partitionMetadata.getBucketMetadataList();
+ assertThat(bucketMetadataList).hasSize(2);
+
+ Map<Integer, BucketMetadata> bucketMap =
+ bucketMetadataList.stream()
+ .collect(Collectors.toMap(BucketMetadata::getBucketId, b -> b));
+
+ BucketMetadata bucket0Metadata = bucketMap.get(0);
+ assertThat(extractLeaderFromBucketMetadata(bucket0Metadata)).isEqualTo(1);
+ assertThat(bucket0Metadata.getReplicas()).containsExactly(1, 2);
+
+ BucketMetadata bucket1Metadata = bucketMap.get(1);
+ assertThat(extractLeaderFromBucketMetadata(bucket1Metadata)).isEqualTo(2);
+ assertThat(bucket1Metadata.getReplicas()).containsExactly(2, 3);
+ }
+
+ @Test
+ void testBatchGetPartitionMetadataFromZkAsync() throws Exception {
+ // Prepare test data with multiple tables and partitions
+ TablePath tablePath1 = TablePath.of("test_db", "table1");
+ TablePath tablePath2 = TablePath.of("test_db", "table2");
+ long tableId1 = 3001L;
+ long tableId2 = 3002L;
+
+ long currentMillis = System.currentTimeMillis();
+ TableRegistration tableReg1 =
+ createTestTableRegistration(tableId1, "table1", currentMillis);
+ TableRegistration tableReg2 =
+ createTestTableRegistration(tableId2, "table2", currentMillis);
+
+ zookeeperClient.registerTable(tablePath1, tableReg1);
+ zookeeperClient.registerTable(tablePath2, tableReg2);
+
+ // Create partitions for table1
+ long partitionId1 = 11L;
+ long partitionId2 = 12L;
+ String partitionName1 = "p1";
+ String partitionName2 = "p2";
+
+ PartitionAssignment partitionAssignment1 =
+ new PartitionAssignment(
+ tableId1, Collections.singletonMap(0, BucketAssignment.of(1, 2)));
+ PartitionAssignment partitionAssignment2 =
+ new PartitionAssignment(
+ tableId1, Collections.singletonMap(1, BucketAssignment.of(2, 3)));
+
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId1, partitionName1, partitionAssignment1, tablePath1, tableId1);
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId2, partitionName2, partitionAssignment2, tablePath1, tableId1);
+
+ // Create partition for table2
+ long partitionId3 = 21L;
+ String partitionName3 = "p1";
+
+ PartitionAssignment partitionAssignment3 =
+ new PartitionAssignment(
+ tableId2, Collections.singletonMap(0, BucketAssignment.of(1, 3)));
+
+ zookeeperClient.registerPartitionAssignmentAndMetadata(
+ partitionId3, partitionName3, partitionAssignment3, tablePath2, tableId2);
+
+ // Create leader and isr for all partition buckets
+ TableBucket bucket1 = new TableBucket(tableId1, partitionId1, 0);
+ TableBucket bucket2 = new TableBucket(tableId1, partitionId2, 1);
+ TableBucket bucket3 = new TableBucket(tableId2, partitionId3, 0);
+
+ zookeeperClient.registerLeaderAndIsr(
+ bucket1, new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000));
+ zookeeperClient.registerLeaderAndIsr(
+ bucket2, new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000));
+ zookeeperClient.registerLeaderAndIsr(
+ bucket3, new LeaderAndIsr(1, 30, Arrays.asList(1, 3), 300, 3000));
+
+ // Test getPartitionsMetadataFromZK
+ List<PhysicalTablePath> partitionPaths =
+ Arrays.asList(
+ PhysicalTablePath.of(tablePath1, partitionName1),
+ PhysicalTablePath.of(tablePath1, partitionName2),
+ PhysicalTablePath.of(tablePath2, partitionName3));
+ List<PartitionMetadata> partitionsMetadataFromZK =
+ metadataProvider.getPartitionsMetadataFromZK(partitionPaths);
+
+ assertThat(partitionsMetadataFromZK).hasSize(3);
+
+ // Verify partition metadata
+ Map<Long, PartitionMetadata> partitionMap =
+ partitionsMetadataFromZK.stream()
+ .collect(Collectors.toMap(PartitionMetadata::getPartitionId, p -> p));
+
+ // Verify partition 1
+ PartitionMetadata partition1Metadata = partitionMap.get(partitionId1);
+ assertThat(partition1Metadata.getTableId()).isEqualTo(tableId1);
+ assertThat(partition1Metadata.getPartitionName()).isEqualTo(partitionName1);
+ assertThat(partition1Metadata.getBucketMetadataList()).hasSize(1);
+ assertThat(partition1Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(0);
+ assertThat(
+ extractLeaderFromBucketMetadata(
+ partition1Metadata.getBucketMetadataList().get(0)))
+ .isEqualTo(1);
+
+ // Verify partition 2
+ PartitionMetadata partition2Metadata = partitionMap.get(partitionId2);
+ assertThat(partition2Metadata.getTableId()).isEqualTo(tableId1);
+ assertThat(partition2Metadata.getPartitionName()).isEqualTo(partitionName2);
+ assertThat(partition2Metadata.getBucketMetadataList()).hasSize(1);
+ assertThat(partition2Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(1);
+ assertThat(
+ extractLeaderFromBucketMetadata(
+ partition2Metadata.getBucketMetadataList().get(0)))
+ .isEqualTo(2);
+
+ // Verify partition 3
+ PartitionMetadata partition3Metadata = partitionMap.get(partitionId3);
+ assertThat(partition3Metadata.getTableId()).isEqualTo(tableId2);
+ assertThat(partition3Metadata.getPartitionName()).isEqualTo(partitionName3);
+ assertThat(partition3Metadata.getBucketMetadataList()).hasSize(1);
+ assertThat(partition3Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(0);
+ assertThat(
+ extractLeaderFromBucketMetadata(
+ partition3Metadata.getBucketMetadataList().get(0)))
+ .isEqualTo(1);
+ }
+
+ @Test
+ void testBatchGetPartitionMetadataFromZkWithNonExistentPartition() {
+ // Test error handling when partition doesn't exist
+ PhysicalTablePath nonexistPath = PhysicalTablePath.of("test_db", "table1", "nonexist");
+ assertThatThrownBy(
+ () ->
+ metadataProvider.getPartitionsMetadataFromZK(
+ Collections.singleton(nonexistPath)))
+ .isInstanceOf(PartitionNotExistException.class)
+ .hasMessageContaining(
+ "Table partition 'test_db.table1(p=nonexist)' does not exist");
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Helper methods for test data creation
+ // --------------------------------------------------------------------------------------------
+
+ private TableRegistration createTestTableRegistration(
+ long tableId, String comment, long currentMillis) {
+ Map<String, String> options = new HashMap<>();
+ options.put("option-1", "100");
+ return new TableRegistration(
+ tableId,
+ comment,
+ Arrays.asList("a", "b"),
+ new TableDescriptor.TableDistribution(3, Collections.singletonList("a")),
+ options,
+ Collections.emptyMap(),
+ currentMillis,
+ currentMillis);
+ }
+
+ /**
+ * Helper method to extract leader ID from BucketMetadata, handling OptionalInt. Returns null if
+ * leader is not present.
+ */
+ private static Integer extractLeaderFromBucketMetadata(BucketMetadata bucketMetadata) {
+ return bucketMetadata.getLeaderId().isPresent()
+ ? bucketMetadata.getLeaderId().getAsInt()
+ : null;
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
index 88e7ec065..8a97596c7 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
@@ -194,8 +194,7 @@ public class ReplicaTestBase {
conf, zkClient, logManager, TestingMetricGroups.TABLET_SERVER_METRICS);
kvManager.startup();
- serverMetadataCache =
- new TabletServerMetadataCache(new MetadataManager(zkClient, conf), zkClient);
+ serverMetadataCache = new TabletServerMetadataCache(new MetadataManager(zkClient, conf));
initMetadataCache(serverMetadataCache);
rpcClient = RpcClient.create(conf, TestingClientMetricGroup.newInstance(), false);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
index 38c4e075c..ba6e9d70d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
@@ -412,7 +412,7 @@ public class ReplicaFetcherThreadTest {
null,
zkClient,
serverId,
- new TabletServerMetadataCache(new MetadataManager(null, conf), null),
+ new TabletServerMetadataCache(new MetadataManager(null, conf)),
RpcClient.create(conf, TestingClientMetricGroup.newInstance(), false),
TestingMetricGroups.TABLET_SERVER_METRICS,
manualClock);
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 454aac244..afd10c943 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
@@ -18,7 +18,6 @@
package org.apache.fluss.server.tablet;
import org.apache.fluss.config.ConfigOptions;
-import org.apache.fluss.exception.CacheMissException;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.InvalidRequiredAcksException;
import org.apache.fluss.metadata.LogFormat;
@@ -31,14 +30,11 @@ import org.apache.fluss.record.DefaultKvRecordBatch;
import org.apache.fluss.record.DefaultValueRecordBatch;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
import org.apache.fluss.row.encode.ValueEncoder;
-import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.FetchLogResponse;
import org.apache.fluss.rpc.messages.InitWriterRequest;
import org.apache.fluss.rpc.messages.InitWriterResponse;
import org.apache.fluss.rpc.messages.ListOffsetsResponse;
-import org.apache.fluss.rpc.messages.MetadataRequest;
-import org.apache.fluss.rpc.messages.MetadataResponse;
import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest;
import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse;
import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
@@ -82,7 +78,6 @@ import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutionException;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V0;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V1;
@@ -107,7 +102,6 @@ import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newFetchLogR
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newLimitScanRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newListOffsetsRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newLookupRequest;
-import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newMetadataRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPrefixLookupRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newProduceLogRequest;
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest;
@@ -972,137 +966,4 @@ public class TabletServiceITCase {
return ServerRpcMessageUtils.makeNotifyLeaderAndIsrRequest(
0, Collections.singletonList(reqForBucket));
}
-
- @Test
- void testCacheOnlyTrue_NonExistentTable_ShouldThrowTableNotExistException() {
- AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
-
- // Create request for a non-existent table with cache_only = true
- // Note: For non-existent tables, system throws TableNotExistException
- // rather than CacheMissException since table validation happens first
- TablePath nonExistentTable = TablePath.of("non_existent_db", "non_existent_table");
- MetadataRequest request =
- newMetadataRequest(Collections.singletonList(nonExistentTable)).setCacheOnly(true);
-
- // Execute request and verify TableNotExistException is thrown
- // This demonstrates that cache_only doesn't bypass table existence validation
- assertThatThrownBy(() -> gateway.metadata(request).get())
- .isInstanceOf(ExecutionException.class)
- .hasMessageContaining("does not exist");
- }
-
- @Test
- void testCacheOnlyFalse_NonExistentTable_ShouldNotThrowCacheMissException() {
- AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
-
- // Create request for a non-existent table with cache_only = false (default)
- TablePath nonExistentTable = TablePath.of("non_existent_db", "non_existent_table");
- MetadataRequest request = newMetadataRequest(Collections.singletonList(nonExistentTable));
-
- // Execute request - should not throw CacheMissException but may throw other exceptions
- // for non-existent table (which is expected behavior)
- assertThatThrownBy(() -> gateway.metadata(request).get())
- .isInstanceOf(ExecutionException.class);
-
- // Verify that the exception is NOT CacheMissException
- try {
- gateway.metadata(request).get();
- } catch (ExecutionException | InterruptedException e) {
- assertThat(e.getCause()).isNotInstanceOf(CacheMissException.class);
- }
- }
-
- @Test
- void testCacheOnlyTrue_ExistingTable_ShouldReturnFromCache() throws Exception {
- // First create a table to populate cache
- long tableId =
- createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
- TableBucket tb = new TableBucket(tableId, 0);
- FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
-
- // Wait for metadata to be available in all gateways
- FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
-
- AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
-
- // Create request for existing table with cache_only = true
- MetadataRequest request =
- newMetadataRequest(Collections.singletonList(DATA1_TABLE_PATH)).setCacheOnly(true);
-
- // Execute request - should succeed and return metadata from cache
- MetadataResponse response = gateway.metadata(request).get();
- assertThat(response).isNotNull();
- assertThat(response.hasCoordinatorServer()).isTrue();
- assertThat(response.getTabletServersCount()).isGreaterThan(0);
- }
-
- @Test
- void testCacheOnlyTrue_EmptyRequest_ShouldSucceed() throws Exception {
- FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
-
- AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
-
- // Create empty request with cache_only = true
- MetadataRequest request = new MetadataRequest().setCacheOnly(true);
-
- // Execute request - should succeed without any cache misses
- MetadataResponse response = gateway.metadata(request).get();
- assertThat(response).isNotNull();
- // Empty request should return coordinator and tablet servers info
- assertThat(response.hasCoordinatorServer()).isTrue();
- assertThat(response.getTabletServersCount()).isGreaterThan(0);
- }
-
- @Test
- void testCacheOnlyTrue_MultipleTables_MixedCacheStatus_ShouldThrowCacheMissException()
- throws Exception {
- // Create one table to ensure it's in cache
- long tableId =
- createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
- TableBucket tb = new TableBucket(tableId, 0);
- FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
- FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
-
- AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
-
- // Create request with both existing table (in cache) and non-existent table (not in cache)
- List<TablePath> tablePaths =
- Arrays.asList(
- DATA1_TABLE_PATH, // exists in cache
- TablePath.of("non_existent_db", "non_existent_table") // not in cache
- );
- MetadataRequest request = newMetadataRequest(tablePaths).setCacheOnly(true);
-
- // Execute request and verify that an exception is thrown
- // Note: With existing + non-existent tables, TableNotExistException is thrown first
- // for the non-existent table, rather than CacheMissException
- assertThatThrownBy(() -> gateway.metadata(request).get())
- .isInstanceOf(ExecutionException.class)
- .hasMessageContaining("does not exist");
- }
-
- @Test
- void testCacheOnlyTrue_RequestPartitionIds_CacheMiss_ShouldThrowCacheMissException()
- throws Exception {
- // First create a table
- long tableId =
- createTable(FLUSS_CLUSTER_EXTENSION, DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR);
- TableBucket tb = new TableBucket(tableId, 0);
- FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tb);
- FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
-
- AdminReadOnlyGateway gateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
-
- // Create request with cache_only = true, but requesting non-existent partition ID
- // This should trigger CacheMissException since the partition ID doesn't exist in cache
- MetadataRequest request =
- newMetadataRequest(Collections.singletonList(DATA1_TABLE_PATH)).setCacheOnly(true);
- request.addPartitionsId(99999L); // Non-existent partition ID
-
- // Execute request and verify CacheMissException is thrown
- assertThatThrownBy(() -> gateway.metadata(request).get())
- .isInstanceOf(ExecutionException.class)
- .hasCauseInstanceOf(CacheMissException.class)
- .hasMessageContaining("is not available in local cache");
- }
}
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 ded494026..9c15121e3 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
@@ -21,7 +21,6 @@ import org.apache.fluss.cluster.Endpoint;
import org.apache.fluss.cluster.TabletServerInfo;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
-import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
@@ -29,8 +28,6 @@ import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePartition;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.server.entity.RegisterTableBucketLeadAndIsrInfo;
-import org.apache.fluss.server.metadata.BucketMetadata;
-import org.apache.fluss.server.metadata.PartitionMetadata;
import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.BucketSnapshot;
import org.apache.fluss.server.zk.data.CoordinatorAddress;
@@ -59,12 +56,10 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import static org.apache.fluss.server.utils.TableAssignmentUtils.generateAssignment;
@@ -581,263 +576,4 @@ class ZooKeeperClientTest {
.isEqualTo("zookeeper2");
}
}
-
- @Test
- void testGetTableMetadataFromZkAsync() throws Exception {
- // Prepare test data using TestData constants
- TablePath tablePath = TablePath.of("test_db", "test_table");
- long tableId = 1001L;
-
- // Create table assignment
- TableAssignment tableAssignment =
- TableAssignment.builder()
- .add(0, BucketAssignment.of(1, 2, 3))
- .add(1, BucketAssignment.of(2, 3, 4))
- .build();
-
- zookeeperClient.registerTableAssignment(tableId, tableAssignment);
-
- // Create leader and isr for buckets
- TableBucket tableBucket0 = new TableBucket(tableId, 0);
- TableBucket tableBucket1 = new TableBucket(tableId, 1);
-
- LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3), 100, 1000);
- LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3, 4), 200, 2000);
-
- zookeeperClient.registerLeaderAndIsr(tableBucket0, leaderAndIsr0);
- zookeeperClient.registerLeaderAndIsr(tableBucket1, leaderAndIsr1);
-
- // Test getTableMetadataFromZkAsync
- CompletableFuture<List<BucketMetadata>> future =
- zookeeperClient.getTableMetadataFromZkAsync(tablePath, tableId, false);
-
- List<BucketMetadata> bucketMetadataList = future.get();
-
- assertThat(bucketMetadataList).hasSize(2);
-
- // Verify bucket metadata
- Map<Integer, BucketMetadata> bucketMap =
- bucketMetadataList.stream()
- .collect(Collectors.toMap(BucketMetadata::getBucketId, b -> b));
-
- BucketMetadata bucket0Metadata = bucketMap.get(0);
- assertThat(extractLeaderFromBucketMetadata(bucket0Metadata)).isEqualTo(1);
- assertThat(bucket0Metadata.getReplicas()).containsExactly(1, 2, 3);
-
- BucketMetadata bucket1Metadata = bucketMap.get(1);
- assertThat(extractLeaderFromBucketMetadata(bucket1Metadata)).isEqualTo(2);
- assertThat(bucket1Metadata.getReplicas()).containsExactly(2, 3, 4);
- }
-
- @Test
- void testGetPartitionMetadataFromZkAsync() throws Exception {
- // Prepare test data
- TablePath tablePath = TablePath.of("test_db", "test_partition_table");
- long tableId = 2001L;
- long partitionId = 1L;
- String partitionName = "p1";
-
- long currentMillis = System.currentTimeMillis();
- TableRegistration tableReg =
- createTestTableRegistration(tableId, "partition table", currentMillis);
- zookeeperClient.registerTable(tablePath, tableReg);
-
- // Create partition assignment
- Map<Integer, BucketAssignment> bucketAssignments = new HashMap<>();
- bucketAssignments.put(0, BucketAssignment.of(1, 2));
- bucketAssignments.put(1, BucketAssignment.of(2, 3));
- PartitionAssignment partitionAssignment =
- new PartitionAssignment(tableId, bucketAssignments);
-
- zookeeperClient.registerPartitionAssignmentAndMetadata(
- partitionId, partitionName, partitionAssignment, tablePath, tableId);
-
- // Create leader and isr for partition buckets
- TableBucket partitionBucket0 = new TableBucket(tableId, partitionId, 0);
- TableBucket partitionBucket1 = new TableBucket(tableId, partitionId, 1);
-
- LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000);
- LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000);
-
- zookeeperClient.registerLeaderAndIsr(partitionBucket0, leaderAndIsr0);
- zookeeperClient.registerLeaderAndIsr(partitionBucket1, leaderAndIsr1);
-
- // Test getPartitionMetadataFromZkAsync
- PhysicalTablePath partitionPath = PhysicalTablePath.of(tablePath, partitionName);
- CompletableFuture<PartitionMetadata> future =
- zookeeperClient.getPartitionMetadataFromZkAsync(partitionPath);
-
- PartitionMetadata partitionMetadata = future.get();
-
- assertThat(partitionMetadata.getTableId()).isEqualTo(tableId);
- assertThat(partitionMetadata.getPartitionName()).isEqualTo(partitionName);
- assertThat(partitionMetadata.getPartitionId()).isEqualTo(partitionId);
-
- List<BucketMetadata> bucketMetadataList = partitionMetadata.getBucketMetadataList();
- assertThat(bucketMetadataList).hasSize(2);
-
- Map<Integer, BucketMetadata> bucketMap =
- bucketMetadataList.stream()
- .collect(Collectors.toMap(BucketMetadata::getBucketId, b -> b));
-
- BucketMetadata bucket0Metadata = bucketMap.get(0);
- assertThat(extractLeaderFromBucketMetadata(bucket0Metadata)).isEqualTo(1);
- assertThat(bucket0Metadata.getReplicas()).containsExactly(1, 2);
-
- BucketMetadata bucket1Metadata = bucketMap.get(1);
- assertThat(extractLeaderFromBucketMetadata(bucket1Metadata)).isEqualTo(2);
- assertThat(bucket1Metadata.getReplicas()).containsExactly(2, 3);
- }
-
- @Test
- void testBatchGetPartitionMetadataFromZkAsync() throws Exception {
- // Prepare test data with multiple tables and partitions
- TablePath tablePath1 = TablePath.of("test_db", "table1");
- TablePath tablePath2 = TablePath.of("test_db", "table2");
- long tableId1 = 3001L;
- long tableId2 = 3002L;
-
- long currentMillis = System.currentTimeMillis();
- TableRegistration tableReg1 =
- createTestTableRegistration(tableId1, "table1", currentMillis);
- TableRegistration tableReg2 =
- createTestTableRegistration(tableId2, "table2", currentMillis);
-
- zookeeperClient.registerTable(tablePath1, tableReg1);
- zookeeperClient.registerTable(tablePath2, tableReg2);
-
- // Create partitions for table1
- long partitionId1 = 11L;
- long partitionId2 = 12L;
- String partitionName1 = "p1";
- String partitionName2 = "p2";
-
- PartitionAssignment partitionAssignment1 =
- new PartitionAssignment(
- tableId1, Collections.singletonMap(0, BucketAssignment.of(1, 2)));
- PartitionAssignment partitionAssignment2 =
- new PartitionAssignment(
- tableId1, Collections.singletonMap(1, BucketAssignment.of(2, 3)));
-
- zookeeperClient.registerPartitionAssignmentAndMetadata(
- partitionId1, partitionName1, partitionAssignment1, tablePath1, tableId1);
- zookeeperClient.registerPartitionAssignmentAndMetadata(
- partitionId2, partitionName2, partitionAssignment2, tablePath1, tableId1);
-
- // Create partition for table2
- long partitionId3 = 21L;
- String partitionName3 = "p1";
-
- PartitionAssignment partitionAssignment3 =
- new PartitionAssignment(
- tableId2, Collections.singletonMap(0, BucketAssignment.of(1, 3)));
-
- zookeeperClient.registerPartitionAssignmentAndMetadata(
- partitionId3, partitionName3, partitionAssignment3, tablePath2, tableId2);
-
- // Create leader and isr for all partition buckets
- TableBucket bucket1 = new TableBucket(tableId1, partitionId1, 0);
- TableBucket bucket2 = new TableBucket(tableId1, partitionId2, 1);
- TableBucket bucket3 = new TableBucket(tableId2, partitionId3, 0);
-
- zookeeperClient.registerLeaderAndIsr(
- bucket1, new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000));
- zookeeperClient.registerLeaderAndIsr(
- bucket2, new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000));
- zookeeperClient.registerLeaderAndIsr(
- bucket3, new LeaderAndIsr(1, 30, Arrays.asList(1, 3), 300, 3000));
-
- // Test batchGetPartitionMetadataFromZkAsync
- List<TablePath> tablePaths = Arrays.asList(tablePath1, tablePath2);
- Set<Long> partitionIdSet =
- new HashSet<>(Arrays.asList(partitionId1, partitionId2, partitionId3));
-
- CompletableFuture<List<PartitionMetadata>> future =
- zookeeperClient.batchGetPartitionMetadataFromZkAsync(tablePaths, partitionIdSet);
-
- List<PartitionMetadata> partitionMetadataList = future.get();
-
- assertThat(partitionMetadataList).hasSize(3);
-
- // Verify partition metadata
- Map<Long, PartitionMetadata> partitionMap =
- partitionMetadataList.stream()
- .collect(Collectors.toMap(PartitionMetadata::getPartitionId, p -> p));
-
- // Verify partition 1
- PartitionMetadata partition1Metadata = partitionMap.get(partitionId1);
- assertThat(partition1Metadata.getTableId()).isEqualTo(tableId1);
- assertThat(partition1Metadata.getPartitionName()).isEqualTo(partitionName1);
- assertThat(partition1Metadata.getBucketMetadataList()).hasSize(1);
- assertThat(partition1Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(0);
- assertThat(
- extractLeaderFromBucketMetadata(
- partition1Metadata.getBucketMetadataList().get(0)))
- .isEqualTo(1);
-
- // Verify partition 2
- PartitionMetadata partition2Metadata = partitionMap.get(partitionId2);
- assertThat(partition2Metadata.getTableId()).isEqualTo(tableId1);
- assertThat(partition2Metadata.getPartitionName()).isEqualTo(partitionName2);
- assertThat(partition2Metadata.getBucketMetadataList()).hasSize(1);
- assertThat(partition2Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(1);
- assertThat(
- extractLeaderFromBucketMetadata(
- partition2Metadata.getBucketMetadataList().get(0)))
- .isEqualTo(2);
-
- // Verify partition 3
- PartitionMetadata partition3Metadata = partitionMap.get(partitionId3);
- assertThat(partition3Metadata.getTableId()).isEqualTo(tableId2);
- assertThat(partition3Metadata.getPartitionName()).isEqualTo(partitionName3);
- assertThat(partition3Metadata.getBucketMetadataList()).hasSize(1);
- assertThat(partition3Metadata.getBucketMetadataList().get(0).getBucketId()).isEqualTo(0);
- assertThat(
- extractLeaderFromBucketMetadata(
- partition3Metadata.getBucketMetadataList().get(0)))
- .isEqualTo(1);
- }
-
- @Test
- void testBatchGetPartitionMetadataFromZkAsyncWithNonExistentPartition() {
- // Test error handling when partition doesn't exist
- TablePath tablePath = TablePath.of("test_db", "table1");
- Set<Long> nonExistentPartitionIds = new HashSet<>(Arrays.asList(999L, 1000L));
-
- CompletableFuture<List<PartitionMetadata>> future =
- zookeeperClient.batchGetPartitionMetadataFromZkAsync(
- Collections.singletonList(tablePath), nonExistentPartitionIds);
-
- assertThatThrownBy(future::get)
- .hasCauseInstanceOf(org.apache.fluss.exception.PartitionNotExistException.class);
- }
-
- // --------------------------------------------------------------------------------------------
- // Helper methods for test data creation
- // --------------------------------------------------------------------------------------------
-
- private TableRegistration createTestTableRegistration(
- long tableId, String comment, long currentMillis) {
- Map<String, String> options = new HashMap<>();
- options.put("option-1", "100");
- return new TableRegistration(
- tableId,
- comment,
- Arrays.asList("a", "b"),
- new TableDescriptor.TableDistribution(3, Collections.singletonList("a")),
- options,
- Collections.emptyMap(),
- currentMillis,
- currentMillis);
- }
-
- /**
- * Helper method to extract leader ID from BucketMetadata, handling OptionalInt. Returns null if
- * leader is not present.
- */
- private static Integer extractLeaderFromBucketMetadata(BucketMetadata bucketMetadata) {
- return bucketMetadata.getLeaderId().isPresent()
- ? bucketMetadata.getLeaderId().getAsInt()
- : null;
- }
}