From 93a9711d99387b77f2afbf393af58ad0a366f1cc Mon Sep 17 00:00:00 2001 From: fhan Date: Tue, 18 Aug 2026 16:23:28 +0800 Subject: [PATCH 1/2] [client] Add Admin API to describe buckets --- .../org/apache/fluss/client/admin/Admin.java | 41 ++++ .../apache/fluss/client/admin/FlussAdmin.java | 29 +++ .../client/utils/ClientRpcMessageUtils.java | 24 ++ .../client/admin/DescribeBucketsITCase.java | 215 ++++++++++++++++++ .../fluss/client/admin/FlussAdminITCase.java | 3 +- .../acl/FlussAuthorizationITCase.java | 12 +- .../utils/ClientRpcMessageUtilsTest.java | 48 ++++ .../org/apache/fluss/metadata/BucketInfo.java | 180 +++++++++++++++ .../apache/fluss/metadata/BucketInfoTest.java | 130 +++++++++++ .../sink/testutils/TestAdminAdapter.java | 12 + .../rpc/gateway/AdminReadOnlyGateway.java | 11 + .../apache/fluss/rpc/protocol/ApiKeys.java | 3 +- fluss-rpc/src/main/proto/FlussApi.proto | 23 ++ .../rpc/TestingTabletGatewayService.java | 8 + .../apache/fluss/server/RpcServiceBase.java | 131 +++++++++++ .../server/coordinator/MetadataManager.java | 3 + .../fluss/server/metadata/BucketMetadata.java | 26 ++- .../fluss/server/zk/ZooKeeperClient.java | 56 +++-- .../coordinator/TestCoordinatorGateway.java | 8 + .../tablet/TestTabletServerGateway.java | 8 + .../fluss/server/zk/ZooKeeperClientTest.java | 5 + 21 files changed, 947 insertions(+), 29 deletions(-) create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/admin/DescribeBucketsITCase.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/metadata/BucketInfo.java create mode 100644 fluss-common/src/test/java/org/apache/fluss/metadata/BucketInfoTest.java diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java index 5d749b6d432..d9c39b45ec1 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java @@ -55,6 +55,7 @@ import org.apache.fluss.exception.TableNotPartitionedException; import org.apache.fluss.exception.TooManyBucketsException; import org.apache.fluss.exception.TooManyPartitionsException; +import org.apache.fluss.metadata.BucketInfo; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; @@ -253,6 +254,46 @@ CompletableFuture createTable( */ CompletableFuture getTableInfo(TablePath tablePath); + /** + * Describes the buckets of the given table asynchronously. + * + *

For a non-partitioned table, this returns the table buckets. For a partitioned table, this + * returns the buckets of all partitions. For a partitioned table with many partitions, prefer + * {@link #describeBuckets(TablePath, PartitionSpec)} to limit the result. + * + *

The following exceptions can be anticipated when calling {@code get()} on the returned + * future. + * + *

    + *
  • {@link TableNotExistException} if the table does not exist. + *
+ * + * @param tablePath The table path of the table. + * @since 1.0 + */ + CompletableFuture> describeBuckets(TablePath tablePath); + + /** + * Describes the buckets matching the given partition spec asynchronously. + * + *

The partition spec may contain all partition keys or a subset of them. + * + *

The following exceptions can be anticipated when calling {@code get()} on the returned + * future. + * + *

    + *
  • {@link TableNotExistException} if the table does not exist. + *
  • {@link TableNotPartitionedException} if the table is not partitioned. + *
  • {@link InvalidPartitionException} if the partition spec is invalid. + *
+ * + * @param tablePath The table path of the table. + * @param partitionSpec The complete or partial partition spec. + * @since 1.0 + */ + CompletableFuture> describeBuckets( + TablePath tablePath, PartitionSpec partitionSpec); + /** * Drop the table with the given table path asynchronously. * diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java index a0909ceec73..40355d8beef 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java @@ -34,6 +34,7 @@ import org.apache.fluss.config.cluster.ConfigEntry; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.LeaderNotAvailableException; +import org.apache.fluss.metadata.BucketInfo; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; @@ -66,6 +67,7 @@ import org.apache.fluss.rpc.messages.DatabaseExistsRequest; import org.apache.fluss.rpc.messages.DatabaseExistsResponse; import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; import org.apache.fluss.rpc.messages.DropAclsRequest; import org.apache.fluss.rpc.messages.DropDatabaseRequest; @@ -347,6 +349,33 @@ public CompletableFuture getTableInfo(TablePath tablePath) { r.getModifiedTime())); } + @Override + public CompletableFuture> describeBuckets(TablePath tablePath) { + tablePath.validate(); + DescribeBucketsRequest request = new DescribeBucketsRequest(); + request.setTablePath() + .setDatabaseName(tablePath.getDatabaseName()) + .setTableName(tablePath.getTableName()); + return readOnlyGateway + .describeBuckets(request) + .thenApply(ClientRpcMessageUtils::toBucketInfos); + } + + @Override + public CompletableFuture> describeBuckets( + TablePath tablePath, PartitionSpec partitionSpec) { + tablePath.validate(); + checkNotNull(partitionSpec, "partitionSpec must not be null"); + DescribeBucketsRequest request = new DescribeBucketsRequest(); + request.setTablePath() + .setDatabaseName(tablePath.getDatabaseName()) + .setTableName(tablePath.getTableName()); + request.setPartitionSpec(makePbPartitionSpec(partitionSpec)); + return readOnlyGateway + .describeBuckets(request) + .thenApply(ClientRpcMessageUtils::toBucketInfos); + } + @Override public CompletableFuture dropTable(TablePath tablePath, boolean ignoreIfNotExists) { DropTableRequest request = new DropTableRequest(); diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java index 3c7512945dc..1064f6e7e5d 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java @@ -42,6 +42,7 @@ import org.apache.fluss.fs.FsPathAndFileName; import org.apache.fluss.fs.token.ObtainedSecurityToken; import org.apache.fluss.metadata.AggFunction; +import org.apache.fluss.metadata.BucketInfo; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseSummary; import org.apache.fluss.metadata.PartitionInfo; @@ -55,6 +56,7 @@ import org.apache.fluss.rpc.messages.AlterDatabaseRequest; import org.apache.fluss.rpc.messages.AlterTableRequest; import org.apache.fluss.rpc.messages.CreatePartitionRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; import org.apache.fluss.rpc.messages.DropPartitionRequest; import org.apache.fluss.rpc.messages.GetClusterHealthResponse; import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse; @@ -73,6 +75,7 @@ import org.apache.fluss.rpc.messages.MetadataRequest; import org.apache.fluss.rpc.messages.PbAddColumn; import org.apache.fluss.rpc.messages.PbAlterConfig; +import org.apache.fluss.rpc.messages.PbBucketInfo; import org.apache.fluss.rpc.messages.PbBucketOffset; import org.apache.fluss.rpc.messages.PbDatabaseSummary; import org.apache.fluss.rpc.messages.PbDescribeConfig; @@ -649,6 +652,27 @@ public static List toPartitionInfos(ListPartitionInfosResponse re .collect(Collectors.toList()); } + public static List toBucketInfos(DescribeBucketsResponse response) { + return response.getBucketInfosList().stream() + .map(ClientRpcMessageUtils::toBucketInfo) + .collect(Collectors.toList()); + } + + private static BucketInfo toBucketInfo(PbBucketInfo pbBucketInfo) { + return new BucketInfo( + TablePath.of( + pbBucketInfo.getTablePath().getDatabaseName(), + pbBucketInfo.getTablePath().getTableName()), + pbBucketInfo.getTableId(), + pbBucketInfo.hasPartitionId() ? pbBucketInfo.getPartitionId() : null, + pbBucketInfo.hasPartitionName() ? pbBucketInfo.getPartitionName() : null, + pbBucketInfo.getBucketId(), + pbBucketInfo.hasLeaderId() ? pbBucketInfo.getLeaderId() : null, + pbBucketInfo.hasLeaderEpoch() ? pbBucketInfo.getLeaderEpoch() : null, + Arrays.stream(pbBucketInfo.getReplicaIds()).boxed().collect(Collectors.toList()), + Arrays.stream(pbBucketInfo.getIsrIds()).boxed().collect(Collectors.toList())); + } + public static Map toKeyValueMap(List pbKeyValues) { return pbKeyValues.stream() .collect( diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/DescribeBucketsITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/DescribeBucketsITCase.java new file mode 100644 index 00000000000..7d0ade2e1a8 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/DescribeBucketsITCase.java @@ -0,0 +1,215 @@ +/* + * 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.client.admin; + +import org.apache.fluss.exception.InvalidPartitionException; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.exception.TableNotPartitionedException; +import org.apache.fluss.metadata.BucketInfo; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Integration test for describing table buckets through {@link Admin}. */ +class DescribeBucketsITCase extends ClientToServerITCaseBase { + + private static final TablePath NON_PARTITIONED_TABLE_PATH = + TablePath.of("test_db", "non_partitioned_table"); + private static final TablePath PARTITIONED_TABLE_PATH = + TablePath.of("test_db", "partitioned_table"); + + @Test + void testDescribeBucketsForNonPartitionedTable() throws Exception { + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .primaryKey("id") + .build()) + .distributedBy(3, "id") + .build(); + long tableId = createTable(NON_PARTITIONED_TABLE_PATH, tableDescriptor, false); + + List bucketInfos = waitAndDescribeBuckets(NON_PARTITIONED_TABLE_PATH, null, 3); + assertThat(bucketInfos).extracting(BucketInfo::getBucketId).containsExactly(0, 1, 2); + bucketInfos.forEach( + bucketInfo -> { + assertBucketInfo(bucketInfo, NON_PARTITIONED_TABLE_PATH, tableId, null); + assertThat(bucketInfo.getPartitionName()).isNull(); + }); + + assertThatThrownBy( + () -> + admin.describeBuckets( + NON_PARTITIONED_TABLE_PATH, + newPartitionSpec("pt", "2025")) + .get()) + .cause() + .isInstanceOf(TableNotPartitionedException.class); + assertThatThrownBy( + () -> admin.describeBuckets(TablePath.of("test_db", "missing_table")).get()) + .cause() + .isInstanceOf(TableNotExistException.class); + } + + @Test + void testDescribeBucketsForPartitionedTable() throws Exception { + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("id", DataTypes.STRING()) + .column("pt", DataTypes.STRING()) + .column("region", DataTypes.STRING()) + .build()) + .distributedBy(2, "id") + .partitionedBy("pt", "region") + .build(); + long tableId = createTable(PARTITIONED_TABLE_PATH, tableDescriptor, false); + PartitionSpec p2025Cn = + newPartitionSpec(Arrays.asList("pt", "region"), Arrays.asList("2025", "cn")); + PartitionSpec p2025Us = + newPartitionSpec(Arrays.asList("pt", "region"), Arrays.asList("2025", "us")); + PartitionSpec p2026Cn = + newPartitionSpec(Arrays.asList("pt", "region"), Arrays.asList("2026", "cn")); + admin.createPartition(PARTITIONED_TABLE_PATH, p2025Cn, false).get(); + admin.createPartition(PARTITIONED_TABLE_PATH, p2025Us, false).get(); + admin.createPartition(PARTITIONED_TABLE_PATH, p2026Cn, false).get(); + + Map partitionIds = + admin.listPartitionInfos(PARTITIONED_TABLE_PATH).get().stream() + .collect( + Collectors.toMap( + PartitionInfo::getPartitionName, + PartitionInfo::getPartitionId)); + + List allPartitionBuckets = + waitAndDescribeBuckets(PARTITIONED_TABLE_PATH, null, 6); + assertThat(allPartitionBuckets) + .extracting( + bucketInfo -> + bucketInfo.getPartitionName() + ":" + bucketInfo.getBucketId()) + .containsExactly( + "2025$cn:0", + "2025$cn:1", + "2025$us:0", + "2025$us:1", + "2026$cn:0", + "2026$cn:1"); + allPartitionBuckets.forEach( + bucketInfo -> + assertBucketInfo( + bucketInfo, + PARTITIONED_TABLE_PATH, + tableId, + partitionIds.get(bucketInfo.getPartitionName()))); + + List partialPartitionBuckets = + waitAndDescribeBuckets(PARTITIONED_TABLE_PATH, newPartitionSpec("pt", "2025"), 4); + assertThat(partialPartitionBuckets) + .extracting(BucketInfo::getPartitionName) + .containsExactly("2025$cn", "2025$cn", "2025$us", "2025$us"); + + List exactPartitionBuckets = + waitAndDescribeBuckets(PARTITIONED_TABLE_PATH, p2025Cn, 2); + assertThat(exactPartitionBuckets) + .extracting(BucketInfo::getPartitionName) + .containsOnly("2025$cn"); + assertThat(exactPartitionBuckets).extracting(BucketInfo::getBucketId).containsExactly(0, 1); + + assertThat( + admin.describeBuckets( + PARTITIONED_TABLE_PATH, newPartitionSpec("pt", "missing")) + .get()) + .isEmpty(); + assertThatThrownBy( + () -> + admin.describeBuckets( + PARTITIONED_TABLE_PATH, + newPartitionSpec("unknown", "2025")) + .get()) + .cause() + .isInstanceOf(InvalidPartitionException.class) + .hasMessageContaining("unknown"); + } + + private List waitAndDescribeBuckets( + TablePath tablePath, @Nullable PartitionSpec partitionSpec, int expectedBucketCount) + throws Exception { + waitUntil( + () -> { + List bucketInfos = describeBuckets(tablePath, partitionSpec); + return bucketInfos.size() == expectedBucketCount + && bucketInfos.stream() + .allMatch( + bucketInfo -> + bucketInfo.getLeaderId().isPresent() + && bucketInfo + .getLeaderEpoch() + .isPresent() + && !bucketInfo.getIsr().isEmpty()); + }, + Duration.ofMinutes(1), + "Waiting for bucket metadata"); + return describeBuckets(tablePath, partitionSpec); + } + + private List describeBuckets( + TablePath tablePath, @Nullable PartitionSpec partitionSpec) throws Exception { + return partitionSpec == null + ? admin.describeBuckets(tablePath).get() + : admin.describeBuckets(tablePath, partitionSpec).get(); + } + + private static void assertBucketInfo( + BucketInfo bucketInfo, + TablePath tablePath, + long tableId, + @Nullable Long expectedPartitionId) { + assertThat(bucketInfo.getTablePath()).isEqualTo(tablePath); + assertThat(bucketInfo.getTableId()).isEqualTo(tableId); + if (expectedPartitionId == null) { + assertThat(bucketInfo.getPartitionId()).isEmpty(); + } else { + assertThat(bucketInfo.getPartitionId()).hasValue(expectedPartitionId); + } + assertThat(bucketInfo.getReplicas()).hasSize(3); + assertThat(bucketInfo.getIsr()).isNotEmpty(); + assertThat(bucketInfo.getReplicas()).containsAll(bucketInfo.getIsr()); + assertThat(bucketInfo.getIsr()).contains(bucketInfo.getLeaderId().getAsInt()); + } +} diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index 93733b32bb8..5c62238f6fc 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java @@ -37,7 +37,6 @@ import org.apache.fluss.exception.DatabaseAlreadyExistException; import org.apache.fluss.exception.DatabaseNotEmptyException; import org.apache.fluss.exception.DatabaseNotExistException; -import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.InvalidAlterTableException; import org.apache.fluss.exception.InvalidConfigException; import org.apache.fluss.exception.InvalidDatabaseException; @@ -1270,7 +1269,7 @@ void testListPartitionInfosByPartitionSpec() throws Exception { admin.listPartitionInfos(partitionedTablePath, invalidPartitionSpec) .get()) .cause() - .isInstanceOf(FlussRuntimeException.class) + .isInstanceOf(InvalidPartitionException.class) .hasMessageContaining("table don't contains this partitionKey: pt1"); } diff --git a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java index bd694f38a1d..3e9d6850c04 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java @@ -420,8 +420,9 @@ void testDescribeTableOperation() throws Exception { // 3. getTableSchema // 4. getLatestKvSnapshots // 5. listPartitionInfos - // 6. getLatestLakeSnapshot - // 7. listOffsets + // 6. describeBuckets + // 7. getLatestLakeSnapshot + // 8. listOffsets // first check call these methods without authorization. assertThat(guestAdmin.listTables(DATA1_TABLE_PATH_PK.getDatabaseName()).get()) @@ -430,6 +431,7 @@ void testDescribeTableOperation() throws Exception { assertNoTableDescribeAuth(() -> guestAdmin.getTableSchema(DATA1_TABLE_PATH_PK).get()); assertNoTableDescribeAuth(() -> guestAdmin.getLatestKvSnapshots(DATA1_TABLE_PATH_PK).get()); assertNoTableDescribeAuth(() -> guestAdmin.listPartitionInfos(DATA1_TABLE_PATH_PK).get()); + assertNoTableDescribeAuth(() -> guestAdmin.describeBuckets(DATA1_TABLE_PATH_PK).get()); assertNoTableDescribeAuth( () -> guestAdmin.getLatestLakeSnapshot(DATA1_TABLE_PATH_PK).get()); assertNoTableDescribeAuth( @@ -465,6 +467,12 @@ void testDescribeTableOperation() throws Exception { assertThat(guestAdmin.tableExists(DATA1_TABLE_PATH_PK).get()).isTrue(); assertThat(guestAdmin.getLatestKvSnapshots(DATA1_TABLE_PATH_PK).get().getBucketIds()) .containsExactlyInAnyOrder(0, 1, 2); + assertThat(guestAdmin.describeBuckets(DATA1_TABLE_PATH_PK).get()) + .hasSize(3) + .allSatisfy( + bucketInfo -> + assertThat(bucketInfo.getTablePath()) + .isEqualTo(DATA1_TABLE_PATH_PK)); assertThatThrownBy(() -> guestAdmin.listPartitionInfos(DATA1_TABLE_PATH_PK).get()) .rootCause() .isInstanceOf(TableNotPartitionedException.class) diff --git a/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java b/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java index 3ed17da7da5..74f926b8913 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/utils/ClientRpcMessageUtilsTest.java @@ -21,9 +21,13 @@ import org.apache.fluss.client.write.ReadyWriteBatch; import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.memory.PreAllocatedPagedOutputView; +import org.apache.fluss.metadata.BucketInfo; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; +import org.apache.fluss.rpc.messages.PbBucketInfo; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.protocol.MergeMode; @@ -126,6 +130,50 @@ void testMakePutKvRequestWithSingleBatch() throws Exception { assertThat(request.getAggMode()).isEqualTo(MergeMode.OVERWRITE.getProtoValue()); } + @Test + void testToBucketInfos() { + DescribeBucketsResponse response = new DescribeBucketsResponse(); + PbBucketInfo tableBucket = + response.addBucketInfo().setTableId(10L).setBucketId(0).setLeaderId(1); + tableBucket.setTablePath().setDatabaseName("db").setTableName("table"); + tableBucket.setLeaderEpoch(7); + tableBucket.addReplicaId(1); + tableBucket.addReplicaId(2); + tableBucket.addReplicaId(3); + tableBucket.addIsrId(1); + tableBucket.addIsrId(3); + + PbBucketInfo partitionBucket = response.addBucketInfo().setTableId(10L).setBucketId(1); + partitionBucket.setTablePath().setDatabaseName("db").setTableName("table"); + partitionBucket.setPartitionId(100L).setPartitionName("p1"); + partitionBucket.addReplicaId(2); + partitionBucket.addReplicaId(3); + + List bucketInfos = ClientRpcMessageUtils.toBucketInfos(response); + + assertThat(bucketInfos).hasSize(2); + BucketInfo tableBucketInfo = bucketInfos.get(0); + assertThat(tableBucketInfo.getTablePath()).isEqualTo(TablePath.of("db", "table")); + assertThat(tableBucketInfo.getTableId()).isEqualTo(10L); + assertThat(tableBucketInfo.getPartitionId()).isEmpty(); + assertThat(tableBucketInfo.getPartitionName()).isNull(); + assertThat(tableBucketInfo.getBucketId()).isEqualTo(0); + assertThat(tableBucketInfo.getLeaderId()).hasValue(1); + assertThat(tableBucketInfo.getLeaderEpoch()).hasValue(7); + assertThat(tableBucketInfo.getReplicas()).containsExactly(1, 2, 3); + assertThat(tableBucketInfo.getIsr()).containsExactly(1, 3); + + BucketInfo partitionBucketInfo = bucketInfos.get(1); + assertThat(partitionBucketInfo.getTablePath()).isEqualTo(TablePath.of("db", "table")); + assertThat(partitionBucketInfo.getPartitionId()).hasValue(100L); + assertThat(partitionBucketInfo.getPartitionName()).isEqualTo("p1"); + assertThat(partitionBucketInfo.getBucketId()).isEqualTo(1); + assertThat(partitionBucketInfo.getLeaderId()).isEmpty(); + assertThat(partitionBucketInfo.getLeaderEpoch()).isEmpty(); + assertThat(partitionBucketInfo.getReplicas()).containsExactly(2, 3); + assertThat(partitionBucketInfo.getIsr()).isEmpty(); + } + private KvWriteBatch createKvWriteBatch(int bucketId, MergeMode mergeMode) throws Exception { MemorySegment segment = MemorySegment.allocateHeapMemory(1024); PreAllocatedPagedOutputView outputView = diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/BucketInfo.java b/fluss-common/src/main/java/org/apache/fluss/metadata/BucketInfo.java new file mode 100644 index 00000000000..e2d6b67a095 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/BucketInfo.java @@ -0,0 +1,180 @@ +/* + * 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.metadata; + +import org.apache.fluss.annotation.PublicEvolving; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.OptionalInt; +import java.util.OptionalLong; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * Information about a physical table bucket, including its replicas and leader/ISR state. + * + * @since 1.0 + */ +@PublicEvolving +public final class BucketInfo { + private final TablePath tablePath; + private final long tableId; + private final @Nullable Long partitionId; + private final @Nullable String partitionName; + private final int bucketId; + private final @Nullable Integer leaderId; + private final @Nullable Integer leaderEpoch; + private final List replicas; + private final List isr; + + /** Creates bucket information. */ + public BucketInfo( + TablePath tablePath, + long tableId, + @Nullable Long partitionId, + @Nullable String partitionName, + int bucketId, + @Nullable Integer leaderId, + @Nullable Integer leaderEpoch, + List replicas, + List isr) { + this.tablePath = checkNotNull(tablePath, "tablePath should not be null."); + this.tableId = tableId; + this.partitionId = partitionId; + this.partitionName = partitionName; + this.bucketId = bucketId; + this.leaderId = leaderId; + this.leaderEpoch = leaderEpoch; + this.replicas = + Collections.unmodifiableList( + new ArrayList<>(checkNotNull(replicas, "replicas should not be null."))); + this.isr = + Collections.unmodifiableList( + new ArrayList<>(checkNotNull(isr, "isr should not be null."))); + } + + /** Returns the table path. */ + public TablePath getTablePath() { + return tablePath; + } + + /** Returns the table ID. */ + public long getTableId() { + return tableId; + } + + /** Returns the partition ID, or an empty optional for a non-partitioned table. */ + public OptionalLong getPartitionId() { + return partitionId == null ? OptionalLong.empty() : OptionalLong.of(partitionId); + } + + /** Returns the partition name, or {@code null} for a non-partitioned table. */ + @Nullable + public String getPartitionName() { + return partitionName; + } + + /** Returns the bucket ID. */ + public int getBucketId() { + return bucketId; + } + + /** Returns the leader ID, or an empty optional if no leader has been elected. */ + public OptionalInt getLeaderId() { + return leaderId == null ? OptionalInt.empty() : OptionalInt.of(leaderId); + } + + /** Returns the leader epoch, or an empty optional if no leader has been elected. */ + public OptionalInt getLeaderEpoch() { + return leaderEpoch == null ? OptionalInt.empty() : OptionalInt.of(leaderEpoch); + } + + /** Returns the replica IDs. */ + public List getReplicas() { + return replicas; + } + + /** Returns the in-sync replica IDs. */ + public List getIsr() { + return isr; + } + + @Override + public String toString() { + return "BucketInfo{" + + "tablePath=" + + tablePath + + ", tableId=" + + tableId + + ", partitionId=" + + partitionId + + ", partitionName='" + + partitionName + + '\'' + + ", bucketId=" + + bucketId + + ", leaderId=" + + leaderId + + ", leaderEpoch=" + + leaderEpoch + + ", replicas=" + + replicas + + ", isr=" + + isr + + '}'; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof BucketInfo)) { + return false; + } + BucketInfo that = (BucketInfo) o; + return tableId == that.tableId + && bucketId == that.bucketId + && Objects.equals(tablePath, that.tablePath) + && Objects.equals(partitionId, that.partitionId) + && Objects.equals(partitionName, that.partitionName) + && Objects.equals(leaderId, that.leaderId) + && Objects.equals(leaderEpoch, that.leaderEpoch) + && replicas.equals(that.replicas) + && isr.equals(that.isr); + } + + @Override + public int hashCode() { + return Objects.hash( + tablePath, + tableId, + partitionId, + partitionName, + bucketId, + leaderId, + leaderEpoch, + replicas, + isr); + } +} diff --git a/fluss-common/src/test/java/org/apache/fluss/metadata/BucketInfoTest.java b/fluss-common/src/test/java/org/apache/fluss/metadata/BucketInfoTest.java new file mode 100644 index 00000000000..a931b01af85 --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/metadata/BucketInfoTest.java @@ -0,0 +1,130 @@ +/* + * 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.metadata; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test for {@link BucketInfo}. */ +class BucketInfoTest { + + @Test + void testBucketInfoWithPartitionAndLeader() { + TablePath tablePath = TablePath.of("db", "table"); + List replicas = new ArrayList<>(Arrays.asList(1, 2, 3)); + List isr = new ArrayList<>(Arrays.asList(1, 3)); + + BucketInfo bucketInfo = new BucketInfo(tablePath, 10L, 100L, "p1", 0, 1, 7, replicas, isr); + + assertThat(bucketInfo.getTablePath()).isEqualTo(tablePath); + assertThat(bucketInfo.getTableId()).isEqualTo(10L); + assertThat(bucketInfo.getPartitionId()).hasValue(100L); + assertThat(bucketInfo.getPartitionName()).isEqualTo("p1"); + assertThat(bucketInfo.getBucketId()).isEqualTo(0); + assertThat(bucketInfo.getLeaderId()).hasValue(1); + assertThat(bucketInfo.getLeaderEpoch()).hasValue(7); + assertThat(bucketInfo.getReplicas()).containsExactly(1, 2, 3); + assertThat(bucketInfo.getIsr()).containsExactly(1, 3); + + replicas.add(4); + isr.clear(); + assertThat(bucketInfo.getReplicas()).containsExactly(1, 2, 3); + assertThat(bucketInfo.getIsr()).containsExactly(1, 3); + assertThatThrownBy(() -> bucketInfo.getReplicas().add(4)) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> bucketInfo.getIsr().clear()) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void testBucketInfoWithoutPartitionAndLeader() { + BucketInfo bucketInfo = + new BucketInfo( + TablePath.of("db", "table"), + 10L, + null, + null, + 0, + null, + null, + Collections.singletonList(1), + Collections.emptyList()); + + assertThat(bucketInfo.getPartitionId()).isEmpty(); + assertThat(bucketInfo.getPartitionName()).isNull(); + assertThat(bucketInfo.getLeaderId()).isEmpty(); + assertThat(bucketInfo.getLeaderEpoch()).isEmpty(); + assertThat(bucketInfo.getReplicas()).containsExactly(1); + assertThat(bucketInfo.getIsr()).isEmpty(); + } + + @Test + void testBucketInfoRejectsNullRequiredFields() { + TablePath tablePath = TablePath.of("db", "table"); + + assertThatThrownBy( + () -> + new BucketInfo( + null, + 10L, + null, + null, + 0, + null, + null, + Collections.singletonList(1), + Collections.emptyList())) + .isInstanceOf(NullPointerException.class) + .hasMessageContaining("tablePath should not be null"); + assertThatThrownBy( + () -> + new BucketInfo( + tablePath, + 10L, + null, + null, + 0, + null, + null, + null, + Collections.emptyList())) + .isInstanceOf(NullPointerException.class) + .hasMessageContaining("replicas should not be null"); + assertThatThrownBy( + () -> + new BucketInfo( + tablePath, + 10L, + null, + null, + 0, + null, + null, + Collections.singletonList(1), + null)) + .isInstanceOf(NullPointerException.class) + .hasMessageContaining("isr should not be null"); + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java index e8120c83d26..49f4c41db55 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java @@ -37,6 +37,7 @@ import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.config.cluster.AlterConfig; import org.apache.fluss.config.cluster.ConfigEntry; +import org.apache.fluss.metadata.BucketInfo; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; @@ -147,6 +148,17 @@ public CompletableFuture getTableInfo(TablePath tablePath) { throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); } + @Override + public CompletableFuture> describeBuckets(TablePath tablePath) { + throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); + } + + @Override + public CompletableFuture> describeBuckets( + TablePath tablePath, PartitionSpec partitionSpec) { + throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); + } + @Override public CompletableFuture dropTable(TablePath tablePath, boolean ignoreIfNotExists) { throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java index 574e1a510dd..639d6de8a03 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java @@ -20,6 +20,8 @@ import org.apache.fluss.rpc.RpcGateway; import org.apache.fluss.rpc.messages.DatabaseExistsRequest; import org.apache.fluss.rpc.messages.DatabaseExistsResponse; +import org.apache.fluss.rpc.messages.DescribeBucketsRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.GetClusterHealthRequest; @@ -104,6 +106,15 @@ public interface AdminReadOnlyGateway extends RpcGateway { @RPC(api = ApiKeys.GET_TABLE_INFO) CompletableFuture getTableInfo(GetTableInfoRequest request); + /** + * Describes bucket metadata for a table, optionally filtered by a partition spec. + * + * @param request Request containing the table path and optional partition spec + * @return The bucket metadata response + */ + @RPC(api = ApiKeys.DESCRIBE_BUCKETS) + CompletableFuture describeBuckets(DescribeBucketsRequest request); + /** * Return a {@link GetTableSchemaResponse} identified by the given {@link * GetTableSchemaRequest}. diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java index e9f18b3d67d..c44f528ea54 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java @@ -109,7 +109,8 @@ public enum ApiKeys { SCAN_KV(1061, 0, 0, PUBLIC), GET_CLUSTER_HEALTH(1062, 0, 0, PUBLIC), LIST_REMOTE_LOG_MANIFESTS(1063, 0, 0, PUBLIC), - LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC); + LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC), + DESCRIBE_BUCKETS(1065, 0, 0, PUBLIC); private static final Map ID_TO_TYPE = Arrays.stream(ApiKeys.values()) diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index fa762da93e3..43275fd7a5c 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -153,6 +153,16 @@ message GetTableInfoResponse { optional string remote_data_dir = 6; } +// describe buckets request and response +message DescribeBucketsRequest { + required PbTablePath table_path = 1; + optional PbPartitionSpec partition_spec = 2; +} + +message DescribeBucketsResponse { + repeated PbBucketInfo bucket_info = 1; +} + // list tables request and response message ListTablesRequest { required string database_name = 1; @@ -876,6 +886,19 @@ message PbBucketMetadata { optional int32 leader_epoch = 4; } +message PbBucketInfo { + required PbTablePath table_path = 1; + required int64 table_id = 2; + optional int64 partition_id = 3; + optional string partition_name = 4; + required int32 bucket_id = 5; + // optional as the leader may not be elected yet + optional int32 leader_id = 6; + repeated int32 replica_id = 7 [packed = true]; + optional int32 leader_epoch = 8; + repeated int32 isr_id = 9 [packed = true]; +} + message PbProduceLogReqForBucket { optional int64 partition_id = 1; required int32 bucket_id = 2; diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java index f465bb4a69a..3a71bd7e21d 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java @@ -21,6 +21,8 @@ import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.DatabaseExistsRequest; import org.apache.fluss.rpc.messages.DatabaseExistsResponse; +import org.apache.fluss.rpc.messages.DescribeBucketsRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; @@ -201,6 +203,12 @@ public CompletableFuture getTableInfo(GetTableInfoRequest return null; } + @Override + public CompletableFuture describeBuckets( + DescribeBucketsRequest request) { + return null; + } + @Override public CompletableFuture getTableSchema(GetTableSchemaRequest request) { return null; 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 a534a35516d..f4a7a28a4aa 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 @@ -44,6 +44,8 @@ import org.apache.fluss.rpc.messages.ApiVersionsResponse; import org.apache.fluss.rpc.messages.DatabaseExistsRequest; import org.apache.fluss.rpc.messages.DatabaseExistsResponse; +import org.apache.fluss.rpc.messages.DescribeBucketsRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest; @@ -71,6 +73,7 @@ import org.apache.fluss.rpc.messages.MetadataRequest; import org.apache.fluss.rpc.messages.MetadataResponse; import org.apache.fluss.rpc.messages.PbApiVersion; +import org.apache.fluss.rpc.messages.PbBucketInfo; import org.apache.fluss.rpc.messages.PbTablePath; import org.apache.fluss.rpc.messages.TableExistsRequest; import org.apache.fluss.rpc.messages.TableExistsResponse; @@ -85,6 +88,7 @@ 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.PartitionNegativeCache; @@ -104,6 +108,8 @@ import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -318,6 +324,131 @@ public CompletableFuture getTableInfo(GetTableInfoRequest return CompletableFuture.completedFuture(response); } + @Override + public CompletableFuture describeBuckets( + DescribeBucketsRequest request) { + TablePath tablePath = toTablePath(request.getTablePath()); + authorizeTable(OperationType.DESCRIBE, tablePath); + + TableInfo tableInfo = metadataManager.getTable(tablePath); + DescribeBucketsResponse response = new DescribeBucketsResponse(); + if (tableInfo.isPartitioned()) { + Map partitionRegistrations = + listPartitionsForDescribeBuckets(request, tablePath); + partitionRegistrations.remove(HISTORICAL_PARTITION_VALUE); + Map> partitionBucketMetadata = + getPartitionBucketMetadataForDescribeBuckets( + tablePath, + partitionRegistrations.values().stream() + .map(PartitionRegistration::getPartitionId) + .collect(Collectors.toList())); + partitionRegistrations.entrySet().stream() + .sorted(Map.Entry.comparingByKey()) + .forEach( + entry -> { + long partitionId = entry.getValue().getPartitionId(); + addBucketInfos( + response, + tablePath, + tableInfo.getTableId(), + partitionId, + entry.getKey(), + partitionBucketMetadata.getOrDefault( + partitionId, Collections.emptyList())); + }); + } else { + if (request.hasPartitionSpec()) { + throw new TableNotPartitionedException( + "Table '" + tablePath + "' is not a partitioned table."); + } + addBucketInfos( + response, + tablePath, + tableInfo.getTableId(), + null, + null, + getTableBucketMetadataForDescribeBuckets(tablePath, tableInfo.getTableId())); + } + return CompletableFuture.completedFuture(response); + } + + private Map listPartitionsForDescribeBuckets( + DescribeBucketsRequest request, TablePath tablePath) { + if (request.hasPartitionSpec()) { + return metadataManager.listPartitions( + tablePath, toResolvedPartitionSpec(request.getPartitionSpec())); + } + return metadataManager.listPartitions(tablePath); + } + + private Map> getPartitionBucketMetadataForDescribeBuckets( + TablePath tablePath, Collection partitionIds) { + try { + return zkClient.getBucketMetadataForPartitions(partitionIds); + } catch (Exception e) { + throw new FlussRuntimeException( + String.format("Failed to describe buckets for table '%s'.", tablePath), e); + } + } + + private List getTableBucketMetadataForDescribeBuckets( + TablePath tablePath, long tableId) { + try { + return zkClient.getBucketMetadataForTables(Collections.singleton(tableId)) + .getOrDefault(tableId, Collections.emptyList()); + } catch (Exception e) { + throw new FlussRuntimeException( + String.format("Failed to describe buckets for table '%s'.", tablePath), e); + } + } + + private static void addBucketInfos( + DescribeBucketsResponse response, + TablePath tablePath, + long tableId, + @Nullable Long partitionId, + @Nullable String partitionName, + List bucketMetadataList) { + bucketMetadataList.stream() + .sorted(Comparator.comparingInt(BucketMetadata::getBucketId)) + .forEach( + bucketMetadata -> + addBucketInfo( + response, + tablePath, + tableId, + partitionId, + partitionName, + bucketMetadata)); + } + + private static void addBucketInfo( + DescribeBucketsResponse response, + TablePath tablePath, + long tableId, + @Nullable Long partitionId, + @Nullable String partitionName, + BucketMetadata bucketMetadata) { + PbBucketInfo pbBucketInfo = + response.addBucketInfo() + .setTableId(tableId) + .setBucketId(bucketMetadata.getBucketId()); + pbBucketInfo + .setTablePath() + .setDatabaseName(tablePath.getDatabaseName()) + .setTableName(tablePath.getTableName()); + if (partitionId != null) { + pbBucketInfo.setPartitionId(partitionId); + } + if (partitionName != null) { + pbBucketInfo.setPartitionName(partitionName); + } + bucketMetadata.getLeaderId().ifPresent(pbBucketInfo::setLeaderId); + bucketMetadata.getLeaderEpoch().ifPresent(pbBucketInfo::setLeaderEpoch); + bucketMetadata.getReplicas().forEach(pbBucketInfo::addReplicaId); + bucketMetadata.getIsr().forEach(pbBucketInfo::addIsrId); + } + @Override public CompletableFuture getTableSchema(GetTableSchemaRequest request) { TablePath tablePath = toTablePath(request.getTablePath()); 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 43d98434252..303cd501904 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 @@ -19,6 +19,7 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.ApiException; import org.apache.fluss.exception.DatabaseAlreadyExistException; import org.apache.fluss.exception.DatabaseNotEmptyException; import org.apache.fluss.exception.DatabaseNotExistException; @@ -305,6 +306,8 @@ public Map listPartitions( return zookeeperClient.getPartitionRegistrations( tablePath, tableInfo.getPartitionKeys(), partitionFilter); } + } catch (ApiException e) { + throw e; } catch (Exception e) { throw new FlussRuntimeException( String.format( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metadata/BucketMetadata.java b/fluss-server/src/main/java/org/apache/fluss/server/metadata/BucketMetadata.java index b6b968f28e6..437ebf76b12 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/metadata/BucketMetadata.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/metadata/BucketMetadata.java @@ -19,6 +19,8 @@ import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.OptionalInt; @@ -29,16 +31,27 @@ public class BucketMetadata { private final @Nullable Integer leaderId; private final @Nullable Integer leaderEpoch; private final List replicas; + private final List isr; public BucketMetadata( int bucketId, @Nullable Integer leaderId, @Nullable Integer leaderEpoch, List replicas) { + this(bucketId, leaderId, leaderEpoch, replicas, Collections.emptyList()); + } + + public BucketMetadata( + int bucketId, + @Nullable Integer leaderId, + @Nullable Integer leaderEpoch, + List replicas, + List isr) { this.bucketId = bucketId; this.leaderId = leaderId; this.leaderEpoch = leaderEpoch; - this.replicas = replicas; + this.replicas = Collections.unmodifiableList(new ArrayList<>(replicas)); + this.isr = Collections.unmodifiableList(new ArrayList<>(isr)); } public int getBucketId() { @@ -57,6 +70,10 @@ public List getReplicas() { return replicas; } + public List getIsr() { + return isr; + } + @Override public String toString() { return "BucketMetadata{" @@ -68,6 +85,8 @@ public String toString() { + leaderEpoch + ", replicas=" + replicas + + ", isr=" + + isr + '}'; } @@ -83,11 +102,12 @@ public boolean equals(Object o) { return bucketId == that.bucketId && Objects.equals(leaderId, that.leaderId) && Objects.equals(leaderEpoch, that.leaderEpoch) - && replicas.equals(that.replicas); + && replicas.equals(that.replicas) + && isr.equals(that.isr); } @Override public int hashCode() { - return Objects.hash(bucketId, leaderId, leaderEpoch, replicas); + return Objects.hash(bucketId, leaderId, leaderEpoch, replicas, isr); } } 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 1016c9512b4..76ece8e1391 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 @@ -916,12 +916,7 @@ public Map> getPartitionsForTables(Collection /** Get the partition registrations of a table in ZK. */ public Map getPartitionRegistrations(TablePath tablePath) throws Exception { - Map partitions = new HashMap<>(); - for (String partitionName : getPartitions(tablePath)) { - Optional optPartition = getPartition(tablePath, partitionName); - optPartition.ifPresent(partition -> partitions.put(partitionName, partition)); - } - return partitions; + return getPartitionRegistrations(tablePath, getPartitions(tablePath)); } /** Get the partition and the id for the partitions of tables in ZK. */ @@ -969,20 +964,38 @@ public Map getPartitionRegistrations( List partitionKeys, ResolvedPartitionSpec partialPartitionSpec) throws Exception { - Map partitions = new HashMap<>(); - - for (String partitionName : getPartitions(tablePath)) { - ResolvedPartitionSpec resolvedPartitionSpec = - fromPartitionName(partitionKeys, partitionName); - boolean contains = resolvedPartitionSpec.contains(partialPartitionSpec); - if (contains) { - Optional optPartition = - getPartition(tablePath, partitionName); - optPartition.ifPresent(partition -> partitions.put(partitionName, partition)); - } - } - - return partitions; + List matchedPartitionNames = + getPartitions(tablePath).stream() + .filter( + partitionName -> + fromPartitionName(partitionKeys, partitionName) + .contains(partialPartitionSpec)) + .collect(Collectors.toList()); + return getPartitionRegistrations(tablePath, matchedPartitionNames); + } + + private Map getPartitionRegistrations( + TablePath tablePath, Collection partitionNames) throws Exception { + Map path2PartitionName = + partitionNames.stream() + .collect( + toMap( + partitionName -> + PartitionZNode.path(tablePath, partitionName), + partitionName -> partitionName)); + List responses = getDataInBackground(path2PartitionName.keySet()); + return processGetDataResponses( + responses, + response -> path2PartitionName.get(response.getPath()), + data -> { + PartitionRegistration partitionRegistration = PartitionZNode.decode(data); + if (partitionRegistration.getRemoteDataDir() == null) { + partitionRegistration = + partitionRegistration.newRemoteDataDir(defaultRemoteDataDir); + } + return partitionRegistration; + }, + "partition registrations"); } /** Get the id and name for the partitions of a table in ZK. */ @@ -1801,8 +1814,9 @@ private BucketMetadata createBucketMetadata( LeaderAndIsr leaderAndIsr = leaderAndIsrs.get(bucket); Integer leader = leaderAndIsr != null ? leaderAndIsr.leader() : null; Integer leaderEpoch = leaderAndIsr != null ? leaderAndIsr.leaderEpoch() : null; + List isr = leaderAndIsr != null ? leaderAndIsr.isr() : Collections.emptyList(); List replicas = assignment.getBucketAssignments().get(bucketId).getReplicas(); - return new BucketMetadata(bucketId, leader, leaderEpoch, replicas); + return new BucketMetadata(bucketId, leader, leaderEpoch, replicas, isr); } /** Close the underlying ZooKeeperClient. */ diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java index 7f3bc32e8c4..c3e758612f8 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java @@ -58,6 +58,8 @@ import org.apache.fluss.rpc.messages.DatabaseExistsResponse; import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest; import org.apache.fluss.rpc.messages.DeleteProducerOffsetsResponse; +import org.apache.fluss.rpc.messages.DescribeBucketsRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.DropAclsRequest; @@ -232,6 +234,12 @@ public CompletableFuture getTableInfo(GetTableInfoRequest throw new UnsupportedOperationException(); } + @Override + public CompletableFuture describeBuckets( + DescribeBucketsRequest request) { + throw new UnsupportedOperationException(); + } + @Override public CompletableFuture getTableSchema(GetTableSchemaRequest request) { throw new UnsupportedOperationException(); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java index c78270ea5ca..59ff00c3435 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java @@ -26,6 +26,8 @@ import org.apache.fluss.rpc.messages.ApiVersionsResponse; import org.apache.fluss.rpc.messages.DatabaseExistsRequest; import org.apache.fluss.rpc.messages.DatabaseExistsResponse; +import org.apache.fluss.rpc.messages.DescribeBucketsRequest; +import org.apache.fluss.rpc.messages.DescribeBucketsResponse; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; @@ -257,6 +259,12 @@ public CompletableFuture getTableInfo(GetTableInfoRequest throw new UnsupportedOperationException(); } + @Override + public CompletableFuture describeBuckets( + DescribeBucketsRequest request) { + throw new UnsupportedOperationException(); + } + @Override public CompletableFuture getTableSchema(GetTableSchemaRequest request) { throw new UnsupportedOperationException(); 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 cbc0b85c6a2..a6a0b85ce35 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 @@ -745,6 +745,11 @@ void testPartition() throws Exception { assertThat(partition.getPartitionId()).isEqualTo(1L); partition = zookeeperClient.getPartition(tablePath, "p2").get(); assertThat(partition.getPartitionId()).isEqualTo(2L); + Map partitionRegistrations = + zookeeperClient.getPartitionRegistrations(tablePath); + assertThat(partitionRegistrations).containsOnlyKeys("p1", "p2"); + assertThat(partitionRegistrations.get("p1").getPartitionId()).isEqualTo(1L); + assertThat(partitionRegistrations.get("p2").getPartitionId()).isEqualTo(2L); assertThat(zookeeperClient.getPartitionsForTables(Arrays.asList(tablePath))) .containsValues(new ArrayList<>(partitions)); From c64e40fc9cf4d651ad37c7a469a2e334927b818d Mon Sep 17 00:00:00 2001 From: fhan Date: Tue, 18 Aug 2026 19:32:14 +0800 Subject: [PATCH 2/2] [client] fix test failure in flink2 module --- .../org/apache/fluss/flink/catalog/FlinkCatalogTest.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java index f53fcec2118..b3be823e933 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java @@ -52,6 +52,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException; +import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException; import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; @@ -783,9 +784,9 @@ void testOperatePartitions() throws Exception { CatalogPartitionSpec invalidTestSpec = new CatalogPartitionSpec(Collections.singletonMap("second", "")); assertThatThrownBy(() -> catalog.listPartitions(path2, invalidTestSpec)) - .isInstanceOf(CatalogException.class) - .hasMessage( - "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"); // NEW: Test dropPartition functionality CatalogPartitionSpec firstPartSpec = catalogPartitionSpecs.get(0);