address jark's comments

Author: swuferhong

fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java

@@ -23,8 +23,6 @@
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.rebalance.GoalType;
-import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket;
-import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket;
 import org.apache.fluss.cluster.rebalance.ServerTag;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.cluster.AlterConfig;
@@ -70,7 +68,6 @@
 
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -509,13 +506,16 @@ ListOffsetsResult listOffsets(
      *
      * <p>If one tabletServer failed adding tag, none of the tags will take effect.
      *
+     * <p>If one tabletServer already has a serverTag, and the serverTag is same with the existing
+     * one, this operation will be ignored.
+     *
      * <ul>
      *   <li>{@link AuthorizationException} If the authenticated user doesn't have cluster
      *       permissions.
      *   <li>{@link ServerNotExistException} If the tabletServer in {@code tabletServers} does not
      *       exist.
      *   <li>{@link ServerTagAlreadyExistException} If the server tag already exists for any one of
-     *       the tabletServers.
+     *       the tabletServers, and the server tag is different from the existing one.
      * </ul>
      *
      * @param tabletServers the tabletServers we want to add server tags.
@@ -528,6 +528,8 @@ ListOffsetsResult listOffsets(
      *
      * <p>If one tabletServer failed removing tag, none of the tags will be removed.
      *
+     * <p>No exception will be thrown if the server already has no any server tag now.
+     *
      * <ul>
      *   <li>{@link AuthorizationException} If the authenticated user doesn't have cluster
      *       permissions.
@@ -563,21 +565,20 @@ ListOffsetsResult listOffsets(
      *     it.
      * @return the generated rebalance plan for all the tableBuckets which need to do rebalance.
      */
-    CompletableFuture<Map<TableBucket, RebalancePlanForBucket>> rebalance(
-            List<GoalType> priorityGoals, boolean dryRun);
+    CompletableFuture<RebalancePlan> rebalance(List<GoalType> priorityGoals, boolean dryRun);
 
     /**
-     * List the rebalance process.
+     * List the rebalance progress.
      *
      * <ul>
      *   <li>{@link AuthorizationException} If the authenticated user doesn't have cluster
      *       permissions.
      *   <li>{@link NoRebalanceInProgressException} If there are no rebalance tasks in progress.
      * </ul>
      *
-     * @return the rebalance process for all the tableBuckets doing rebalance.
+     * @return the rebalance process.
      */
-    CompletableFuture<Map<TableBucket, RebalanceResultForBucket>> listRebalanceProcess();
+    CompletableFuture<RebalanceProgress> listRebalanceProgress();
 
     /**
      * Cannel the rebalance task.

fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java

@@ -25,8 +25,6 @@
 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.RebalancePlanForBucket;
-import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket;
 import org.apache.fluss.cluster.rebalance.ServerTag;
 import org.apache.fluss.config.cluster.AlterConfig;
 import org.apache.fluss.config.cluster.ConfigEntry;
@@ -547,13 +545,13 @@ public CompletableFuture<Void> removeServerTag(
     }
 
     @Override
-    public CompletableFuture<Map<TableBucket, RebalancePlanForBucket>> rebalance(
+    public CompletableFuture<RebalancePlan> rebalance(
             List<GoalType> priorityGoals, boolean dryRun) {
         throw new UnsupportedOperationException("Support soon");
     }
 
     @Override
-    public CompletableFuture<Map<TableBucket, RebalanceResultForBucket>> listRebalanceProcess() {
+    public CompletableFuture<RebalanceProgress> listRebalanceProgress() {
         throw new UnsupportedOperationException("Support soon");
     }
 

fluss-client/src/main/java/org/apache/fluss/client/admin/RebalancePlan.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.admin;
+
+import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket;
+import org.apache.fluss.metadata.TableBucket;
+
+import java.util.Map;
+
+/**
+ * The rebalance plan.
+ *
+ * @since 0.9
+ */
+public class RebalancePlan {
+
+    private final Map<TableBucket, RebalancePlanForBucket> planForBucketMap;
+
+    public RebalancePlan(Map<TableBucket, RebalancePlanForBucket> planForBucketMap) {
+        this.planForBucketMap = planForBucketMap;
+    }
+
+    public Map<TableBucket, RebalancePlanForBucket> getPlanForBucketMap() {
+        return planForBucketMap;
+    }
+}

fluss-client/src/main/java/org/apache/fluss/client/admin/RebalanceProgress.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.admin;
+
+import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket;
+import org.apache.fluss.cluster.rebalance.RebalanceStatus;
+import org.apache.fluss.metadata.TableBucket;
+
+import java.util.Map;
+
+/**
+ * The rebalance progress.
+ *
+ * @since 0.9
+ */
+public class RebalanceProgress {
+
+    /** The rebalance status for the overall rebalance. */
+    private final RebalanceStatus rebalanceStatus;
+
+    /** The rebalance progress for the overall rebalance. Between 0.0d to 1.0d */
+    private final double progress;
+
+    /** The rebalance progress for each tabletBucket. */
+    private final Map<TableBucket, RebalanceResultForBucket> processForBucketMap;
+
+    public RebalanceProgress(
+            RebalanceStatus rebalanceStatus,
+            double progress,
+            Map<TableBucket, RebalanceResultForBucket> processForBucketMap) {
+        this.rebalanceStatus = rebalanceStatus;
+        this.progress = progress;
+        this.processForBucketMap = processForBucketMap;
+    }
+
+    public RebalanceStatus status() {
+        return rebalanceStatus;
+    }
+
+    public double progress() {
+        return progress;
+    }
+
+    public Map<TableBucket, RebalanceResultForBucket> processForBucketMap() {
+        return processForBucketMap;
+    }
+}

fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/GoalType.java

@@ -24,7 +24,7 @@
 /**
  * The type of goal to optimize.
  *
- * @since 0.8
+ * @since 0.9
  */
 @PublicEvolving
 public enum GoalType {
@@ -38,7 +38,7 @@ public enum GoalType {
      * Goal to generate leadership movement and leader replica movement tasks to ensure that the
      * number of leader replicas on each tabletServer is near balanced.
      */
-    LEADER_REPLICA_DISTRIBUTION_GOAL(1);
+    LEADER_DISTRIBUTION_GOAL(1);
 
     public final int value;
 
@@ -49,8 +49,8 @@ public enum GoalType {
     public static GoalType valueOf(int value) {
         if (value == REPLICA_DISTRIBUTION_GOAL.value) {
             return REPLICA_DISTRIBUTION_GOAL;
-        } else if (value == LEADER_REPLICA_DISTRIBUTION_GOAL.value) {
-            return LEADER_REPLICA_DISTRIBUTION_GOAL;
+        } else if (value == LEADER_DISTRIBUTION_GOAL.value) {
+            return LEADER_DISTRIBUTION_GOAL;
         } else {
             throw new IllegalArgumentException(
                     String.format(

fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalancePlanForBucket.java

@@ -26,7 +26,7 @@
 /**
  * a generated rebalance plan for a tableBucket.
  *
- * @since 0.8
+ * @since 0.9
  */
 @PublicEvolving
 public class RebalancePlanForBucket {

fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceResultForBucket.java

@@ -25,43 +25,37 @@
 /**
  * Status of rebalance process for a tabletBucket.
  *
- * @since 0.8
+ * @since 0.9
  */
 @PublicEvolving
 public class RebalanceResultForBucket {
     private final RebalancePlanForBucket rebalancePlanForBucket;
-    private RebalanceStatusForBucket rebalanceStatusForBucket;
+    private final RebalanceStatus rebalanceStatus;
 
     public RebalanceResultForBucket(
-            RebalancePlanForBucket rebalancePlanForBucket,
-            RebalanceStatusForBucket rebalanceStatusForBucket) {
+            RebalancePlanForBucket rebalancePlanForBucket, RebalanceStatus rebalanceStatus) {
         this.rebalancePlanForBucket = rebalancePlanForBucket;
-        this.rebalanceStatusForBucket = rebalanceStatusForBucket;
+        this.rebalanceStatus = rebalanceStatus;
     }
 
     public TableBucket tableBucket() {
         return rebalancePlanForBucket.getTableBucket();
     }
 
-    public RebalancePlanForBucket planForBucket() {
+    public RebalancePlanForBucket plan() {
         return rebalancePlanForBucket;
     }
 
     public List<Integer> newReplicas() {
         return rebalancePlanForBucket.getNewReplicas();
     }
 
-    public RebalanceResultForBucket setNewStatus(RebalanceStatusForBucket status) {
-        this.rebalanceStatusForBucket = status;
-        return this;
-    }
-
-    public RebalanceStatusForBucket status() {
-        return rebalanceStatusForBucket;
+    public RebalanceStatus status() {
+        return rebalanceStatus;
     }
 
     public static RebalanceResultForBucket of(
-            RebalancePlanForBucket planForBucket, RebalanceStatusForBucket status) {
+            RebalancePlanForBucket planForBucket, RebalanceStatus status) {
         return new RebalanceResultForBucket(planForBucket, status);
     }
 
@@ -71,7 +65,7 @@ public String toString() {
                 + "rebalancePlanForBucket="
                 + rebalancePlanForBucket
                 + ", rebalanceStatusForBucket="
-                + rebalanceStatusForBucket
+                + rebalanceStatus
                 + '}';
     }
 }

…/rebalance/RebalanceStatusForBucket.java …s/cluster/rebalance/RebalanceStatus.javafluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceStatusForBucket.java renamed to fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceStatus.java fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceStatusForBucket.java renamed to fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/RebalanceStatus.java

@@ -20,25 +20,29 @@
 import org.apache.fluss.annotation.PublicEvolving;
 
 /**
- * Rebalance status for single bucket.
+ * Rebalance status.
  *
- * @since 0.8
+ * @since 0.9
  */
 @PublicEvolving
-public enum RebalanceStatusForBucket {
+public enum RebalanceStatus {
     PENDING(1),
     REBALANCING(2),
     FAILED(3),
     COMPLETED(4);
 
     private final int code;
 
-    RebalanceStatusForBucket(int code) {
+    RebalanceStatus(int code) {
         this.code = code;
     }
 
-    public static RebalanceStatusForBucket of(int code) {
-        for (RebalanceStatusForBucket status : RebalanceStatusForBucket.values()) {
+    public int getCode() {
+        return code;
+    }
+
+    public static RebalanceStatus of(int code) {
+        for (RebalanceStatus status : RebalanceStatus.values()) {
             if (status.code == code) {
                 return status;
             }

fluss-common/src/main/java/org/apache/fluss/cluster/rebalance/ServerTag.java

@@ -24,11 +24,17 @@
 /**
  * The tag of tabletServer.
  *
- * @since 0.8
+ * @since 0.9
  */
 @PublicEvolving
 public enum ServerTag {
+    /**
+     * The tabletServer is permanently offline. Such as the host where the tabletServer on is
+     * upcoming decommissioning.
+     */
     PERMANENT_OFFLINE(0),
+
+    /** The tabletServer is temporarily offline. Such as the tabletServer is upcoming upgrading. */
     TEMPORARY_OFFLINE(1);
 
     public final int value;

fluss-common/src/main/java/org/apache/fluss/exception/NoRebalanceInProgressException.java

@@ -22,7 +22,7 @@
 /**
  * Thrown if there are no rebalance tasks in progress when list rebalance process.
  *
- * @since 0.8
+ * @since 0.9
  */
 @PublicEvolving
 public class NoRebalanceInProgressException extends ApiException {