From 369e297ae4db965d33126630d8a0c6cada63a4a7 Mon Sep 17 00:00:00 2001 From: morazow Date: Thu, 13 Aug 2026 18:43:32 +0200 Subject: [PATCH 1/2] [server] Retain bounded rebalance history and add listRebalances API Rebalance tasks now carry started/completed timestamps (RebalanceTask JSON serde v2, backward compatible with v1 znodes). On completion or cancellation the final task is also written to a bounded ZooKeeper history at /cluster/rebalance_history/ (last 10 entries kept), and the new Admin#listRebalances() API returns a summary (id, status, timestamps) of the current rebalance plus the retained history, newest first. Per-bucket detail remains available via listRebalanceProgress(id). Part of #3965 (timestamps on listRebalanceProgress follow separately). --- .../org/apache/fluss/client/admin/Admin.java | 14 + .../apache/fluss/client/admin/FlussAdmin.java | 8 + .../client/utils/ClientRpcMessageUtils.java | 20 + .../fluss/client/admin/RebalanceITCase.java | 54 +++ .../acl/FlussAuthorizationITCase.java | 28 ++ .../utils/ClientRpcMessageUtilsTest.java | 37 ++ .../cluster/rebalance/RebalanceInfo.java | 106 ++++++ .../sink/testutils/TestAdminAdapter.java | 6 + .../fluss/rpc/gateway/AdminGateway.java | 5 + .../apache/fluss/rpc/protocol/ApiKeys.java | 3 +- fluss-rpc/src/main/proto/FlussApi.proto | 14 + .../CoordinatorEventProcessor.java | 37 +- .../coordinator/CoordinatorService.java | 14 + .../event/ListRebalancesEvent.java | 36 ++ .../rebalance/RebalanceManager.java | 158 +++++++- .../server/utils/ServerRpcMessageUtils.java | 21 ++ .../fluss/server/zk/ZooKeeperClient.java | 67 ++++ .../fluss/server/zk/data/RebalanceTask.java | 33 +- .../zk/data/RebalanceTaskJsonSerde.java | 11 +- .../apache/fluss/server/zk/data/ZkData.java | 36 ++ .../CoordinatorEventProcessorTest.java | 17 + .../coordinator/TestCoordinatorGateway.java | 7 + .../rebalance/RebalanceManagerTest.java | 356 +++++++++++++++++- .../fluss/server/zk/ZooKeeperClientTest.java | 194 +++++++++- .../zk/data/RebalanceTaskJsonSerdeTest.java | 26 +- 25 files changed, 1282 insertions(+), 26 deletions(-) create mode 100644 fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceInfo.java create mode 100644 fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ListRebalancesEvent.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..30ca8eb5f50 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 @@ -25,6 +25,7 @@ import org.apache.fluss.client.metadata.RemoteLogManifestInfo; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.rebalance.GoalType; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.config.ConfigOptions; @@ -712,6 +713,19 @@ CompletableFuture> listRebalanceProgress( */ CompletableFuture cancelRebalance(@Nullable String rebalanceId); + /** + * List a summary of all known rebalance tasks. The currently running rebalance, if any, is + * listed first. Past tasks follow, ordered from most recent to oldest. + * + *
    + *
  • {@link AuthorizationException} If the authenticated user doesn't have cluster + * permissions. + *
+ * + * @return the rebalance summaries. + */ + CompletableFuture> listRebalances(); + // ================================================================================== // Producer Offset Management APIs (for Exactly-Once Semantics) // ================================================================================== 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..6ddd4c88b06 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 @@ -28,6 +28,7 @@ import org.apache.fluss.cluster.Cluster; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.rebalance.GoalType; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.config.cluster.AlterConfig; @@ -88,6 +89,7 @@ import org.apache.fluss.rpc.messages.ListOffsetsResponse; import org.apache.fluss.rpc.messages.ListPartitionInfosRequest; import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest; +import org.apache.fluss.rpc.messages.ListRebalancesRequest; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsRequest; import org.apache.fluss.rpc.messages.ListTablesRequest; import org.apache.fluss.rpc.messages.ListTablesResponse; @@ -758,6 +760,12 @@ public CompletableFuture cancelRebalance(@Nullable String rebalanceId) { return gateway.cancelRebalance(request).thenApply(r -> null); } + @Override + public CompletableFuture> listRebalances() { + ListRebalancesRequest request = new ListRebalancesRequest(); + return gateway.listRebalances(request).thenApply(ClientRpcMessageUtils::toRebalanceInfos); + } + // ================================================================================== // Producer Offset Management APIs (for Exactly-Once Semantics) // ================================================================================== 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..672ea290129 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 @@ -31,6 +31,7 @@ import org.apache.fluss.client.metadata.RemoteLogManifestInfo; import org.apache.fluss.client.write.KvWriteBatch; import org.apache.fluss.client.write.ReadyWriteBatch; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket; @@ -68,6 +69,7 @@ import org.apache.fluss.rpc.messages.ListOffsetsRequest; import org.apache.fluss.rpc.messages.ListPartitionInfosResponse; import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse; import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.MetadataRequest; @@ -89,6 +91,7 @@ import org.apache.fluss.rpc.messages.PbProduceLogReqForBucket; import org.apache.fluss.rpc.messages.PbProducerTableOffsets; import org.apache.fluss.rpc.messages.PbPutKvReqForBucket; +import org.apache.fluss.rpc.messages.PbRebalanceInfo; import org.apache.fluss.rpc.messages.PbRebalancePlanForBucket; import org.apache.fluss.rpc.messages.PbRebalanceProgressForBucket; import org.apache.fluss.rpc.messages.PbRebalanceProgressForTable; @@ -617,6 +620,23 @@ public static Optional toRebalanceProgress( rebalanceProgress)); } + public static List toRebalanceInfos(ListRebalancesResponse response) { + List rebalanceInfos = new ArrayList<>(); + for (PbRebalanceInfo pbRebalanceInfo : response.getRebalanceInfosList()) { + rebalanceInfos.add( + new RebalanceInfo( + pbRebalanceInfo.getRebalanceId(), + RebalanceStatus.of(pbRebalanceInfo.getRebalanceStatus()), + pbRebalanceInfo.hasStartedAtMs() + ? pbRebalanceInfo.getStartedAtMs() + : -1, + pbRebalanceInfo.hasCompletedAtMs() + ? pbRebalanceInfo.getCompletedAtMs() + : -1)); + } + return rebalanceInfos; + } + private static RebalancePlanForBucket toRebalancePlanForBucket( long tableId, PbRebalancePlanForBucket rebalancePlan) { TableBucket tableBucket = diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java index 0800afdfddb..b4a6c5ca474 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java @@ -20,6 +20,7 @@ import org.apache.fluss.client.Connection; import org.apache.fluss.client.ConnectionFactory; import org.apache.fluss.cluster.rebalance.GoalType; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceStatus; import org.apache.fluss.cluster.rebalance.ServerTag; @@ -42,6 +43,7 @@ import java.time.Duration; import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.Optional; import static org.apache.fluss.record.TestData.DATA1_SCHEMA; @@ -270,6 +272,58 @@ void testListRebalanceProgress() throws Exception { "Rebalance task id unexisted-rebalance-id2 to cancel is not the current rebalance task id"); } + @Test + void testListRebalances() throws Exception { + String dbName = "db-rebalance-list-summary"; + admin.createDatabase(dbName, DatabaseDescriptor.EMPTY, false).get(); + + // add server tag PERMANENT_OFFLINE for server 3, this will avoid to generate bucket + // assignment on server 3 when create table. + admin.addServerTag(Collections.singletonList(3), ServerTag.PERMANENT_OFFLINE).get(); + + // create some none partitioned log table. + for (int i = 0; i < 6; i++) { + long tableId = + createTable( + new TablePath(dbName, "test-rebalance_table-" + i), + DATA1_TABLE_DESCRIPTOR); + FLUSS_CLUSTER_EXTENSION.waitUntilTableReady(tableId); + } + + // remove tag after crated table. + admin.removeServerTag(Collections.singletonList(3), ServerTag.PERMANENT_OFFLINE).get(); + + // no rebalance has happened yet for this admin session; the summary list may still + // contain entries left over by other test methods sharing this cluster, so just make + // sure the call succeeds without asserting emptiness here. + admin.listRebalances().get(); + + // trigger rebalance with goal set[ReplicaDistributionGoal, LeaderReplicaDistributionGoal] + String rebalanceId = + admin.rebalance( + Arrays.asList( + GoalType.REPLICA_DISTRIBUTION, + GoalType.LEADER_DISTRIBUTION)) + .get(); + retry( + Duration.ofMinutes(2), + () -> { + Optional progressOpt = + admin.listRebalanceProgress(rebalanceId).get(); + assertThat(progressOpt).isPresent(); + assertThat(progressOpt.get().status()).isEqualTo(RebalanceStatus.COMPLETED); + }); + + List rebalanceInfos = admin.listRebalances().get(); + // the just-completed rebalance is still the "current" one, so it must be first. + assertThat(rebalanceInfos).isNotEmpty(); + RebalanceInfo info = rebalanceInfos.get(0); + assertThat(info.rebalanceId()).isEqualTo(rebalanceId); + assertThat(info.status()).isEqualTo(RebalanceStatus.COMPLETED); + assertThat(info.startedAtMs()).isGreaterThanOrEqualTo(0); + assertThat(info.completedAtMs()).isGreaterThanOrEqualTo(info.startedAtMs()); + } + @Test void testSendRebalanceWhileRebalanceTaskExists() throws Exception { String dbName = "db-balance-exists"; 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..4e0a9b29584 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 @@ -1019,6 +1019,34 @@ void testListRebalanceProgress() throws Exception { guestAdmin.listRebalanceProgress(null).get(); } + @Test + void testListRebalances() throws Exception { + // test listRebalances without DESCRIBE permission on cluster resource + assertThatThrownBy(() -> guestAdmin.listRebalances().get()) + .rootCause() + .hasMessageContaining( + String.format( + "Principal %s have no authorization to operate DESCRIBE on resource Resource{type=CLUSTER, name='fluss-cluster'}", + guestPrincipal)); + + // add DESCRIBE permission to guest user on cluster resource + rootAdmin + .createAcls( + Collections.singletonList( + new AclBinding( + Resource.cluster(), + new AccessControlEntry( + guestPrincipal, + "*", + OperationType.DESCRIBE, + PermissionType.ALLOW)))) + .all() + .get(); + + // test listRebalances with DESCRIBE permission should succeed + guestAdmin.listRebalances().get(); + } + @Test void testCancelRebalance() throws Exception { // test cancelRebalance without WRITE permission on cluster resource 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..319e9313c98 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 @@ -19,11 +19,15 @@ import org.apache.fluss.client.write.KvWriteBatch; import org.apache.fluss.client.write.ReadyWriteBatch; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; +import org.apache.fluss.cluster.rebalance.RebalanceStatus; import org.apache.fluss.memory.MemorySegment; import org.apache.fluss.memory.PreAllocatedPagedOutputView; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; +import org.apache.fluss.rpc.messages.PbRebalanceInfo; import org.apache.fluss.rpc.messages.PutKvRequest; import org.apache.fluss.rpc.protocol.MergeMode; @@ -126,6 +130,39 @@ void testMakePutKvRequestWithSingleBatch() throws Exception { assertThat(request.getAggMode()).isEqualTo(MergeMode.OVERWRITE.getProtoValue()); } + @Test + void testToRebalanceInfosWithUnsetTimestampsMapToMinusOne() { + ListRebalancesResponse response = new ListRebalancesResponse(); + PbRebalanceInfo pbRebalanceInfo = response.addRebalanceInfo(); + pbRebalanceInfo + .setRebalanceId("rebalance-1") + .setRebalanceStatus(RebalanceStatus.COMPLETED.getCode()); + + List rebalanceInfos = ClientRpcMessageUtils.toRebalanceInfos(response); + + assertThat(rebalanceInfos) + .containsExactly( + new RebalanceInfo("rebalance-1", RebalanceStatus.COMPLETED, -1, -1)); + } + + @Test + void testToRebalanceInfosWithSetTimestampsPassThrough() { + ListRebalancesResponse response = new ListRebalancesResponse(); + PbRebalanceInfo pbRebalanceInfo = response.addRebalanceInfo(); + pbRebalanceInfo + .setRebalanceId("rebalance-1") + .setRebalanceStatus(RebalanceStatus.COMPLETED.getCode()) + .setStartedAtMs(1_000L) + .setCompletedAtMs(2_000L); + + List rebalanceInfos = ClientRpcMessageUtils.toRebalanceInfos(response); + + assertThat(rebalanceInfos) + .containsExactly( + new RebalanceInfo( + "rebalance-1", RebalanceStatus.COMPLETED, 1_000L, 2_000L)); + } + 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/cluster/rebalance/RebalanceInfo.java b/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceInfo.java new file mode 100644 index 00000000000..1d068357598 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceInfo.java @@ -0,0 +1,106 @@ +/* + * 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.cluster.rebalance; + +import org.apache.fluss.annotation.PublicEvolving; + +import java.util.Objects; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * A summary of a rebalance task. + * + *

