[client] Add Admin API to describe buckets - #4029
Conversation
|
|
||
| message PbBucketInfo { | ||
| required PbTablePath table_path = 1; | ||
| required int64 table_id = 2; |
There was a problem hiding this comment.
I think we should move out the two table_path and table_id from here into the above DescribeBucketsResponse.
message DescribeBucketsResponse {
required PbTablePath table_path = 1;
required int64 table_id = 2;
repeated PbBucketInfo bucket_info = 3;
}
The DescribeBucketsResponse answers for one table (it carries table_path), but this PbBucketInfo repeats table_path message and the table_id.
Given a table with many partitions the same two strings are serialized once per bucket (e.g., 1000 partitions x 16 buckets = 16k copies).
There was a problem hiding this comment.
Good point. I moved table_path and table_id to DescribeBucketsResponse and updated the server/client conversion, so they are serialized only once per response. Thanks!
| } | ||
|
|
||
| return partitions; | ||
| List<String> matchedPartitionNames = |
There was a problem hiding this comment.
Was there any reason for these change?
I think we have behaviour change here, the processGetDataResponses only logs warning for Zookeeper decode errors. The previous getPartition / getOrEmpty methods rethrows ZK errors (except for NoNode).
So if one znode read fails for a transient reason, the caller now gets back a partition list that looks complete but isn't, instead of an error. For an admin describe call I'd rather fail the request than silently omit partitions.
This also isn't limited to the new API: the refactor changes the existing listPartitionInfos RPC too.
There was a problem hiding this comment.
Good catch. I kept the batched reads but restored the previous semantics: NONODE is ignored, while all other ZooKeeper errors fail the request. I also added tests for NONODE and CONNECTIONLOSS.
| return zookeeperClient.getPartitionRegistrations( | ||
| tablePath, tableInfo.getPartitionKeys(), partitionFilter); | ||
| } | ||
| } catch (ApiException e) { |
There was a problem hiding this comment.
Why this change is needed?
There was a problem hiding this comment.
This is needed to preserve InvalidPartitionException for invalid partial partition specs, as declared by the Admin APIs, instead of wrapping it in FlussRuntimeException. I narrowed the catch from ApiException to InvalidPartitionException to avoid changing other API exception handling.
| TablePath tablePath = toTablePath(request.getTablePath()); | ||
| authorizeTable(OperationType.DESCRIBE, tablePath); | ||
|
|
||
| TableInfo tableInfo = metadataManager.getTable(tablePath); |
There was a problem hiding this comment.
Maybe we can do some refactoring here.
The metadataManager.getTable(tablePath) is again called in the listPartitions / listPartitionInfos methods. Maybe good idea to pass already obtained TableInfo 🤝
There was a problem hiding this comment.
Good suggestion. I added an overload that accepts the already-fetched TableInfo and now reuse it in both describeBuckets and listPartitionInfos, avoiding the duplicate lookup.
| "Failed to list partitions of table fluss.partitioned_t1 in test-catalog, by partitionSpec CatalogPartitionSpec{{second=}}"); | ||
| .isInstanceOf(PartitionSpecInvalidException.class) | ||
| .hasMessageContaining( | ||
| "PartitionSpec CatalogPartitionSpec{{second=}} does not match"); |
There was a problem hiding this comment.
It is unclear to me why we are changing the existing tests, I think this is related to the new throw in MetadataManager below, but hard to understand why it was required.
There was a problem hiding this comment.
This test change follows the intentional exception propagation above: InvalidPartitionException is now mapped by FlinkCatalog to its declared PartitionSpecInvalidException instead of falling through to CatalogException. I kept the updated assertion and verified the full FlinkCatalogTest suite.
There was a problem hiding this comment.
Got it, thanks for explanation 🤝
|
please help review, thanks! @morazow @loserwang1024 |
| "Failed to list partitions of table fluss.partitioned_t1 in test-catalog, by partitionSpec CatalogPartitionSpec{{second=}}"); | ||
| .isInstanceOf(PartitionSpecInvalidException.class) | ||
| .hasMessageContaining( | ||
| "PartitionSpec CatalogPartitionSpec{{second=}} does not match"); |
There was a problem hiding this comment.
Got it, thanks for explanation 🤝
|
|
||
| /** Returns the partition name, or {@code null} for a non-partitioned table. */ | ||
| @Nullable | ||
| public String getPartitionName() { |
There was a problem hiding this comment.
nit: would be good idea to return Optional<String> here, to keep consisted with other getters?
| return CompletableFuture.completedFuture(response); | ||
| } | ||
|
|
||
| private Map<String, PartitionRegistration> listPartitionsForDescribeBuckets( |
There was a problem hiding this comment.
nit: I wonder if we can consolidate the three ...ForDescribeBuckets methods into one.
Use e.g.,
private <T> T readBucketMetadata(TablePath tablePath, Callable<T> zkRead) {
try {
return zkClient.getBucketMetadataForPartitions(partitionIds);
return zkRead.call();
} catch (Exception e) {
throw new FlussRuntimeException(
String.format("Failed to describe buckets for table '%s'.", tablePath), e);
}
}For listPartitionsForDescribeBuckets, directly call the metadataManager with ternary on the parition spec:
metadataManager.listPartitions(...)For other two getPartitionBucket.. and getTableBucket... methods use above readBucketMetadata method with zk client callable.
What do you think?
| return getPartitionRegistrations(tablePath, matchedPartitionNames); | ||
| } | ||
|
|
||
| private Map<String, PartitionRegistration> getPartitionRegistrations( |
There was a problem hiding this comment.
nit: comment for clarifying why we don't use processGetDataResponses
// Deliberately not using processGetDataResponses here: that helper logs and skips
// failed responses, which would silently return an incomplete partition list.
// This method propagates errors, only {@code NONODE} is tolerated.
Add maybe add why we tolerate NONODE. Is it because of concurrently dropping partitions?
There was a problem hiding this comment.
Pull request overview
Adds a new public Admin API and corresponding RPC endpoint to retrieve bucket-level metadata (leader/epoch, replicas, ISR) for tables, supporting both non-partitioned tables and partitioned tables with optional (partial or complete) partition filtering. This extends Fluss’s administrative introspection capabilities and provides the foundation needed for the planned Flink sys.describe_buckets procedure (to be added in a follow-up PR).
Changes:
- Introduces a new public
BucketInfoDTO and addsAdmin#describeBuckets(...)overloads returning bucket metadata. - Adds the public
DESCRIBE_BUCKETSRPC (API key 1065), protobuf messages, gateway wiring, and client/server conversion/handling. - Improves/extends ZooKeeper-based partition/bucket metadata retrieval (including batched reads) and adds unit/integration/authorization coverage.
Reviewed changes
Copilot reviewed 22 out of 22 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java | Extends ZK client tests to cover partition registration batch reads and error preservation. |
| fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java | Adds describeBuckets stub to keep gateway test implementation in sync with new RPC. |
| fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java | Adds describeBuckets stub to keep gateway test implementation in sync with new RPC. |
| fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java | Refactors partition registration reads to a batched getDataInBackground implementation; includes ISR in bucket metadata creation. |
| fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java | Implements server-side describeBuckets RPC handling with authorization, partition filtering, ZK reads, and deterministic ordering. |
| fluss-server/src/main/java/org/apache/fluss/server/metadata/BucketMetadata.java | Extends internal bucket metadata to carry ISR and makes replica/ISR lists immutable copies. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java | Adds a listPartitions overload that reuses provided TableInfo and preserves InvalidPartitionException without wrapping. |
| fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java | Adds describeBuckets stub for RPC test gateway service compatibility. |
| fluss-rpc/src/main/proto/FlussApi.proto | Adds DescribeBucketsRequest/Response and the PbBucketInfo wire message (including ISR). |
| fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java | Registers new public API key DESCRIBE_BUCKETS(1065). |
| fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java | Exposes describeBuckets on the admin read-only RPC gateway. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java | Updates Flink-side test adapter to compile with new Admin#describeBuckets API. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java | Updates expected exception type/message for invalid partition spec handling. |
| fluss-common/src/test/java/org/apache/fluss/metadata/BucketInfoTest.java | Adds unit tests for the new public BucketInfo DTO (immutability + null checks). |
| fluss-common/src/main/java/org/apache/fluss/metadata/BucketInfo.java | Introduces the public BucketInfo DTO for bucket replica/leader/ISR metadata. |
| fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java | Adds tests for converting DescribeBucketsResponse to List<BucketInfo>. |
| fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java | Extends authorization IT coverage to include Admin#describeBuckets. |
| fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java | Updates partition listing invalid-spec assertion to expect InvalidPartitionException. |
| fluss-client/src/test/java/org/apache/fluss/client/admin/DescribeBucketsITCase.java | Adds integration tests for describe-buckets behavior across partitioned/non-partitioned tables and filters. |
| fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java | Adds conversion from RPC DescribeBucketsResponse/PbBucketInfo into public BucketInfo. |
| fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java | Implements new Admin#describeBuckets methods via the read-only gateway and RPC conversion. |
| fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java | Adds the two new public describeBuckets method signatures with Javadoc. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| + ", partitionName='" | ||
| + partitionName | ||
| + '\'' | ||
| + ", bucketId=" |
|
@fhan688 hello! I performed a high-level comparison between this PR and #3785. #3785 makes the internal metadata path authoritative (ISR +
Two small things while here:
|
Purpose
Linked issue: #3436
Fluss currently does not provide a public Admin API for retrieving bucket-level metadata, including replica placement and leader/ISR state. This information is required by administrative tools and by the planned Flink
sys.describe_bucketsprocedure.This is the first of two PRs planned for #3436. It introduces the core Admin API, RPC contract, and server-side implementation. A follow-up PR will add the Flink procedure and close the issue.
Brief change log
BucketInforesult type, exposing the table path and ID, optional partition ID and name, bucket ID, optional leader and leader epoch, replicas, and ISR.Admin#describeBuckets(TablePath)andAdmin#describeBuckets(TablePath, PartitionSpec).DESCRIBE_BUCKETSRPC (API key 1065), including request/response messages, gateway wiring, and client-side conversion.DESCRIBEpermission and preserve table/partition validation exceptions at the Admin API boundary.The Flink
sys.describe_bucketsprocedure is intentionally excluded from this PR and will be added in the follow-up PR.Tests
BucketInfoTestClientRpcMessageUtilsTestApiKeysTestApiManagerTestZooKeeperClientTest#testPartitionDescribeBucketsITCaseFlussAuthorizationITCase#testDescribeTableOperationfluss-flink-commontest compilation forAdmininterface compatibilitygit diff --checkAPI and Format
@PublicEvolvingBucketInfoclass and two new methods toAdmin.DESCRIBE_BUCKETSAPI (key 1065) and its protobuf messages.PbBucketMetadatawire message is unchanged, avoiding overlap with the separate ISR metadata work in feat: shell tool #3360.Documentation
Public API Javadocs are included for
BucketInfoand the newAdminmethods.No user-facing Flink procedure documentation is added in this PR because the procedure is outside this PR's scope. The follow-up PR will add the procedure, integration tests, and corresponding documentation.