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..6800dd67ac1 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, current (if any) followed by history, newest + * first. + * + * + * + * @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..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 @@ -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; @@ -614,7 +617,26 @@ 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) { + 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( 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..1a385b250a4 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; @@ -255,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()) @@ -270,6 +274,59 @@ 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 (covered by + // RebalanceManagerTest instead). + 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..359944a5b08 --- /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 (current or historical), returned by {@link + * org.apache.fluss.client.admin.Admin#listRebalances()}. 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-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..8a51f87dfae 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,53 @@ 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: one row per known rebalance (current + retained history), newest
+        // first. Only the current rebalance carries per-bucket progress and plan detail.
+        // The two RPCs are separate round-trips: a rebalance finishing or starting between
+        // them can render one transiently stale row, corrected on 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..00babb04749 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,31 @@ 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();
                     }
                 });
+
+        // Without an id: one row per known rebalance (current + history), newest first; the
+        // just-completed rebalance is still the current one, so its row carries plan detail.
+        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-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..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 {
@@ -765,6 +767,13 @@ message CancelRebalanceRequest {
 message CancelRebalanceResponse {
 }
 
+message ListRebalancesRequest {
+}
+
+message ListRebalancesResponse {
+  repeated PbRebalanceInfo rebalance_infos = 1;
+}
+
 // ------------------------------------------------------------------------------------------
 // Producer Offset Management
 // ------------------------------------------------------------------------------------------
@@ -1245,6 +1254,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..c6c7eaf82d9
--- /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 (current and history). */
+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..5a8b147f459 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,22 @@ 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.
+     *
+     * 
Used both when a new rebalance is triggered (timestamps taken from the just-built {@link
+     * RebalanceTask}) and when the coordinator restores a task from ZooKeeper on failover (see
+     * {@link #initialize()}), so the restored timestamps are not lost.
+     */
+    public void registerRebalance(
+            String rebalanceId,
+            Map rebalancePlan,
+            RebalanceStatus newStatus,
+            long startedAtMs,
+            long completedAtMs) {
         checkNotClosed();
         registerTime = System.currentTimeMillis();
         // first clear all exists tasks.
@@ -206,7 +236,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;
         }
@@ -284,7 +324,71 @@ 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);
+    }
+
+    /**
+     * 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 bounded ZooKeeper
+     * history, newest first. The current rebalance is de-duplicated against history: once it
+     * reaches a final status it is also written to history (see {@link #completeRebalance()} and
+     * {@link #cancelRebalance(String)}), so it would otherwise appear twice.
+     *
+     * 
Reads ZooKeeper and only touches the passed-in snapshot, so unlike the rest of this class
+     * it may run off the coordinator event thread (see the ioExecutor offload in {@link
+     * CoordinatorEventProcessor}).
+     */
+    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) {
@@ -308,20 +412,39 @@ 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(
+                // Prefer the just-read task's startedAtMs; fall back to the in-memory value when
+                // the znode predates the timestamp fields (version 1), where both are -1.
+                long startedAtMs =
+                        rebalanceTask.getStartedAtMs() >= 0
+                                ? rebalanceTask.getStartedAtMs()
+                                : currentStartedAtMs;
+                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 +515,44 @@ private void processNewRebalanceTask() {
 
     private void completeRebalance() {
         checkNotClosed();
+        long completedAtMs = clock.milliseconds();
         try {
             Optional rebalanceTaskOpt = zkClient.getRebalanceTask();
             Map bucketPlan;
+            // Prefer the just-read task's startedAtMs; fall back to the in-memory value when the
+            // znode predates the timestamp fields (version 1), or when the read below finds no
+            // task at all.
+            long startedAtMs = currentStartedAtMs;
             if (rebalanceTaskOpt.isPresent()) {
-                bucketPlan = rebalanceTaskOpt.get().getExecutePlan();
+                RebalanceTask rebalanceTask = rebalanceTaskOpt.get();
+                bucketPlan = rebalanceTask.getExecutePlan();
+                if (rebalanceTask.getStartedAtMs() >= 0) {
+                    startedAtMs = rebalanceTask.getStartedAtMs();
+                }
             } 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();
@@ -469,7 +613,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 +677,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..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
@@ -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;
@@ -2161,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 :
@@ -2187,6 +2196,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..5025bdaa3a0 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())) {
+            // getOrEmpty performs the ZK read; let genuine I/O failures propagate. Only the
+            // decode of already-fetched bytes is isolated per entry below.
+            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..742ecb31ed4 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,40 @@ 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..918acda9643 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,21 @@ 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 event
+        // manager only logs escaping throwables, so the handler must complete the callback itself
+        // or the client blocks until the RPC times out.
+        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..8da3e9afb57 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,231 @@ 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("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 (locked by
+        // RebalanceTaskJsonSerdeTest#testVersion1Compatibility). 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 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(rebalanceId, COMPLETED, new HashMap<>()));
+                .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 +425,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 +474,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 +516,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 +538,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..9e7d5629e8a 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 used to throw NodeExistsException (and skip
+        // the trim below it); it must now be a no-op setData instead.
+        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);
+    }
 }
diff --git a/website/docs/engine-flink/procedures.md b/website/docs/engine-flink/procedures.md
index 24baa9c08ea..2be02bc7a96 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 rebalance ID is returned when calling 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
diff --git a/website/docs/maintenance/operations/rebalance.md b/website/docs/maintenance/operations/rebalance.md
index 418b7c585f2..160216097f7 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,17 @@ if (progress.isPresent()) {
     // Check if rebalance is complete
     if (p.status() == RebalanceStatus.COMPLETED) {
         System.out.println("Rebalance completed successfully!");
+        // startedAtMs()/completedAtMs() are epoch milliseconds, -1 if unset
+        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 +223,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: