[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.
2 participants
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.