[server] Retain bounded rebalance history and add listRebalances API

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<Optional<RebalanceProgress>> listRebalanceProgress(
      */
     CompletableFuture<Void> cancelRebalance(@Nullable String rebalanceId);
 
+    /**
+     * List a summary of all known rebalance tasks, current (if any) followed by history, newest
+     * first.
+     *
+     * <ul>
+     *   <li>{@link AuthorizationException} If the authenticated user doesn't have cluster
+     *       permissions.
+     * </ul>
+     *
+     * @return the rebalance summaries.
+     */
+    CompletableFuture<List<RebalanceInfo>> listRebalances();
+
     // ==================================================================================
     // Producer Offset Management APIs (for Exactly-Once Semantics)
     // ==================================================================================

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<Void> cancelRebalance(@Nullable String rebalanceId) {
         return gateway.cancelRebalance(request).thenApply(r -> null);
     }
 
+    @Override
+    public CompletableFuture<List<RebalanceInfo>> listRebalances() {
+        ListRebalancesRequest request = new ListRebalancesRequest();
+        return gateway.listRebalances(request).thenApply(ClientRpcMessageUtils::toRebalanceInfos);
+    }
+
     // ==================================================================================
     // Producer Offset Management APIs (for Exactly-Once Semantics)
     // ==================================================================================

fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java

@@ -31,6 +31,7 @@
 import org.apache.fluss.client.metadata.RemoteLogManifestInfo;
 import org.apache.fluss.client.write.KvWriteBatch;
 import org.apache.fluss.client.write.ReadyWriteBatch;
+import org.apache.fluss.cluster.rebalance.RebalanceInfo;
 import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket;
 import org.apache.fluss.cluster.rebalance.RebalanceProgress;
 import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket;
@@ -68,6 +69,7 @@
 import org.apache.fluss.rpc.messages.ListOffsetsRequest;
 import org.apache.fluss.rpc.messages.ListPartitionInfosResponse;
 import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse;
+import org.apache.fluss.rpc.messages.ListRebalancesResponse;
 import org.apache.fluss.rpc.messages.ListRemoteLogManifestsResponse;
 import org.apache.fluss.rpc.messages.LookupRequest;
 import org.apache.fluss.rpc.messages.MetadataRequest;
@@ -89,6 +91,7 @@
 import org.apache.fluss.rpc.messages.PbProduceLogReqForBucket;
 import org.apache.fluss.rpc.messages.PbProducerTableOffsets;
 import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
+import org.apache.fluss.rpc.messages.PbRebalanceInfo;
 import org.apache.fluss.rpc.messages.PbRebalancePlanForBucket;
 import org.apache.fluss.rpc.messages.PbRebalanceProgressForBucket;
 import org.apache.fluss.rpc.messages.PbRebalanceProgressForTable;
@@ -617,6 +620,23 @@ public static Optional<RebalanceProgress> toRebalanceProgress(
                         rebalanceProgress));
     }
 
+    public static List<RebalanceInfo> toRebalanceInfos(ListRebalancesResponse response) {
+        List<RebalanceInfo> rebalanceInfos = new ArrayList<>();
+        for (PbRebalanceInfo pbRebalanceInfo : response.getRebalanceInfosList()) {
+            rebalanceInfos.add(
+                    new RebalanceInfo(
+                            pbRebalanceInfo.getRebalanceId(),
+                            RebalanceStatus.of(pbRebalanceInfo.getRebalanceStatus()),
+                            pbRebalanceInfo.hasStartedAtMs()
+                                    ? pbRebalanceInfo.getStartedAtMs()
+                                    : -1,
+                            pbRebalanceInfo.hasCompletedAtMs()
+                                    ? pbRebalanceInfo.getCompletedAtMs()
+                                    : -1));
+        }
+        return rebalanceInfos;
+    }
+
     private static RebalancePlanForBucket toRebalancePlanForBucket(
             long tableId, PbRebalancePlanForBucket rebalancePlan) {
         TableBucket tableBucket =

fluss-client/src/test/java/org/apache/fluss/client/admin/RebalanceITCase.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.client.Connection;
 import org.apache.fluss.client.ConnectionFactory;
 import org.apache.fluss.cluster.rebalance.GoalType;
+import org.apache.fluss.cluster.rebalance.RebalanceInfo;
 import org.apache.fluss.cluster.rebalance.RebalanceProgress;
 import org.apache.fluss.cluster.rebalance.RebalanceStatus;
 import org.apache.fluss.cluster.rebalance.ServerTag;
@@ -42,6 +43,7 @@
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.Optional;
 
 import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
@@ -270,6 +272,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<RebalanceProgress> progressOpt =
+                            admin.listRebalanceProgress(rebalanceId).get();
+                    assertThat(progressOpt).isPresent();
+                    assertThat(progressOpt.get().status()).isEqualTo(RebalanceStatus.COMPLETED);
+                });
+
+        List<RebalanceInfo> 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";

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

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<RebalanceInfo> 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<RebalanceInfo> 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 =

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
+                + '}';
+    }
+}

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<Void> cancelRebalance(@Nullable String rebalanceId) {
         throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
     }
 
+    @Override
+    public CompletableFuture<List<RebalanceInfo>> listRebalances() {
+        throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
+    }
+
     @Override
     public CompletableFuture<RegisterResult> registerProducerOffsets(
             String producerId, Map<TableBucket, Long> offsets) {