Unlike {@link RebalanceProgress}, this does not carry per-bucket detail. + * + * @since 1.0 + */ +@PublicEvolving +public class RebalanceInfo { + + /** The rebalance id. */ + private final String rebalanceId; + + /** The final or current rebalance status. */ + private final RebalanceStatus status; + + /** The time when this rebalance task was started, or {@code -1} if unset. */ + private final long startedAtMs; + + /** The time when this rebalance task reached a final status, or {@code -1} if unset. */ + private final long completedAtMs; + + public RebalanceInfo( + String rebalanceId, RebalanceStatus status, long startedAtMs, long completedAtMs) { + this.rebalanceId = checkNotNull(rebalanceId); + this.status = checkNotNull(status); + this.startedAtMs = startedAtMs; + this.completedAtMs = completedAtMs; + } + + public String rebalanceId() { + return rebalanceId; + } + + public RebalanceStatus status() { + return status; + } + + public long startedAtMs() { + return startedAtMs; + } + + public long completedAtMs() { + return completedAtMs; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + RebalanceInfo that = (RebalanceInfo) o; + return startedAtMs == that.startedAtMs + && completedAtMs == that.completedAtMs + && Objects.equals(rebalanceId, that.rebalanceId) + && status == that.status; + } + + @Override + public int hashCode() { + return Objects.hash(rebalanceId, status, startedAtMs, completedAtMs); + } + + @Override + public String toString() { + return "RebalanceInfo{" + + "rebalanceId='" + + rebalanceId + + '\'' + + ", status=" + + status + + ", startedAtMs=" + + startedAtMs + + ", completedAtMs=" + + completedAtMs + + '}'; + } +} 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..fce091d0c61 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 @@ -33,6 +33,7 @@ import org.apache.fluss.client.metadata.RemoteLogManifestInfo; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.rebalance.GoalType; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.config.cluster.AlterConfig; @@ -290,6 +291,11 @@ public CompletableFuture cancelRebalance(@Nullable String rebalanceId) { throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); } + @Override + public CompletableFuture> listRebalances() { + throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); + } + @Override public CompletableFuture registerProducerOffsets( String producerId, Map offsets) { diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java index 44c362a54d5..c00a92127ab 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java @@ -55,6 +55,8 @@ import org.apache.fluss.rpc.messages.ListKvSnapshotsResponse; import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest; import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse; +import org.apache.fluss.rpc.messages.ListRebalancesRequest; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsRequest; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse; import org.apache.fluss.rpc.messages.RebalanceRequest; @@ -172,6 +174,9 @@ CompletableFuture listRebalanceProgress( @RPC(api = ApiKeys.CANCEL_REBALANCE) CompletableFuture cancelRebalance(CancelRebalanceRequest request); + @RPC(api = ApiKeys.LIST_REBALANCES) + CompletableFuture listRebalances(ListRebalancesRequest request); + // ================================================================================== // Producer Offset Management APIs (for Exactly-Once Semantics) // ================================================================================== 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..f50192deac5 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), + LIST_REBALANCES(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 4a6971583cf..2986205ecab 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -765,6 +765,13 @@ message CancelRebalanceRequest { message CancelRebalanceResponse { } +message ListRebalancesRequest { +} + +message ListRebalancesResponse { + repeated PbRebalanceInfo rebalance_infos = 1; +} + // ------------------------------------------------------------------------------------------ // Producer Offset Management // ------------------------------------------------------------------------------------------ @@ -1245,6 +1252,13 @@ message PbRebalanceProgressForBucket { required int32 rebalance_status = 2; } +message PbRebalanceInfo { + required string rebalance_id = 1; + required int32 rebalance_status = 2; + optional int64 started_at_ms = 3; + optional int64 completed_at_ms = 4; +} + message PbRebalancePlanForBucket { optional int64 partition_id = 1; required int32 bucket_id = 2; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java index 96220fee120..80d737bf83c 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java @@ -21,6 +21,7 @@ import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceStatus; @@ -55,6 +56,7 @@ import org.apache.fluss.rpc.messages.CommitRemoteLogManifestResponse; import org.apache.fluss.rpc.messages.ControlledShutdownResponse; import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.PbCommitLakeTableSnapshotRespForTable; import org.apache.fluss.rpc.messages.RebalanceResponse; import org.apache.fluss.rpc.messages.RemoveServerTagResponse; @@ -79,6 +81,7 @@ import org.apache.fluss.server.coordinator.event.EventProcessor; import org.apache.fluss.server.coordinator.event.FencedCoordinatorEvent; import org.apache.fluss.server.coordinator.event.ListRebalanceProgressEvent; +import org.apache.fluss.server.coordinator.event.ListRebalancesEvent; import org.apache.fluss.server.coordinator.event.NewCoordinatorEvent; import org.apache.fluss.server.coordinator.event.NewTabletServerEvent; import org.apache.fluss.server.coordinator.event.NotifyKvSnapshotOffsetEvent; @@ -162,6 +165,7 @@ import static org.apache.fluss.server.coordinator.statemachine.ReplicaState.ReplicaMigrationStarted; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeAdjustIsrResponse; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeListRebalanceProgressResponse; +import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeListRebalancesResponse; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeRebalanceResponse; import static org.apache.fluss.utils.concurrent.FutureUtils.completeFromCallable; @@ -774,6 +778,8 @@ public void process(CoordinatorEvent event) { completeFromCallable( listRebalanceProgressEvent.getRespCallback(), () -> processListRebalanceProgress(listRebalanceProgressEvent)); + } else if (event instanceof ListRebalancesEvent) { + processListRebalances((ListRebalancesEvent) event); } else if (event instanceof AccessContextEvent) { AccessContextEvent accessContextEvent = (AccessContextEvent) event; processAccessContext(accessContextEvent); @@ -1538,7 +1544,11 @@ private RebalanceResponse processRebalance(RebalanceEvent rebalanceEvent) { Map executePlan = rebalanceTask.getExecutePlan(); zooKeeperClient.registerRebalanceTask(rebalanceTask); rebalanceManager.registerRebalance( - rebalanceTask.getRebalanceId(), executePlan, RebalanceStatus.NOT_STARTED); + rebalanceTask.getRebalanceId(), + executePlan, + RebalanceStatus.NOT_STARTED, + rebalanceTask.getStartedAtMs(), + rebalanceTask.getCompletedAtMs()); } catch (Exception e) { throw new RebalanceFailureException( String.format( @@ -1568,6 +1578,31 @@ private ListRebalanceProgressResponse processListRebalanceProgress( return makeListRebalanceProgressResponse(rebalanceProgress); } + private void processListRebalances(ListRebalancesEvent event) { + CompletableFuture callback = event.getRespCallback(); + try { + // Snapshot the current rebalance on the event thread (where it mutates), then read the + // ZK-backed history on the ioExecutor: this involves IO operation (ZK), so we must not + // block the event loop on it. + RebalanceInfo currentRebalance = rebalanceManager.currentRebalanceInfo(); + ioExecutor.execute( + () -> { + try { + callback.complete( + makeListRebalancesResponse( + rebalanceManager.listRebalances(currentRebalance))); + } catch (Throwable t) { + callback.completeExceptionally(t); + } + }); + } catch (Throwable t) { + // On shutdown the snapshot throws (manager closed) and execute() throws + // RejectedExecutionException. The event manager only logs escaping throwables, so + // completing here is what stops the client blocking until the RPC times out. + callback.completeExceptionally(t); + } + } + /** * This method can be trigger by: * 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 04119bb0fd9..6a7bdbc6d01 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 @@ -112,6 +112,8 @@ import org.apache.fluss.rpc.messages.ListKvSnapshotsResponse; import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest; import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse; +import org.apache.fluss.rpc.messages.ListRebalancesRequest; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsRequest; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse; import org.apache.fluss.rpc.messages.MetadataRequest; @@ -158,6 +160,7 @@ import org.apache.fluss.server.coordinator.event.ControlledShutdownEvent; import org.apache.fluss.server.coordinator.event.EventManager; import org.apache.fluss.server.coordinator.event.ListRebalanceProgressEvent; +import org.apache.fluss.server.coordinator.event.ListRebalancesEvent; import org.apache.fluss.server.coordinator.event.RebalanceEvent; import org.apache.fluss.server.coordinator.event.RemoveServerTagEvent; import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseHandler; @@ -1577,6 +1580,17 @@ public CompletableFuture cancelRebalance( return response; } + @Override + public CompletableFuture listRebalances(ListRebalancesRequest request) { + if (authorizer != null) { + authorizer.authorize(currentSession(), OperationType.DESCRIBE, Resource.cluster()); + } + + CompletableFuture response = new CompletableFuture<>(); + eventManagerSupplier.get().put(new ListRebalancesEvent(response)); + return response; + } + @Override public CompletableFuture getClusterHealth( GetClusterHealthRequest request) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ListRebalancesEvent.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ListRebalancesEvent.java new file mode 100644 index 00000000000..e806ff66ea2 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/ListRebalancesEvent.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.server.coordinator.event; + +import org.apache.fluss.rpc.messages.ListRebalancesResponse; + +import java.util.concurrent.CompletableFuture; + +/** The event for listing all known rebalances. */ +public class ListRebalancesEvent implements CoordinatorEvent { + + private final CompletableFuture respCallback; + + public ListRebalancesEvent(CompletableFuture respCallback) { + this.respCallback = respCallback; + } + + public CompletableFuture getRespCallback() { + return respCallback; + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java index cc29b982b07..4c25d5bcd25 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java @@ -18,11 +18,13 @@ package org.apache.fluss.server.coordinator.rebalance; import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket; import org.apache.fluss.cluster.rebalance.RebalanceStatus; import org.apache.fluss.cluster.rebalance.ServerTag; +import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.NoRebalanceInProgressException; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.coordinator.CoordinatorContext; @@ -48,6 +50,7 @@ import javax.annotation.Nullable; import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -84,6 +87,9 @@ public class RebalanceManager { /** Hardcoded interval for the periodic timeout check: 30 seconds. */ private static final long TIMEOUT_CHECK_INTERVAL_MS = 30 * 1000L; + /** Hardcoded bound on the number of completed rebalance tasks retained in ZK history. */ + private static final int HISTORY_RETENTION_COUNT = 10; + private final ZooKeeperClient zkClient; private final CoordinatorEventProcessor eventProcessor; private final EventManager eventManager; @@ -105,6 +111,12 @@ public class RebalanceManager { private volatile long registerTime; private volatile @Nullable RebalanceStatus rebalanceStatus; private volatile @Nullable String currentRebalanceId; + + /** The started/completed timestamps of {@link #currentRebalanceId}, or -1 if unset. */ + private volatile long currentStartedAtMs = -1; + + private volatile long currentCompletedAtMs = -1; + private volatile boolean isClosed = false; /** @@ -182,7 +194,9 @@ private void initialize() { registerRebalance( rebalancePlan.getRebalanceId(), rebalancePlan.getExecutePlan(), - rebalancePlan.getRebalanceStatus())); + rebalancePlan.getRebalanceStatus(), + rebalancePlan.getStartedAtMs(), + rebalancePlan.getCompletedAtMs())); } catch (Exception e) { LOG.error( "Failed to get rebalance plan from zookeeper, it will be treated as no" @@ -195,6 +209,16 @@ public void registerRebalance( String rebalanceId, Map rebalancePlan, RebalanceStatus newStatus) { + registerRebalance(rebalanceId, rebalancePlan, newStatus, clock.milliseconds(), -1); + } + + /** Registers a rebalance task, retaining its started/completed timestamps. */ + public void registerRebalance( + String rebalanceId, + Map rebalancePlan, + RebalanceStatus newStatus, + long startedAtMs, + long completedAtMs) { checkNotClosed(); registerTime = System.currentTimeMillis(); // first clear all exists tasks. @@ -206,7 +230,17 @@ public void registerRebalance( inflightTaskStartMs = -1; currentRebalanceId = rebalanceId; + currentStartedAtMs = startedAtMs; + currentCompletedAtMs = completedAtMs; if (rebalancePlan.isEmpty()) { + if (FINAL_STATUSES.contains(newStatus)) { + // Restoring an already-final empty-plan task on failover (see initialize()): the + // timestamps above already came from the restored task, so just adopt its status + // rather than re-running completion (which would re-stamp completedAtMs and + // rewrite history). + rebalanceStatus = newStatus; + return; + } completeRebalance(); return; } @@ -287,6 +321,65 @@ public void finishRebalanceTask(TableBucket tableBucket, RebalanceStatus statusF currentRebalanceId, rebalanceStatus, 0.0, progressForBucketMap); } + /** + * Returns a summary of the current rebalance, or {@code null} if there is none. + * + *

Must be called from the coordinator event thread so the id/status/timestamp fields form a + * consistent snapshot. + */ + public @Nullable RebalanceInfo currentRebalanceInfo() { + checkNotClosed(); + if (currentRebalanceId == null) { + return null; + } + return new RebalanceInfo( + currentRebalanceId, rebalanceStatus, currentStartedAtMs, currentCompletedAtMs); + } + + /** Event-thread convenience overload of {@link #listRebalances(RebalanceInfo)}. */ + @VisibleForTesting + List listRebalances() { + return listRebalances(currentRebalanceInfo()); + } + + /** + * Returns the given current-rebalance summary (if any) followed by the ZooKeeper history, + * newest first. + * + *

A finished rebalance stays current while also appearing in history, so the matching + * history entry is skipped to avoid a duplicate. + * + *

Reads only ZooKeeper and the {@code currentRebalance}, so it may run off the coordinator + * event thread. + */ + public List listRebalances(@Nullable RebalanceInfo currentRebalance) { + checkNotClosed(); + List rebalanceInfos = new ArrayList<>(); + String currentId = currentRebalance == null ? null : currentRebalance.rebalanceId(); + if (currentRebalance != null) { + rebalanceInfos.add(currentRebalance); + } + + List history; + try { + history = zkClient.getRebalanceHistory(); + } catch (Exception e) { + throw new FlussRuntimeException("Failed to get rebalance history from zookeeper.", e); + } + for (RebalanceTask historyTask : history) { + if (historyTask.getRebalanceId().equals(currentId)) { + continue; + } + rebalanceInfos.add( + new RebalanceInfo( + historyTask.getRebalanceId(), + historyTask.getRebalanceStatus(), + historyTask.getStartedAtMs(), + historyTask.getCompletedAtMs())); + } + return rebalanceInfos; + } + public void cancelRebalance(@Nullable String rebalanceId) { checkNotClosed(); @@ -308,20 +401,34 @@ public void cancelRebalance(@Nullable String rebalanceId) { return; } + long completedAtMs = clock.milliseconds(); try { Optional rebalanceTaskOpt = zkClient.getRebalanceTask(); if (rebalanceTaskOpt.isPresent()) { RebalanceTask rebalanceTask = rebalanceTaskOpt.get(); - zkClient.registerRebalanceTask( + long startedAtMs = resolveStartedAtMs(rebalanceTask); + RebalanceTask finalTask = new RebalanceTask( rebalanceTask.getRebalanceId(), CANCELED, - rebalanceTask.getExecutePlan())); + rebalanceTask.getExecutePlan(), + startedAtMs, + completedAtMs); + zkClient.registerRebalanceTask(finalTask); + try { + zkClient.registerRebalanceHistory(finalTask, HISTORY_RETENTION_COUNT); + } catch (Exception e) { + LOG.error( + "Error when writing rebalance task {} to history.", + finalTask.getRebalanceId(), + e); + } } } catch (Exception e) { LOG.error("Error when delete rebalance plan from zookeeper.", e); } + currentCompletedAtMs = completedAtMs; rebalanceStatus = CANCELED; inProgressRebalanceTasksQueue.clear(); inProgressRebalanceTasks.clear(); @@ -392,23 +499,40 @@ private void processNewRebalanceTask() { private void completeRebalance() { checkNotClosed(); + long completedAtMs = clock.milliseconds(); try { Optional rebalanceTaskOpt = zkClient.getRebalanceTask(); Map bucketPlan; + // When the read below finds no task at all, keep the in-memory startedAtMs. + long startedAtMs = currentStartedAtMs; if (rebalanceTaskOpt.isPresent()) { - bucketPlan = rebalanceTaskOpt.get().getExecutePlan(); + RebalanceTask rebalanceTask = rebalanceTaskOpt.get(); + bucketPlan = rebalanceTask.getExecutePlan(); + startedAtMs = resolveStartedAtMs(rebalanceTask); } else { LOG.warn( "Rebalance task is empty in zk when complete rebalance. " + "It will be treated as no rebalance tasks."); bucketPlan = new HashMap<>(); } - zkClient.registerRebalanceTask( - new RebalanceTask(currentRebalanceId, COMPLETED, bucketPlan)); + RebalanceTask finalTask = + new RebalanceTask( + currentRebalanceId, COMPLETED, bucketPlan, startedAtMs, completedAtMs); + zkClient.registerRebalanceTask(finalTask); + // Only record history once the current-task znode carries the final status; a + // COMPLETED history entry next to a non-final current task would re-execute the + // finished rebalance on failover. + try { + zkClient.registerRebalanceHistory(finalTask, HISTORY_RETENTION_COUNT); + } catch (Exception e) { + LOG.error( + "Error when writing rebalance task {} to history.", currentRebalanceId, e); + } } catch (Exception e) { LOG.error("Error when update rebalance plan from zookeeper.", e); } + currentCompletedAtMs = completedAtMs; rebalanceStatus = COMPLETED; inProgressRebalanceTasks.clear(); inProgressRebalanceTasksQueue.clear(); @@ -419,6 +543,16 @@ private void completeRebalance() { LOG.info("Rebalance complete with {} ms.", System.currentTimeMillis() - registerTime); } + /** + * Prefers the just-read task's startedAtMs; falls back to the in-memory value when the znode + * predates the timestamp fields (version 1), where both are -1. + */ + private long resolveStartedAtMs(RebalanceTask rebalanceTask) { + return rebalanceTask.getStartedAtMs() >= 0 + ? rebalanceTask.getStartedAtMs() + : currentStartedAtMs; + } + private ClusterModel buildClusterModel(CoordinatorContext coordinatorContext) { Map liveTabletServers = coordinatorContext.getLiveTabletServers(); Map serverTags = coordinatorContext.getServerTags(); @@ -469,7 +603,7 @@ private RebalanceTask buildRebalanceTask( for (RebalancePlanForBucket rebalancePlanForBucket : rebalancePlanForBuckets) { bucketPlan.put(rebalancePlanForBucket.getTableBucket(), rebalancePlanForBucket); } - return new RebalanceTask(rebalanceId, NOT_STARTED, bucketPlan); + return new RebalanceTask(rebalanceId, NOT_STARTED, bucketPlan, clock.milliseconds(), -1); } private boolean isOfflineTagged(ServerTag serverTag) { @@ -533,4 +667,14 @@ public ClusterModel buildClusterModel() { RebalanceStatus getRebalanceStatus() { return rebalanceStatus; } + + @VisibleForTesting + long getCurrentStartedAtMs() { + return currentStartedAtMs; + } + + @VisibleForTesting + long getCurrentCompletedAtMs() { + return currentCompletedAtMs; + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index 89ae6f7138e..c6f32decb76 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -20,6 +20,7 @@ import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket; @@ -91,6 +92,7 @@ import org.apache.fluss.rpc.messages.ListOffsetsResponse; import org.apache.fluss.rpc.messages.ListPartitionInfosResponse; import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse; import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.LookupResponse; @@ -144,6 +146,7 @@ import org.apache.fluss.rpc.messages.PbProducerTableOffsets; import org.apache.fluss.rpc.messages.PbPutKvReqForBucket; import org.apache.fluss.rpc.messages.PbPutKvRespForBucket; +import org.apache.fluss.rpc.messages.PbRebalanceInfo; import org.apache.fluss.rpc.messages.PbRebalancePlanForBucket; import org.apache.fluss.rpc.messages.PbRebalanceProgressForBucket; import org.apache.fluss.rpc.messages.PbRemoteLogManifestEntry; @@ -2187,6 +2190,24 @@ public static ListRebalanceProgressResponse makeListRebalanceProgressResponse( return response; } + public static ListRebalancesResponse makeListRebalancesResponse( + List rebalanceInfos) { + ListRebalancesResponse response = new ListRebalancesResponse(); + for (RebalanceInfo rebalanceInfo : rebalanceInfos) { + PbRebalanceInfo pbRebalanceInfo = + response.addRebalanceInfo() + .setRebalanceId(rebalanceInfo.rebalanceId()) + .setRebalanceStatus(rebalanceInfo.status().getCode()); + if (rebalanceInfo.startedAtMs() >= 0) { + pbRebalanceInfo.setStartedAtMs(rebalanceInfo.startedAtMs()); + } + if (rebalanceInfo.completedAtMs() >= 0) { + pbRebalanceInfo.setCompletedAtMs(rebalanceInfo.completedAtMs()); + } + } + return response; + } + private static PbRebalancePlanForBucket toPbRebalancePlanForBucket( RebalancePlanForBucket planForBucket) { PbRebalancePlanForBucket pbRebalancePlanForBucket = 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..f7a4e3657b7 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 @@ -77,6 +77,8 @@ import org.apache.fluss.server.zk.data.ZkData.PartitionsZNode; import org.apache.fluss.server.zk.data.ZkData.ProducerIdZNode; import org.apache.fluss.server.zk.data.ZkData.ProducersZNode; +import org.apache.fluss.server.zk.data.ZkData.RebalanceHistoryTaskZNode; +import org.apache.fluss.server.zk.data.ZkData.RebalanceHistoryZNode; import org.apache.fluss.server.zk.data.ZkData.RebalanceZNode; import org.apache.fluss.server.zk.data.ZkData.ResourceAclNode; import org.apache.fluss.server.zk.data.ZkData.SchemaZNode; @@ -114,6 +116,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -1677,6 +1680,70 @@ public void deleteRebalanceTask() throws Exception { deletePath(RebalanceZNode.path()); } + /** + * Writes a finished rebalance task to the bounded ZooKeeper-backed history, then trims the + * oldest entries beyond {@code retentionCount} (ordered by {@link + * RebalanceTask#getCompletedAtMs()}). + */ + public void registerRebalanceHistory(RebalanceTask rebalanceTask, int retentionCount) + throws Exception { + String path = RebalanceHistoryTaskZNode.path(rebalanceTask.getRebalanceId()); + try { + zkClient.create() + .creatingParentsIfNeeded() + .withMode(CreateMode.PERSISTENT) + .forPath(path, RebalanceHistoryTaskZNode.encode(rebalanceTask)); + } catch (KeeperException.NodeExistsException e) { + zkClient.setData().forPath(path, RebalanceHistoryTaskZNode.encode(rebalanceTask)); + } + + List history = getRebalanceHistory(); + for (RebalanceTask expired : + history.subList(Math.min(retentionCount, history.size()), history.size())) { + deletePath(RebalanceHistoryTaskZNode.path(expired.getRebalanceId())); + } + } + + /** + * Returns the bounded rebalance history, newest first (by {@code completed_at_ms}, ties broken + * by rebalance id). + * + *

A child znode that fails to decode, or decodes with a null status, is logged and skipped + * rather than failing the whole listing; such an entry is therefore also excluded from the + * retention trim performed by {@link #registerRebalanceHistory}. + */ + public List getRebalanceHistory() throws Exception { + List history = new ArrayList<>(); + for (String rebalanceId : getChildren(RebalanceHistoryZNode.path())) { + Optional data = getOrEmpty(RebalanceHistoryTaskZNode.path(rebalanceId)); + if (!data.isPresent()) { + continue; + } + RebalanceTask task; + try { + task = RebalanceHistoryTaskZNode.decode(data.get()); + } catch (Exception e) { + LOG.warn( + "Failed to decode rebalance history entry {}, skipping it.", + rebalanceId, + e); + continue; + } + if (task.getRebalanceStatus() == null) { + LOG.warn( + "Rebalance history entry {} has an unknown status, skipping it.", + rebalanceId); + continue; + } + history.add(task); + } + history.sort( + Comparator.comparingLong(RebalanceTask::getCompletedAtMs) + .reversed() + .thenComparing(RebalanceTask::getRebalanceId)); + return history; + } + // -------------------------------------------------------------------------------------------- // Utils // -------------------------------------------------------------------------------------------- diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTask.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTask.java index 2e342123a1a..c3cf6300beb 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTask.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTask.java @@ -52,14 +52,24 @@ public class RebalanceTask { private final Map> planForBucketsOfPartitionedTable; + /** The time when this rebalance task was started, or {@code -1} if unset. */ + private final long startedAtMs; + + /** The time when this rebalance task reached a final status, or {@code -1} if unset. */ + private final long completedAtMs; + public RebalanceTask( String rebalanceId, RebalanceStatus rebalanceStatus, - Map bucketPlan) { + Map bucketPlan, + long startedAtMs, + long completedAtMs) { this.rebalanceId = rebalanceId; this.rebalanceStatus = rebalanceStatus; this.planForBuckets = new HashMap<>(); this.planForBucketsOfPartitionedTable = new HashMap<>(); + this.startedAtMs = startedAtMs; + this.completedAtMs = completedAtMs; for (Map.Entry entry : bucketPlan.entrySet()) { TableBucket tableBucket = entry.getKey(); @@ -94,6 +104,14 @@ public Map> getPlanForBucketsOfPart return planForBucketsOfPartitionedTable; } + public long getStartedAtMs() { + return startedAtMs; + } + + public long getCompletedAtMs() { + return completedAtMs; + } + public Map getExecutePlan() { Map executePlan = new HashMap<>(); planForBuckets.forEach( @@ -125,6 +143,10 @@ public String toString() { + planForBuckets + ", planForBucketsOfPartitionedTable=" + planForBucketsOfPartitionedTable + + ", startedAtMs=" + + startedAtMs + + ", completedAtMs=" + + completedAtMs + '}'; } @@ -139,6 +161,8 @@ public boolean equals(Object o) { RebalanceTask that = (RebalanceTask) o; return rebalanceStatus == that.rebalanceStatus + && startedAtMs == that.startedAtMs + && completedAtMs == that.completedAtMs && Objects.equals(rebalanceId, that.rebalanceId) && Objects.equals(planForBuckets, that.planForBuckets) && Objects.equals( @@ -148,6 +172,11 @@ public boolean equals(Object o) { @Override public int hashCode() { return Objects.hash( - rebalanceId, rebalanceStatus, planForBuckets, planForBucketsOfPartitionedTable); + rebalanceId, + rebalanceStatus, + planForBuckets, + planForBucketsOfPartitionedTable, + startedAtMs, + completedAtMs); } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerde.java index dfb920125a1..d723e549370 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerde.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerde.java @@ -43,6 +43,8 @@ public class RebalanceTaskJsonSerde private static final String REBALANCE_ID = "rebalance_id"; private static final String REBALANCE_STATUS = "rebalance_status"; private static final String REBALANCE_PLAN = "rebalance_plan"; + private static final String STARTED_AT_MS = "started_at_ms"; + private static final String COMPLETED_AT_MS = "completed_at_ms"; private static final String TABLE_ID = "table_id"; private static final String PARTITION_ID = "partition_id"; @@ -54,7 +56,7 @@ public class RebalanceTaskJsonSerde private static final String ORIGIN_REPLICAS = "origin_replicas"; private static final String NEW_REPLICAS = "new_replicas"; - private static final int VERSION = 1; + private static final int VERSION = 2; @Override public void serialize(RebalanceTask rebalanceTask, JsonGenerator generator) throws IOException { @@ -62,6 +64,8 @@ public void serialize(RebalanceTask rebalanceTask, JsonGenerator generator) thro generator.writeNumberField(VERSION_KEY, VERSION); generator.writeStringField(REBALANCE_ID, rebalanceTask.getRebalanceId()); generator.writeNumberField(REBALANCE_STATUS, rebalanceTask.getRebalanceStatus().getCode()); + generator.writeNumberField(STARTED_AT_MS, rebalanceTask.getStartedAtMs()); + generator.writeNumberField(COMPLETED_AT_MS, rebalanceTask.getCompletedAtMs()); generator.writeArrayFieldStart(REBALANCE_PLAN); // first to write none-partitioned tables. @@ -102,6 +106,8 @@ public RebalanceTask deserialize(JsonNode node) { String rebalanceId = node.get(REBALANCE_ID).asText(); RebalanceStatus rebalanceStatus = RebalanceStatus.of(node.get(REBALANCE_STATUS).asInt()); + long startedAtMs = node.has(STARTED_AT_MS) ? node.get(STARTED_AT_MS).asLong() : -1L; + long completedAtMs = node.has(COMPLETED_AT_MS) ? node.get(COMPLETED_AT_MS).asLong() : -1L; Map planForBuckets = new HashMap<>(); for (JsonNode tablePartitionPlanNode : rebalancePlanNode) { @@ -140,7 +146,8 @@ public RebalanceTask deserialize(JsonNode node) { } } - return new RebalanceTask(rebalanceId, rebalanceStatus, planForBuckets); + return new RebalanceTask( + rebalanceId, rebalanceStatus, planForBuckets, startedAtMs, completedAtMs); } private void serializeRebalancePlanForBucket( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java index 941512e0f2d..d26d7aeadcc 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java @@ -903,6 +903,42 @@ public static RebalanceTask decode(byte[] json) { } } + /** + * The znode for the bounded history of completed rebalance tasks. + * + *

This is the root node for the individual {@link RebalanceHistoryTaskZNode}s. Kept as a + * sibling of, rather than a child of, {@link RebalanceZNode} because that znode already stores + * the current task as its own data -- nesting history under it would prevent deleting the + * current task once any history exists (a znode with children cannot be deleted). The znode + * path is: + * + *

/cluster/rebalance_history + */ + public static final class RebalanceHistoryZNode { + public static String path() { + return "/cluster/rebalance_history"; + } + } + + /** + * The znode for a single historical rebalance task, keyed by rebalance id. The znode path is: + * + *

/cluster/rebalance_history/[rebalanceId] + */ + public static final class RebalanceHistoryTaskZNode { + public static String path(String rebalanceId) { + return RebalanceHistoryZNode.path() + "/" + rebalanceId; + } + + public static byte[] encode(RebalanceTask rebalanceTask) { + return JsonSerdeUtils.writeValueAsBytes(rebalanceTask, RebalanceTaskJsonSerde.INSTANCE); + } + + public static RebalanceTask decode(byte[] json) { + return JsonSerdeUtils.readValue(json, RebalanceTaskJsonSerde.INSTANCE); + } + } + // ------------------------------------------------------------------------------------------ // ZNodes under "/producers/" // ------------------------------------------------------------------------------------------ diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java index 0bbb40b5c8d..e6f50b43189 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java @@ -41,6 +41,7 @@ import org.apache.fluss.rpc.messages.ApiMessage; import org.apache.fluss.rpc.messages.CommitKvSnapshotResponse; import org.apache.fluss.rpc.messages.CommitRemoteLogManifestResponse; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.NotifyKvSnapshotOffsetRequest; import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest; import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse; @@ -54,6 +55,7 @@ import org.apache.fluss.server.coordinator.event.CommitKvSnapshotEvent; import org.apache.fluss.server.coordinator.event.CommitRemoteLogManifestEvent; import org.apache.fluss.server.coordinator.event.CoordinatorEventManager; +import org.apache.fluss.server.coordinator.event.ListRebalancesEvent; import org.apache.fluss.server.coordinator.event.NotifyLeaderAndIsrResponseReceivedEvent; import org.apache.fluss.server.coordinator.event.RetryOfflineLeaderEvent; import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager; @@ -115,6 +117,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -2155,6 +2158,20 @@ void testLeaderOnlyRebalanceIgnoresSuccessResponseFromOldLeader() throws Excepti verifyIsr(tb0, 1, Arrays.asList(0, 1, 2)); } + @Test + void testListRebalancesCompletesCallbackWhenRebalanceManagerClosed() { + // Once the manager is closed, the snapshot taken on the event thread throws; the handler + // must still complete the callback rather than leave the client to hit the RPC timeout. + eventProcessor.getRebalanceManager().close(); + + CompletableFuture callback = new CompletableFuture<>(); + eventProcessor.getCoordinatorEventManager().put(new ListRebalancesEvent(callback)); + + assertThatThrownBy(() -> callback.get(10, TimeUnit.SECONDS)) + .isInstanceOf(ExecutionException.class) + .hasMessageContaining("RebalanceManager is already closed"); + } + private void verifyIsr(TableBucket tb, int expectedLeader, List expectedIsr) throws Exception { LeaderAndIsr leaderAndIsr = 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..46f21d19ac4 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 @@ -100,6 +100,8 @@ import org.apache.fluss.rpc.messages.ListPartitionInfosResponse; import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest; import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse; +import org.apache.fluss.rpc.messages.ListRebalancesRequest; +import org.apache.fluss.rpc.messages.ListRebalancesResponse; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsRequest; import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse; import org.apache.fluss.rpc.messages.ListTablesRequest; @@ -422,6 +424,11 @@ public CompletableFuture listRebalanceProgress( throw new UnsupportedOperationException(); } + @Override + public CompletableFuture listRebalances(ListRebalancesRequest request) { + throw new UnsupportedOperationException(); + } + @Override public CompletableFuture cancelRebalance( CancelRebalanceRequest request) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java index 57731571a1e..15c9df6d6b1 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java @@ -17,11 +17,13 @@ package org.apache.fluss.server.coordinator.rebalance; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket; import org.apache.fluss.cluster.rebalance.RebalanceStatus; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.coordinator.AutoPartitionManager; import org.apache.fluss.server.coordinator.CoordinatorContext; @@ -43,6 +45,8 @@ import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.ZooKeeperExtension; import org.apache.fluss.server.zk.data.RebalanceTask; +import org.apache.fluss.server.zk.data.ZkData.RebalanceHistoryTaskZNode; +import org.apache.fluss.server.zk.data.ZkData.RebalanceHistoryZNode; import org.apache.fluss.testutils.common.AllCallbackWrapper; import org.apache.fluss.utils.clock.ManualClock; import org.apache.fluss.utils.clock.SystemClock; @@ -59,6 +63,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -69,6 +74,9 @@ import static org.apache.fluss.cluster.rebalance.RebalanceStatus.NOT_STARTED; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** Test for {@link RebalanceManager}. */ public class RebalanceManagerTest { @@ -149,6 +157,9 @@ void afterEach() throws Exception { scheduler.shutdown(); } zookeeperClient.deleteRebalanceTask(); + for (String rebalanceId : zookeeperClient.getChildren(RebalanceHistoryZNode.path())) { + zookeeperClient.deletePath(RebalanceHistoryTaskZNode.path(rebalanceId)); + } metadataManager = new MetadataManager( zookeeperClient, @@ -162,19 +173,273 @@ void testRebalanceWithoutTask() throws Exception { assertThat(rebalanceManager.getRebalanceStatus()).isNull(); String rebalanceId = "test-rebalance-id"; - RebalanceTask rebalanceTask = new RebalanceTask(rebalanceId, NOT_STARTED, new HashMap<>()); + RebalanceTask rebalanceTask = + new RebalanceTask(rebalanceId, NOT_STARTED, new HashMap<>(), -1, -1); zookeeperClient.registerRebalanceTask(rebalanceTask); assertThat(zookeeperClient.getRebalanceTask()).hasValue(rebalanceTask); // register a rebalance task with empty plan. + long beforeRegister = System.currentTimeMillis(); rebalanceManager.registerRebalance(rebalanceId, new HashMap<>(), NOT_STARTED); + long afterRegister = System.currentTimeMillis(); assertThat(rebalanceManager.getRebalanceId()).isEqualTo(rebalanceId); RebalanceStatus status = rebalanceManager.getRebalanceStatus(); assertThat(status).isNotNull(); assertThat(status).isEqualTo(COMPLETED); + + // An empty plan completes immediately, so started/completed are both stamped with + // "now" (real clock, since this test's rebalanceManager uses SystemClock). + RebalanceTask finalTask = zookeeperClient.getRebalanceTask().get(); + assertThat(finalTask.getRebalanceId()).isEqualTo(rebalanceId); + assertThat(finalTask.getRebalanceStatus()).isEqualTo(COMPLETED); + assertThat(finalTask.getExecutePlan()).isEmpty(); + assertThat(finalTask.getStartedAtMs()).isBetween(beforeRegister, afterRegister); + assertThat(finalTask.getCompletedAtMs()).isBetween(beforeRegister, afterRegister); + } + + @Test + void testGenerateRebalanceTaskStampsStartedAtMs() throws Exception { + ManualClock clock = new ManualClock(12_345L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + manager.startup(); + + RebalanceTask task = manager.generateRebalanceTask(Collections.emptyList()); + + assertThat(task.getStartedAtMs()).isEqualTo(12_345L); + assertThat(task.getCompletedAtMs()).isEqualTo(-1L); + + manager.close(); + } + + @Test + void testInitializeRestoresTimestampsOnFailover() throws Exception { + TableBucket tb1 = new TableBucket(1L, 0); + Map plan = new HashMap<>(); + plan.put( + tb1, + new RebalancePlanForBucket( + tb1, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); + + // Simulate a task left behind by a previous coordinator, with a real startedAtMs. + zookeeperClient.registerRebalanceTask( + new RebalanceTask("failover-test", NOT_STARTED, plan, 5_000L, -1)); + + ManualClock clock = new ManualClock(20_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + + // startup() -> initialize() restores the task and its timestamps from ZooKeeper. + manager.startup(); + + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(5_000L); + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(-1L); + + // Completing the restored task must keep the original startedAtMs. + manager.finishRebalanceTask(tb1, COMPLETED); + + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(5_000L); + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(20_000L); assertThat(zookeeperClient.getRebalanceTask()) - .hasValue(new RebalanceTask(rebalanceId, COMPLETED, new HashMap<>())); + .hasValue(new RebalanceTask("failover-test", COMPLETED, plan, 5_000L, 20_000L)); + + manager.close(); + } + + @Test + void testInitializeRestoresTaskWithoutTimestamps() throws Exception { + TableBucket tb1 = new TableBucket(1L, 0); + Map plan = new HashMap<>(); + plan.put( + tb1, + new RebalancePlanForBucket( + tb1, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); + + // A version-1 znode has no timestamp fields, so the deserializer yields -1. Completing + // such a restored task must leave startedAtMs unset rather than back-date it to the + // failover clock. + zookeeperClient.registerRebalanceTask( + new RebalanceTask("failover-v1-test", NOT_STARTED, plan, -1, -1)); + + ManualClock clock = new ManualClock(20_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + + manager.startup(); + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(-1L); + + manager.finishRebalanceTask(tb1, COMPLETED); + + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(-1L); + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(20_000L); + assertThat(zookeeperClient.getRebalanceTask()) + .hasValue(new RebalanceTask("failover-v1-test", COMPLETED, plan, -1, 20_000L)); + + manager.close(); + } + + @Test + void testInitializeDoesNotReCompleteRestoredFinalEmptyPlanTask() throws Exception { + // Simulate an empty-plan rebalance that already reached COMPLETED before the coordinator + // failed over, including its already-written history entry. + Map emptyPlan = new HashMap<>(); + RebalanceTask alreadyCompletedTask = + new RebalanceTask("failover-empty-plan-test", COMPLETED, emptyPlan, 5_000L, 8_000L); + zookeeperClient.registerRebalanceTask(alreadyCompletedTask); + zookeeperClient.registerRebalanceHistory(alreadyCompletedTask, 10); + + ManualClock clock = new ManualClock(50_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + + // startup() -> initialize() restores the already-final task; it must not be re-completed + // with the (much later) failover clock time. + manager.startup(); + + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(5_000L); + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(8_000L); + assertThat(zookeeperClient.getRebalanceTask()).hasValue(alreadyCompletedTask); + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(alreadyCompletedTask); + assertThat(manager.listRebalances()) + .containsExactly( + new RebalanceInfo("failover-empty-plan-test", COMPLETED, 5_000L, 8_000L)); + + manager.close(); + } + + @Test + void testInitializeKeepsStatusOfRestoredCanceledEmptyPlanTask() throws Exception { + // A restored final empty-plan task must keep its own status: re-running completion on + // failover would coerce CANCELED to COMPLETED and rewrite the znode and history. + Map emptyPlan = new HashMap<>(); + RebalanceTask canceledTask = + new RebalanceTask( + "failover-canceled-empty-plan-test", + RebalanceStatus.CANCELED, + emptyPlan, + 5_000L, + 8_000L); + zookeeperClient.registerRebalanceTask(canceledTask); + zookeeperClient.registerRebalanceHistory(canceledTask, 10); + + ManualClock clock = new ManualClock(50_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + + manager.startup(); + + assertThat(manager.getRebalanceStatus()).isEqualTo(RebalanceStatus.CANCELED); + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(5_000L); + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(8_000L); + assertThat(zookeeperClient.getRebalanceTask()).hasValue(canceledTask); + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(canceledTask); + assertThat(manager.listRebalances()) + .containsExactly( + new RebalanceInfo( + "failover-canceled-empty-plan-test", + RebalanceStatus.CANCELED, + 5_000L, + 8_000L)); + + manager.close(); + } + + @Test + void testCompleteRebalanceStampsCompletedAtMsAndWritesHistory() throws Exception { + ManualClock clock = new ManualClock(1_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + manager.startup(); + + TableBucket tb1 = new TableBucket(1L, 0); + Map plan = new HashMap<>(); + plan.put( + tb1, + new RebalancePlanForBucket( + tb1, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); + + zookeeperClient.registerRebalanceTask( + new RebalanceTask("complete-ts-test", NOT_STARTED, plan, 1_000L, -1)); + manager.registerRebalance("complete-ts-test", plan, NOT_STARTED, 1_000L, -1); + + clock.advanceTime(Duration.ofMillis(5_000)); + manager.finishRebalanceTask(tb1, COMPLETED); + + assertThat(manager.getCurrentStartedAtMs()).isEqualTo(1_000L); + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(6_000L); + assertThat(zookeeperClient.getRebalanceTask()) + .hasValue(new RebalanceTask("complete-ts-test", COMPLETED, plan, 1_000L, 6_000L)); + assertThat(zookeeperClient.getRebalanceHistory()) + .contains(new RebalanceTask("complete-ts-test", COMPLETED, plan, 1_000L, 6_000L)); + + manager.close(); + } + + @Test + void testCancelRebalanceStampsCompletedAtMsAndWritesHistory() throws Exception { + ManualClock clock = new ManualClock(1_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + manager.startup(); + + TableBucket tb1 = new TableBucket(1L, 0); + Map plan = new HashMap<>(); + plan.put( + tb1, + new RebalancePlanForBucket( + tb1, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); + + zookeeperClient.registerRebalanceTask( + new RebalanceTask("cancel-ts-test", NOT_STARTED, plan, 1_000L, -1)); + manager.registerRebalance("cancel-ts-test", plan, NOT_STARTED, 1_000L, -1); + + clock.advanceTime(Duration.ofMillis(3_000)); + manager.cancelRebalance("cancel-ts-test"); + + assertThat(manager.getCurrentCompletedAtMs()).isEqualTo(4_000L); + RebalanceTask canceledTask = + new RebalanceTask("cancel-ts-test", RebalanceStatus.CANCELED, plan, 1_000L, 4_000L); + assertThat(zookeeperClient.getRebalanceTask()).hasValue(canceledTask); + assertThat(zookeeperClient.getRebalanceHistory()).contains(canceledTask); + + manager.close(); } @Test @@ -202,7 +467,8 @@ void testTimeoutEnqueuesEvent() throws Exception { new RebalancePlanForBucket( tb2, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); - zookeeperClient.registerRebalanceTask(new RebalanceTask("timeout-test", NOT_STARTED, plan)); + zookeeperClient.registerRebalanceTask( + new RebalanceTask("timeout-test", NOT_STARTED, plan, -1, -1)); manager.registerRebalance("timeout-test", plan, NOT_STARTED); // Not yet timed out. @@ -250,7 +516,7 @@ void testTimeoutAfterCompletionIsNoOp() throws Exception { tb1, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); zookeeperClient.registerRebalanceTask( - new RebalanceTask("completion-test", NOT_STARTED, plan)); + new RebalanceTask("completion-test", NOT_STARTED, plan, -1, -1)); manager.registerRebalance("completion-test", plan, NOT_STARTED); // The task completes normally before timeout. @@ -292,7 +558,7 @@ void testTimeoutTreatsTaskAsCompleted() throws Exception { tb2, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); zookeeperClient.registerRebalanceTask( - new RebalanceTask("completed-test", NOT_STARTED, plan)); + new RebalanceTask("completed-test", NOT_STARTED, plan, -1, -1)); manager.registerRebalance("completed-test", plan, NOT_STARTED); // Timeout fires. @@ -314,6 +580,86 @@ void testTimeoutTreatsTaskAsCompleted() throws Exception { manager.close(); } + @Test + void testListRebalancesEmptyWhenNoRebalanceEverRun() { + assertThat(rebalanceManager.listRebalances()).isEmpty(); + } + + @Test + void testListRebalancesPropagatesZooKeeperReadFailure() throws Exception { + ZooKeeperClient failingZkClient = mock(ZooKeeperClient.class); + when(failingZkClient.getRebalanceHistory()) + .thenThrow(new RuntimeException("zk read failed")); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, + failingZkClient, + eventManager, + SystemClock.getInstance(), + executor); + manager.startup(); + + assertThatThrownBy(manager::listRebalances).isInstanceOf(FlussRuntimeException.class); + + manager.close(); + } + + @Test + void testListRebalancesCurrentFirstThenHistoryNewestFirstDeduped() throws Exception { + ManualClock clock = new ManualClock(1_000L); + RecordingEventManager eventManager = new RecordingEventManager(); + NoOpScheduledExecutor executor = new NoOpScheduledExecutor(); + CoordinatorEventProcessor eventProcessor = + buildCoordinatorEventProcessor(new Configuration()); + RebalanceManager manager = + new RebalanceManager( + eventProcessor, zookeeperClient, eventManager, clock, executor); + manager.startup(); + + TableBucket tb1 = new TableBucket(1L, 0); + Map plan = new HashMap<>(); + plan.put( + tb1, + new RebalancePlanForBucket( + tb1, 0, 0, Arrays.asList(0, 1, 2), Arrays.asList(0, 1, 2))); + + // First rebalance completes and lands in history. + zookeeperClient.registerRebalanceTask( + new RebalanceTask("rebalance-1", NOT_STARTED, plan, 1_000L, -1)); + manager.registerRebalance("rebalance-1", plan, NOT_STARTED, 1_000L, -1); + clock.advanceTime(Duration.ofMillis(1_000)); + manager.finishRebalanceTask(tb1, COMPLETED); + + // Second rebalance completes later, also lands in history. + clock.advanceTime(Duration.ofMillis(1_000)); + zookeeperClient.registerRebalanceTask( + new RebalanceTask("rebalance-2", NOT_STARTED, plan, clock.milliseconds(), -1)); + manager.registerRebalance("rebalance-2", plan, NOT_STARTED, clock.milliseconds(), -1); + clock.advanceTime(Duration.ofMillis(1_000)); + manager.finishRebalanceTask(tb1, COMPLETED); + + // Third rebalance is still in progress (current). + clock.advanceTime(Duration.ofMillis(1_000)); + zookeeperClient.registerRebalanceTask( + new RebalanceTask("rebalance-3", NOT_STARTED, plan, clock.milliseconds(), -1)); + manager.registerRebalance("rebalance-3", plan, NOT_STARTED, clock.milliseconds(), -1); + + List rebalanceInfos = manager.listRebalances(); + + // Current rebalance first, then history newest first; the current one must not be + // duplicated even though it will eventually also be written to history. + assertThat(rebalanceInfos).hasSize(3); + assertThat(rebalanceInfos.get(0).rebalanceId()).isEqualTo("rebalance-3"); + assertThat(rebalanceInfos.get(1).rebalanceId()).isEqualTo("rebalance-2"); + assertThat(rebalanceInfos.get(2).rebalanceId()).isEqualTo("rebalance-1"); + + manager.close(); + } + private CoordinatorEventProcessor buildCoordinatorEventProcessor(Configuration conf) { return new CoordinatorEventProcessor( zookeeperClient, 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..b7cc09297c3 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 @@ -43,6 +43,8 @@ import org.apache.fluss.server.zk.data.TableRegistration; import org.apache.fluss.server.zk.data.TabletServerRegistration; import org.apache.fluss.server.zk.data.ZkData.BucketIdZNode; +import org.apache.fluss.server.zk.data.ZkData.RebalanceHistoryTaskZNode; +import org.apache.fluss.server.zk.data.ZkData.RebalanceHistoryZNode; import org.apache.fluss.server.zk.data.lease.KvSnapshotLeaseMetadata; import org.apache.fluss.shaded.curator5.org.apache.curator.CuratorZookeeperClient; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework; @@ -62,6 +64,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -77,6 +80,7 @@ import static org.apache.fluss.server.utils.TableAssignmentUtils.generateAssignment; import static org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.common.ZKConfig.JUTE_MAXBUFFER; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.anyCollection; import static org.mockito.Mockito.doAnswer; @@ -810,9 +814,9 @@ void testRebalancePlan() throws Exception { Arrays.asList(0, 1, 2), Arrays.asList(1, 2, 3))); zookeeperClient.registerRebalanceTask( - new RebalanceTask("rebalance-task-1", NOT_STARTED, bucketPlan)); + new RebalanceTask("rebalance-task-1", NOT_STARTED, bucketPlan, -1, -1)); assertThat(zookeeperClient.getRebalanceTask()) - .hasValue(new RebalanceTask("rebalance-task-1", NOT_STARTED, bucketPlan)); + .hasValue(new RebalanceTask("rebalance-task-1", NOT_STARTED, bucketPlan, -1, -1)); bucketPlan = new HashMap<>(); bucketPlan.put( @@ -824,14 +828,192 @@ void testRebalancePlan() throws Exception { Arrays.asList(0, 1, 2), Arrays.asList(3, 4, 5))); zookeeperClient.registerRebalanceTask( - new RebalanceTask("rebalance-task-2", NOT_STARTED, bucketPlan)); + new RebalanceTask("rebalance-task-2", NOT_STARTED, bucketPlan, -1, -1)); assertThat(zookeeperClient.getRebalanceTask()) - .hasValue(new RebalanceTask("rebalance-task-2", NOT_STARTED, bucketPlan)); + .hasValue(new RebalanceTask("rebalance-task-2", NOT_STARTED, bucketPlan, -1, -1)); zookeeperClient.registerRebalanceTask( - new RebalanceTask("rebalance-task-2", COMPLETED, bucketPlan)); + new RebalanceTask("rebalance-task-2", COMPLETED, bucketPlan, -1, -1)); assertThat(zookeeperClient.getRebalanceTask()) - .hasValue(new RebalanceTask("rebalance-task-2", COMPLETED, bucketPlan)); + .hasValue(new RebalanceTask("rebalance-task-2", COMPLETED, bucketPlan, -1, -1)); + } + + @Test + void testRebalanceHistory() throws Exception { + Map bucketPlan = new HashMap<>(); + bucketPlan.put( + new TableBucket(0L, 0), + new RebalancePlanForBucket( + new TableBucket(0L, 0), + 0, + 1, + Arrays.asList(0, 1, 2), + Arrays.asList(1, 2, 0))); + + // register 12 completed rebalance tasks with retention count of 10, newest last. + for (int i = 0; i < 12; i++) { + zookeeperClient.registerRebalanceHistory( + new RebalanceTask("history-task-" + i, COMPLETED, bucketPlan, i, 100 + i), 10); + } + + List history = zookeeperClient.getRebalanceHistory(); + assertThat(history).hasSize(10); + // newest first, i.e. history-task-11 (completedAtMs=111) down to history-task-2 + // (completedAtMs=102); history-task-0 and history-task-1 were trimmed. + List rebalanceIds = new ArrayList<>(); + for (RebalanceTask task : history) { + rebalanceIds.add(task.getRebalanceId()); + } + assertThat(rebalanceIds) + .containsExactly( + "history-task-11", + "history-task-10", + "history-task-9", + "history-task-8", + "history-task-7", + "history-task-6", + "history-task-5", + "history-task-4", + "history-task-3", + "history-task-2"); + } + + @Test + void testRebalanceHistoryIsIdempotent() throws Exception { + Map bucketPlan = new HashMap<>(); + bucketPlan.put( + new TableBucket(0L, 0), + new RebalancePlanForBucket( + new TableBucket(0L, 0), + 0, + 1, + Arrays.asList(0, 1, 2), + Arrays.asList(1, 2, 0))); + + RebalanceTask original = + new RebalanceTask("idempotent-history-task", COMPLETED, bucketPlan, 0, 100); + zookeeperClient.registerRebalanceHistory(original, 10); + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(original); + + // A second write for the same rebalance id must overwrite the entry in place rather than + // fail on the existing znode (which would also skip the retention trim that follows it). + RebalanceTask updated = + new RebalanceTask("idempotent-history-task", COMPLETED, bucketPlan, 0, 200); + assertThatCode(() -> zookeeperClient.registerRebalanceHistory(updated, 10)) + .doesNotThrowAnyException(); + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(updated); + + // Retention trimming still runs after the idempotent write: push past the retention + // bound with fresh ids and confirm the oldest entry (the one just re-written) is trimmed. + for (int i = 0; i < 10; i++) { + zookeeperClient.registerRebalanceHistory( + new RebalanceTask( + "idempotent-history-task-" + i, COMPLETED, bucketPlan, i, 300 + i), + 10); + } + + List history = zookeeperClient.getRebalanceHistory(); + assertThat(history).hasSize(10); + List rebalanceIds = + history.stream().map(RebalanceTask::getRebalanceId).collect(Collectors.toList()); + assertThat(rebalanceIds).doesNotContain("idempotent-history-task"); + } + + @Test + void testRebalanceHistorySkipsCorruptEntries() throws Exception { + Map bucketPlan = new HashMap<>(); + bucketPlan.put( + new TableBucket(0L, 0), + new RebalancePlanForBucket( + new TableBucket(0L, 0), + 0, + 1, + Arrays.asList(0, 1, 2), + Arrays.asList(1, 2, 0))); + + RebalanceTask good = new RebalanceTask("good-history-task", COMPLETED, bucketPlan, 0, 100); + zookeeperClient.registerRebalanceHistory(good, 10); + + // Malformed JSON: not decodable at all. + zookeeperClient + .getCuratorClient() + .create() + .creatingParentsIfNeeded() + .forPath( + RebalanceHistoryTaskZNode.path("malformed-history-task"), + "not-json".getBytes(StandardCharsets.UTF_8)); + + // Decodable JSON, but with an unknown rebalance_status code (e.g. written by a newer + // coordinator, then read after a downgrade). + String unknownStatusJson = + "{\"version\":2,\"rebalance_id\":\"unknown-status-history-task\"," + + "\"rebalance_status\":99,\"started_at_ms\":1,\"completed_at_ms\":200," + + "\"rebalance_plan\":[]}"; + zookeeperClient + .getCuratorClient() + .create() + .creatingParentsIfNeeded() + .forPath( + RebalanceHistoryTaskZNode.path("unknown-status-history-task"), + unknownStatusJson.getBytes(StandardCharsets.UTF_8)); + + // Neither corrupt entry aborts the listing; only the good one is returned. + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(good); + + // Corrupt entries are not trimmed (they don't sort), but registering more history past + // the retention bound still trims the decodable entries correctly. + for (int i = 0; i < 10; i++) { + zookeeperClient.registerRebalanceHistory( + new RebalanceTask("more-history-task-" + i, COMPLETED, bucketPlan, i, 300 + i), + 10); + } + List history = zookeeperClient.getRebalanceHistory(); + assertThat(history).hasSize(10); + assertThat(history).doesNotContain(good); + assertThat(zookeeperClient.getChildren(RebalanceHistoryZNode.path())) + .contains("malformed-history-task", "unknown-status-history-task"); + } + + @Test + void testRebalanceHistoryTiesBrokenByRebalanceId() throws Exception { + Map bucketPlan = new HashMap<>(); + bucketPlan.put( + new TableBucket(0L, 0), + new RebalancePlanForBucket( + new TableBucket(0L, 0), + 0, + 1, + Arrays.asList(0, 1, 2), + Arrays.asList(1, 2, 0))); + + RebalanceTask taskB = new RebalanceTask("tie-task-b", COMPLETED, bucketPlan, 0, 500); + RebalanceTask taskA = new RebalanceTask("tie-task-a", COMPLETED, bucketPlan, 0, 500); + zookeeperClient.registerRebalanceHistory(taskB, 10); + zookeeperClient.registerRebalanceHistory(taskA, 10); + + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(taskA, taskB); + } + + @Test + void testDeleteRebalanceTaskDoesNotAffectSiblingHistory() throws Exception { + Map bucketPlan = new HashMap<>(); + bucketPlan.put( + new TableBucket(0L, 0), + new RebalancePlanForBucket( + new TableBucket(0L, 0), + 0, + 1, + Arrays.asList(0, 1, 2), + Arrays.asList(1, 2, 0))); + + zookeeperClient.registerRebalanceTask( + new RebalanceTask("sibling-task", NOT_STARTED, bucketPlan, -1, -1)); + RebalanceTask history = + new RebalanceTask("sibling-history-task", COMPLETED, bucketPlan, 0, 100); + zookeeperClient.registerRebalanceHistory(history, 10); + + assertThatCode(() -> zookeeperClient.deleteRebalanceTask()).doesNotThrowAnyException(); + assertThat(zookeeperClient.getRebalanceHistory()).containsExactly(history); } @Test diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerdeTest.java index 5711f7da3f7..02ca333f432 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerdeTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/RebalanceTaskJsonSerdeTest.java @@ -20,12 +20,18 @@ import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.utils.json.JsonSerdeTestBase; +import org.apache.fluss.utils.json.JsonSerdeUtils; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.HashMap; import java.util.Map; import static org.apache.fluss.cluster.rebalance.RebalanceStatus.NOT_STARTED; +import static org.assertj.core.api.Assertions.assertThat; /** Test for {@link RebalanceTaskJsonSerde}. */ public class RebalanceTaskJsonSerdeTest extends JsonSerdeTestBase { @@ -80,14 +86,15 @@ protected RebalanceTask[] createObjects() { Arrays.asList(0, 1, 2), Arrays.asList(3, 4, 5))); return new RebalanceTask[] { - new RebalanceTask("rebalance-task-21jd", NOT_STARTED, bucketPlan) + new RebalanceTask("rebalance-task-21jd", NOT_STARTED, bucketPlan, 1000L, 2000L) }; } @Override protected String[] expectedJsons() { return new String[] { - "{\"version\":1,\"rebalance_id\":\"rebalance-task-21jd\",\"rebalance_status\":0,\"rebalance_plan\":" + "{\"version\":2,\"rebalance_id\":\"rebalance-task-21jd\",\"rebalance_status\":0," + + "\"started_at_ms\":1000,\"completed_at_ms\":2000,\"rebalance_plan\":" + "[{\"table_id\":0,\"buckets\":" + "[{\"bucket_id\":1,\"original_leader\":1,\"new_leader\":1,\"origin_replicas\":[0,1,2],\"new_replicas\":[1,2,3]}," + "{\"bucket_id\":0,\"original_leader\":0,\"new_leader\":3,\"origin_replicas\":[0,1,2],\"new_replicas\":[3,4,5]}]}," @@ -98,4 +105,19 @@ protected String[] expectedJsons() { + "{\"bucket_id\":0,\"original_leader\":0,\"new_leader\":3,\"origin_replicas\":[0,1,2],\"new_replicas\":[3,4,5]}]}]}" }; } + + @Test + void testVersion1Compatibility() throws IOException { + // A version-1 document has no started_at_ms/completed_at_ms fields; both must + // deserialize as -1. + String v1Json = + "{\"version\":1,\"rebalance_id\":\"rebalance-task-21jd\",\"rebalance_status\":0," + + "\"rebalance_plan\":[]}"; + RebalanceTask actual = + JsonSerdeUtils.readValue( + v1Json.getBytes(StandardCharsets.UTF_8), RebalanceTaskJsonSerde.INSTANCE); + + assertThat(actual.getStartedAtMs()).isEqualTo(-1L); + assertThat(actual.getCompletedAtMs()).isEqualTo(-1L); + } } From 8757b6a971f02c0ac9dc9493ceffb49c9690c497 Mon Sep 17 00:00:00 2001 From: Muhammet Orazov Date: Tue, 18 Aug 2026 15:43:10 +0200 Subject: [PATCH 2/2] [server] Expose rebalance timestamps in progress API and procedures RebalanceProgress now carries startedAtMs/completedAtMs (epoch millis, -1 when unset), populated by the coordinator and transported via two new optional fields on ListRebalanceProgressResponse. The progress JSON serializer includes the timestamps when set. The sys.list_rebalance() Flink procedure gains started_at/completed_at TIMESTAMP_LTZ(3) output columns, and when called without a rebalance id it now returns one row per known rebalance (the current one plus the retained history via listRebalances(), newest first) instead of only the current one; historical rows carry null progress and plan detail. Closes #3965. --- .../client/utils/ClientRpcMessageUtils.java | 4 +- .../fluss/client/admin/RebalanceITCase.java | 2 + .../cluster/rebalance/RebalanceProgress.java | 28 +++++++- .../RebalanceProgressJsonSerializer.java | 8 +++ .../RebalanceProgressJsonSerializerTest.java | 30 +++++++- .../ListRebalanceProcessProcedure.java | 69 ++++++++++++++----- .../flink/procedure/FlinkProcedureITCase.java | 18 ++++- fluss-rpc/src/main/proto/FlussApi.proto | 2 + .../rebalance/RebalanceManager.java | 7 +- .../server/utils/ServerRpcMessageUtils.java | 6 ++ website/docs/engine-flink/procedures.md | 22 +++--- .../docs/maintenance/operations/rebalance.md | 8 ++- 12 files changed, 170 insertions(+), 34 deletions(-) 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 672ea290129..2f50c9f1e53 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 @@ -617,7 +617,9 @@ public static Optional toRebalanceProgress( response.getRebalanceId(), totalRebalanceStatus, progress, - rebalanceProgress)); + rebalanceProgress, + response.hasStartedAtMs() ? response.getStartedAtMs() : -1, + response.hasCompletedAtMs() ? response.getCompletedAtMs() : -1)); } public static List toRebalanceInfos(ListRebalancesResponse response) { diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java index b4a6c5ca474..cdde77b0236 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java @@ -257,6 +257,8 @@ void testListRebalanceProgress() throws Exception { RebalanceProgress progress = progressOpt.get(); assertThat(progress.progress()).isEqualTo(1d); assertThat(progress.status()).isEqualTo(RebalanceStatus.COMPLETED); + assertThat(progress.startedAtMs()).isGreaterThanOrEqualTo(0); + assertThat(progress.completedAtMs()).isGreaterThanOrEqualTo(progress.startedAtMs()); // test list and cancel an un-existed rebalance id. assertThatThrownBy(() -> admin.listRebalanceProgress("unexisted-rebalance-id").get()) diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgress.java b/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgress.java index e203159e116..4837dfbf043 100644 --- a/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgress.java +++ b/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgress.java @@ -46,16 +46,29 @@ public class RebalanceProgress { /** The rebalance progress for each tabletBucket. */ private final Map progressForBucketMap; + /** The time when the rebalance was started, in epoch milliseconds. -1 if unknown. */ + private final long startedAtMs; + + /** + * The time when the rebalance reached a final status, in epoch milliseconds. -1 if the + * rebalance is still in progress or the completion time is unknown. + */ + private final long completedAtMs; + public RebalanceProgress( String rebalanceId, RebalanceStatus rebalanceStatus, double progress, - Map progressForBucketMap) { + Map progressForBucketMap, + long startedAtMs, + long completedAtMs) { this.rebalanceId = rebalanceId; // TODO: we may derive the overall progress and status from progressForBucketMap this.rebalanceStatus = checkNotNull(rebalanceStatus); this.progress = progress; this.progressForBucketMap = checkNotNull(progressForBucketMap); + this.startedAtMs = startedAtMs; + this.completedAtMs = completedAtMs; } public String rebalanceId() { @@ -74,6 +87,19 @@ public Map progressForBucketMap() { return progressForBucketMap; } + /** The time when the rebalance was started, in epoch milliseconds. -1 if unknown. */ + public long startedAtMs() { + return startedAtMs; + } + + /** + * The time when the rebalance reached a final status, in epoch milliseconds. -1 if the + * rebalance is still in progress or the completion time is unknown. + */ + public long completedAtMs() { + return completedAtMs; + } + public String formatAsPercentage() { if (progress < 0) { return "NONE"; diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializer.java b/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializer.java index 448b42f4df5..7011ddf26e8 100644 --- a/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializer.java +++ b/fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializer.java @@ -32,6 +32,8 @@ public class RebalanceProgressJsonSerializer implements JsonSerializer= 0) { + generator.writeNumberField(STARTED_AT_MS, rebalanceProgress.startedAtMs()); + } + if (rebalanceProgress.completedAtMs() >= 0) { + generator.writeNumberField(COMPLETED_AT_MS, rebalanceProgress.completedAtMs()); + } generator.writeStringField(PROGRESS, rebalanceProgress.formatAsPercentage()); Map resultForBucketMap = diff --git a/fluss-common/src/test/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializerTest.java b/fluss-common/src/test/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializerTest.java index 4cf48303c1e..f678726c88d 100644 --- a/fluss-common/src/test/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializerTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/cluster/rebalance/RebalanceProgressJsonSerializerTest.java @@ -42,6 +42,27 @@ public void testSerializer() { assertThat(serialize).isEqualTo(createProgressJson()); } + @Test + public void testSerializerOmitsUnsetTimestamps() { + RebalanceProgress progress = + new RebalanceProgress( + "rebalance-task-21jd", + RebalanceStatus.REBALANCING, + -1d, + new HashMap<>(), + -1, + -1); + String serialize = + new String( + JsonSerdeUtils.writeValueAsBytes( + progress, RebalanceProgressJsonSerializer.INSTANCE), + StandardCharsets.UTF_8); + assertThat(serialize) + .isEqualTo( + "{\"rebalance_id\":\"rebalance-task-21jd\",\"rebalance_status\":1," + + "\"progress\":\"NONE\",\"progress_for_buckets\":[]}"); + } + private RebalanceProgress createProgressObj() { Map progressForBucketMap = new HashMap<>(); progressForBucketMap.put( @@ -65,11 +86,16 @@ private RebalanceProgress createProgressObj() { Arrays.asList(3, 4, 5)), RebalanceStatus.COMPLETED)); return new RebalanceProgress( - "rebalance-task-21jd", RebalanceStatus.COMPLETED, 1d, progressForBucketMap); + "rebalance-task-21jd", + RebalanceStatus.COMPLETED, + 1d, + progressForBucketMap, + 1735689600000L, + 1735689660000L); } private String createProgressJson() { - return "{\"rebalance_id\":\"rebalance-task-21jd\",\"rebalance_status\":3,\"progress\":\"100%\",\"progress_for_buckets\":" + return "{\"rebalance_id\":\"rebalance-task-21jd\",\"rebalance_status\":3,\"started_at_ms\":1735689600000,\"completed_at_ms\":1735689660000,\"progress\":\"100%\",\"progress_for_buckets\":" + "[{\"table_id\":1,\"bucket_id\":0,\"partition_id\":0,\"original_leader\":0,\"new_leader\":3,\"origin_replicas\":[0,1,2],\"new_replicas\":[3,4,5],\"rebalance_status\":3}," + "{\"table_id\":0,\"bucket_id\":0,\"original_leader\":0,\"new_leader\":3,\"origin_replicas\":[0,1,2],\"new_replicas\":[3,4,5],\"rebalance_status\":3}]}"; } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListRebalanceProcessProcedure.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListRebalanceProcessProcedure.java index 6dd06d6b3be..91a0a663533 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListRebalanceProcessProcedure.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListRebalanceProcessProcedure.java @@ -18,6 +18,7 @@ package org.apache.fluss.flink.procedure; import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.cluster.rebalance.RebalanceInfo; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceProgressJsonSerializer; import org.apache.fluss.utils.json.JsonSerdeUtils; @@ -31,18 +32,21 @@ import javax.annotation.Nullable; import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; import java.util.Optional; /** * Procedure to list rebalance progress. * *

This procedure allows querying rebalance progress. See {@link - * Admin#listRebalanceProgress(String)} for more details. + * Admin#listRebalanceProgress(String)} and {@link Admin#listRebalances()} for more details. * *

Usage examples: * * 

- * -- List the rebalance progress without rebalance id
+ * -- List the current rebalance and the retained history of finished rebalances
  * CALL sys.list_rebalance();
  *
  * -- List the rebalance progress with rebalance id
@@ -60,23 +64,54 @@ public class ListRebalanceProcessProcedure extends ProcedureBase {
             },
             output =
                     @DataTypeHint(
-                            "ROW"))
+                            "ROW"))
     public Row[] call(ProcedureContext context, @Nullable String rebalanceId) throws Exception {
-        Optional progressOpt = admin.listRebalanceProgress(rebalanceId).get();
+        if (rebalanceId != null) {
+            Optional progressOpt =
+                    admin.listRebalanceProgress(rebalanceId).get();
+            return progressOpt.map(progress -> new Row[] {toRow(progress)}).orElse(new Row[0]);
+        }
 
-        if (!progressOpt.isPresent()) {
-            return new Row[0];
+        // Without an id, list one row per known rebalance (the current one plus the retained
+        // history), ordered from most recent to oldest. Only the current rebalance carries
+        // per-bucket progress and plan detail. The two RPCs are separate round-trips, so a
+        // rebalance that starts or finishes between them can leave one row briefly out of
+        // date until the next call.
+        Optional currentOpt = admin.listRebalanceProgress(null).get();
+        List rebalanceInfos = admin.listRebalances().get();
+        List rows = new ArrayList<>(rebalanceInfos.size());
+        for (RebalanceInfo info : rebalanceInfos) {
+            if (currentOpt.isPresent()
+                    && currentOpt.get().rebalanceId().equals(info.rebalanceId())) {
+                rows.add(toRow(currentOpt.get()));
+            } else {
+                rows.add(
+                        Row.of(
+                                info.rebalanceId(),
+                                info.status().toString(),
+                                null,
+                                null,
+                                toInstant(info.startedAtMs()),
+                                toInstant(info.completedAtMs())));
+            }
         }
-        RebalanceProgress progress = progressOpt.get();
-        return new Row[] {
-            Row.of(
-                    progress.rebalanceId(),
-                    progress.status().toString(),
-                    progress.formatAsPercentage(),
-                    new String(
-                            JsonSerdeUtils.writeValueAsBytes(
-                                    progress, RebalanceProgressJsonSerializer.INSTANCE),
-                            StandardCharsets.UTF_8))
-        };
+        return rows.toArray(new Row[0]);
+    }
+
+    private static Row toRow(RebalanceProgress progress) {
+        return Row.of(
+                progress.rebalanceId(),
+                progress.status().toString(),
+                progress.formatAsPercentage(),
+                new String(
+                        JsonSerdeUtils.writeValueAsBytes(
+                                progress, RebalanceProgressJsonSerializer.INSTANCE),
+                        StandardCharsets.UTF_8),
+                toInstant(progress.startedAtMs()),
+                toInstant(progress.completedAtMs()));
+    }
+
+    private static @Nullable Instant toInstant(long epochMs) {
+        return epochMs < 0 ? null : Instant.ofEpochMilli(epochMs);
     }
 }
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
index f1c71f0c26c..7ae24d4e9df 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java
@@ -807,13 +807,29 @@ void testListRebalanceProgress() throws Exception {
                                     .collect()) {
                         List listProgressResult = CollectionUtil.iteratorToList(rows);
                         Row row = listProgressResult.get(0);
-                        assertThat(row.getArity()).isEqualTo(4);
+                        assertThat(row.getArity()).isEqualTo(6);
                         assertThat(row.getField(0)).isEqualTo(progress.rebalanceId());
                         assertThat(row.getField(1)).isEqualTo(RebalanceStatus.COMPLETED.toString());
                         assertThat((String) row.getField(2)).endsWith("%");
                         assertThat((String) row.getField(3)).startsWith("{\"rebalance_id\":");
+                        assertThat(row.getField(4)).isNotNull();
+                        assertThat(row.getField(5)).isNotNull();
                     }
                 });
+
+        try (CloseableIterator rows =
+                tEnv.executeSql(String.format("Call %s.sys.list_rebalance()", CATALOG_NAME))
+                        .collect()) {
+            List listResult = CollectionUtil.iteratorToList(rows);
+            assertThat(listResult).isNotEmpty();
+            Row row = listResult.get(0);
+            assertThat(row.getArity()).isEqualTo(6);
+            assertThat(row.getField(0)).isEqualTo(progress.rebalanceId());
+            assertThat(row.getField(1)).isEqualTo(RebalanceStatus.COMPLETED.toString());
+            assertThat((String) row.getField(3)).startsWith("{\"rebalance_id\":");
+            assertThat(row.getField(4)).isNotNull();
+            assertThat(row.getField(5)).isNotNull();
+        }
     }
 
     @Test
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 2986205ecab..eaf68d62695 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -756,6 +756,8 @@ message ListRebalanceProgressResponse {
   optional string rebalance_id = 1;
   optional int32 rebalance_status = 2;
   repeated PbRebalanceProgressForTable table_progress = 3;
+  optional int64 started_at_ms = 4;
+  optional int64 completed_at_ms = 5;
 }
 
 message CancelRebalanceRequest {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
index 4c25d5bcd25..f75336b6a86 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java
@@ -318,7 +318,12 @@ public void finishRebalanceTask(TableBucket tableBucket, RebalanceStatus statusF
         progressForBucketMap.putAll(finishedRebalanceTasks);
         // the progress will be set at client.
         return new RebalanceProgress(
-                currentRebalanceId, rebalanceStatus, 0.0, progressForBucketMap);
+                currentRebalanceId,
+                rebalanceStatus,
+                0.0,
+                progressForBucketMap,
+                currentStartedAtMs,
+                currentCompletedAtMs);
     }
 
     /**
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index c6f32decb76..e3e57eec3ee 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -2164,6 +2164,12 @@ public static ListRebalanceProgressResponse makeListRebalanceProgressResponse(
                 new ListRebalanceProgressResponse()
                         .setRebalanceId(rebalanceProgress.rebalanceId())
                         .setRebalanceStatus(rebalanceProgress.status().getCode());
+        if (rebalanceProgress.startedAtMs() >= 0) {
+            response.setStartedAtMs(rebalanceProgress.startedAtMs());
+        }
+        if (rebalanceProgress.completedAtMs() >= 0) {
+            response.setCompletedAtMs(rebalanceProgress.completedAtMs());
+        }
 
         Map> tableIdToPbBuckets = new HashMap<>();
         for (Map.Entry progressForBucket :
diff --git a/website/docs/engine-flink/procedures.md b/website/docs/engine-flink/procedures.md
index 24baa9c08ea..aa6221a2b64 100644
--- a/website/docs/engine-flink/procedures.md
+++ b/website/docs/engine-flink/procedures.md
@@ -475,12 +475,12 @@ CALL sys.rebalance('RACK_AWARE,REPLICA_DISTRIBUTION,LEADER_DISTRIBUTION');
 
 ### list_rebalance
 
-Query the progress and status of a rebalance operation. This procedure allows you to monitor ongoing or completed rebalance operations to track their progress and view detailed information about bucket movements.
+Query the progress and status of rebalance operations. This procedure allows you to monitor the ongoing rebalance operation, view detailed information about bucket movements, and list the retained history of finished rebalances.
 
 **Syntax:**
 
 ```sql
--- List the most recent rebalance progress
+-- List the current rebalance and the retained history of finished rebalances
 CALL [catalog_name.]sys.list_rebalance()
 
 -- List a specific rebalance progress by ID
@@ -491,18 +491,20 @@ CALL [catalog_name.]sys.list_rebalance(
 
 **Parameters:**
 
-- `rebalanceId` (optional): The rebalance ID to query. If omitted, returns the progress of the most recent rebalance operation. The rebalance ID is returned when calling the `rebalance` procedure.
+- `rebalanceId` (optional): The rebalance ID to query. If omitted, returns one row per known rebalance: the most recent rebalance followed by the retained history of finished rebalances, newest first. The ID to pass here is the value returned by the `rebalance` procedure.
 
-**Returns:** An array of strings containing:
-- Rebalance ID: The unique identifier of the rebalance operation
-- Rebalance total status: The overall status of the rebalance. Possible values are:
+**Returns:** One row per rebalance with the following columns:
+- `rebalance_id`: The unique identifier of the rebalance operation
+- `rebalance_status`: The overall status of the rebalance. Possible values are:
     - `NOT_STARTED`: The rebalance has been created but not yet started
     - `REBALANCING`: The rebalance is currently in progress
     - `COMPLETED`: The rebalance has successfully completed
     - `FAILED`: The rebalance has failed
     - `CANCELED`: The rebalance has been canceled
-- Rebalance progress: The completion percentage (e.g., `75.5%`)
-- Rebalance detail progress for bucket: Detailed progress information for each bucket being moved
+- `rebalance_progress`: The completion percentage (e.g., `75.5%`). `NULL` for historical rows, which do not carry per-bucket detail
+- `rebalance_plan`: Detailed progress information for each bucket being moved, as JSON. `NULL` for historical rows
+- `started_at`: The time the rebalance was started. `NULL` if unknown
+- `completed_at`: The time the rebalance reached a final status. `NULL` while the rebalance is still in progress
 
 If no rebalance is found, returns empty line.
 
@@ -512,7 +514,7 @@ If no rebalance is found, returns empty line.
 -- Use the Fluss catalog (replace 'fluss_catalog' with your catalog name if different)
 USE fluss_catalog;
 
--- List the most recent rebalance progress
+-- List the current rebalance and the retained history of finished rebalances
 CALL sys.list_rebalance();
 
 -- List a specific rebalance progress by ID
@@ -537,7 +539,7 @@ CALL [catalog_name.]sys.cancel_rebalance(
 
 **Parameters:**
 
-- `rebalanceId` (optional): The rebalance ID to cancel. If omitted, cancels the most recent rebalance operation. The rebalance ID is returned when calling the `rebalance` procedure.
+- `rebalanceId` (optional): The rebalance ID to cancel. If omitted, cancels the most recent rebalance operation. The ID to pass here is the value returned by the `rebalance` procedure.
 
 **Returns:** An array with a single element `'success'` if the operation completes successfully.
 
diff --git a/website/docs/maintenance/operations/rebalance.md b/website/docs/maintenance/operations/rebalance.md
index 418b7c585f2..20cd2e4eb8c 100644
--- a/website/docs/maintenance/operations/rebalance.md
+++ b/website/docs/maintenance/operations/rebalance.md
@@ -82,6 +82,7 @@ Goals are processed in the order specified. When using `RACK_AWARE`, always plac
 Track the rebalance operation using the returned rebalance ID:
 
 ```java
+import org.apache.fluss.cluster.rebalance.RebalanceInfo;
 import org.apache.fluss.cluster.rebalance.RebalanceProgress;
 import org.apache.fluss.cluster.rebalance.RebalanceStatus;
 
@@ -97,11 +98,16 @@ if (progress.isPresent()) {
     // Check if rebalance is complete
     if (p.status() == RebalanceStatus.COMPLETED) {
         System.out.println("Rebalance completed successfully!");
+        System.out.println("Took " + (p.completedAtMs() - p.startedAtMs()) + " ms");
     }
 }
 
 // Query the most recent rebalance progress (if rebalanceId is not provided)
 Optional latestProgress = admin.listRebalanceProgress(null).get();
+
+// List the current rebalance plus a bounded history (last 10) of finished rebalances,
+// newest first, as summaries (id, status, started/completed timestamps)
+List rebalances = admin.listRebalances().get();
 ```
 
 Rebalance statuses:
@@ -216,7 +222,7 @@ For rebalancing operations, Fluss provides convenient Flink stored procedures th
 - **add_server_tag**: Tag servers before rebalancing
 - **remove_server_tag**: Remove tags after rebalancing
 - **rebalance**: Trigger rebalance operation
-- **list_rebalance**: Monitor rebalance progress
+- **list_rebalance**: Monitor rebalance progress and list the retained history of finished rebalances
 - **cancel_rebalance**: Cancel ongoing rebalance
 
 Example using Flink SQL: