[COST] add MigrateTimeCost

app/src/main/java/org/astraea/app/web/BalancerHandler.java

@@ -192,7 +192,8 @@ private PlanExecutionProgress progress(String taskId) {
                         solution ->
                             new PlanReport(
                                 changes.apply(solution),
-                                MigrationCost.migrationCosts(contextCluster, solution.proposal())))
+                                MigrationCost.migrationCosts(
+                                    contextCluster, solution.proposal(), solution.clusterBean())))
                     .orElse(null);
     var phase = balancerConsole.taskPhase(taskId).orElseThrow();
     return new PlanExecutionProgress(

app/src/main/java/org/astraea/app/web/SensorHandler.java

@@ -32,7 +32,8 @@ public class SensorHandler implements Handler {
   private static final Set<String> DEFAULT_COSTS =
       Set.of(
           "org.astraea.common.cost.ReplicaLeaderCost",
-          "org.astraea.common.cost.NetworkIngressCost");
+          "org.astraea.common.cost.NetworkIngressCost",
+          "org.astraea.common.cost.PartitionMigrateTimeCost");
 
   SensorHandler(Sensors sensors) {
     this.sensors = sensors;

app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java

@@ -23,7 +23,7 @@
 import static org.astraea.common.balancer.BalancerConsole.TaskPhase.Searched;
 import static org.astraea.common.cost.MigrationCost.REPLICA_LEADERS_TO_ADDED;
 import static org.astraea.common.cost.MigrationCost.REPLICA_LEADERS_TO_REMOVE;
-import static org.astraea.common.cost.MigrationCost.TO_FETCH_BYTES;
+import static org.astraea.common.cost.MigrationCost.TO_FETCHED_BYTES;
 import static org.astraea.common.cost.MigrationCost.TO_SYNC_BYTES;
 
 import java.nio.charset.StandardCharsets;
@@ -322,7 +322,7 @@ void testMoveCost(String leaderLimit, String sizeLimit) {
           migrationCost -> {
             switch (migrationCost.name) {
               case TO_SYNC_BYTES:
-              case TO_FETCH_BYTES:
+              case TO_FETCHED_BYTES:
                 Assertions.assertTrue(
                     migrationCost.brokerCosts.values().stream().mapToLong(Long::intValue).sum()
                         <= DataSize.of(sizeLimit).bytes());

app/src/test/java/org/astraea/app/web/SensorHandlerTest.java

@@ -38,7 +38,7 @@ void testBeans() {
           Assertions.assertInstanceOf(
               SensorHandler.Response.class,
               defaultCostHandler.get(Channel.EMPTY).toCompletableFuture().join());
-      Assertions.assertEquals(2, defaultCostResponse.costs.size());
+      Assertions.assertEquals(3, defaultCostResponse.costs.size());
 
       var changedCostResponse =
           Assertions.assertInstanceOf(

common/src/main/java/org/astraea/common/balancer/Balancer.java

@@ -24,6 +24,7 @@
 import org.astraea.common.balancer.algorithms.GreedyBalancer;
 import org.astraea.common.balancer.algorithms.SingleStepBalancer;
 import org.astraea.common.cost.ClusterCost;
+import org.astraea.common.metrics.ClusterBean;
 
 public interface Balancer {
 
@@ -32,45 +33,12 @@ public interface Balancer {
    */
   Optional<Plan> offer(AlgorithmConfig config);
 
-  class Plan {
-    private final ClusterInfo initialClusterInfo;
-    private final ClusterCost initialClusterCost;
-
-    private final ClusterInfo proposal;
-    private final ClusterCost proposalClusterCost;
-
-    public Plan(
-        ClusterInfo initialClusterInfo,
-        ClusterCost initialClusterCost,
-        ClusterInfo proposal,
-        ClusterCost proposalClusterCost) {
-      this.initialClusterInfo = initialClusterInfo;
-      this.initialClusterCost = initialClusterCost;
-      this.proposal = proposal;
-      this.proposalClusterCost = proposalClusterCost;
-    }
-
-    public ClusterInfo initialClusterInfo() {
-      return initialClusterInfo;
-    }
-
-    /**
-     * The {@link ClusterCost} score of the original {@link ClusterInfo} when this plan is start
-     * generating.
-     */
-    public ClusterCost initialClusterCost() {
-      return initialClusterCost;
-    }
-
-    public ClusterInfo proposal() {
-      return proposal;
-    }
-
-    /** The {@link ClusterCost} score of the proposed new allocation. */
-    public ClusterCost proposalClusterCost() {
-      return proposalClusterCost;
-    }
-  }
+  record Plan(
+      ClusterBean clusterBean,
+      ClusterInfo initialClusterInfo,
+      ClusterCost initialClusterCost,
+      ClusterInfo proposal,
+      ClusterCost proposalClusterCost) {}
 
   /** The official implementation of {@link Balancer}. */
   enum Official implements EnumInfo {

common/src/main/java/org/astraea/common/balancer/BalancerProblemFormat.java

@@ -42,6 +42,7 @@ public class BalancerProblemFormat {
           "org.astraea.common.cost.RecordSizeCost",
           "org.astraea.common.cost.ReplicaNumberCost",
           "org.astraea.common.cost.ReplicaLeaderSizeCost",
+          "org.astraea.common.cost.PartitionMigrateTimeCost",
           "org.astraea.common.cost.BrokerDiskSpaceCost");
 
   public AlgorithmConfig parse() {

common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java

@@ -193,6 +193,7 @@ public Optional<Plan> offer(AlgorithmConfig config) {
                 .map(
                     newAllocation ->
                         new Plan(
+                            config.clusterBean(),
                             config.clusterInfo(),
                             initialCost,
                             newAllocation,
@@ -238,6 +239,7 @@ public Optional<Plan> offer(AlgorithmConfig config) {
                   .overflow()) {
             return Optional.of(
                 new Plan(
+                    config.clusterBean(),
                     config.clusterInfo(),
                     config.clusterCostFunction().clusterCost(config.clusterInfo(), clusterBean),
                     currentClusterInfo,

common/src/main/java/org/astraea/common/balancer/algorithms/SingleStepBalancer.java

@@ -120,6 +120,7 @@ public Optional<Plan> offer(AlgorithmConfig config) {
         .map(
             newAllocation ->
                 new Plan(
+                    config.clusterBean(),
                     config.clusterInfo(),
                     currentCost,
                     newAllocation,
@@ -139,6 +140,7 @@ public Optional<Plan> offer(AlgorithmConfig config) {
                       .overflow()) {
                 return Optional.of(
                     new Plan(
+                        config.clusterBean(),
                         config.clusterInfo(),
                         config.clusterCostFunction().clusterCost(config.clusterInfo(), clusterBean),
                         currentClusterInfo,

common/src/main/java/org/astraea/common/cost/MigrationCost.java

@@ -18,34 +18,42 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalDouble;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.astraea.common.admin.Broker;
 import org.astraea.common.admin.ClusterInfo;
 import org.astraea.common.admin.Replica;
+import org.astraea.common.metrics.ClusterBean;
+import org.astraea.common.metrics.HasBeanObject;
+import org.astraea.common.metrics.broker.HasMaxRate;
 
 public class MigrationCost {
 
   public final String name;
-
   public final Map<Integer, Long> brokerCosts;
   public static final String TO_SYNC_BYTES = "record size to sync (bytes)";
-  public static final String TO_FETCH_BYTES = "record size to fetch (bytes)";
+  public static final String TO_FETCHED_BYTES = "record size to fetched (bytes)";
   public static final String REPLICA_LEADERS_TO_ADDED = "leader number to add";
   public static final String REPLICA_LEADERS_TO_REMOVE = "leader number to remove";
   public static final String CHANGED_REPLICAS = "changed replicas";
+  public static final String PARTITION_MIGRATED_TIME = "partition migrated time";
 
-  public static List<MigrationCost> migrationCosts(ClusterInfo before, ClusterInfo after) {
+  public static List<MigrationCost> migrationCosts(
+      ClusterInfo before, ClusterInfo after, ClusterBean clusterBean) {
     var migrateInBytes = recordSizeToSync(before, after);
-    var migrateOutBytes = recordSizeToFetch(before, after);
+    var migrateOutBytes = recordSizeToFetched(before, after);
     var migrateReplicaNum = replicaNumChanged(before, after);
     var migrateInLeader = replicaLeaderToAdd(before, after);
     var migrateOutLeader = replicaLeaderToRemove(before, after);
     return List.of(
         new MigrationCost(TO_SYNC_BYTES, migrateInBytes),
-        new MigrationCost(TO_FETCH_BYTES, migrateOutBytes),
+        new MigrationCost(TO_FETCHED_BYTES, migrateOutBytes),
+        new MigrationCost(CHANGED_REPLICAS, migrateReplicaNum),
+        new MigrationCost(
+            PARTITION_MIGRATED_TIME, brokerMigrationSecond(before, after, clusterBean)),
         new MigrationCost(REPLICA_LEADERS_TO_ADDED, migrateInLeader),
         new MigrationCost(REPLICA_LEADERS_TO_REMOVE, migrateOutLeader),
         new MigrationCost(CHANGED_REPLICAS, migrateReplicaNum));
@@ -56,7 +64,7 @@ public MigrationCost(String name, Map<Integer, Long> brokerCosts) {
     this.brokerCosts = brokerCosts;
   }
 
-  static Map<Integer, Long> recordSizeToFetch(ClusterInfo before, ClusterInfo after) {
+  static Map<Integer, Long> recordSizeToFetched(ClusterInfo before, ClusterInfo after) {
     return migratedChanged(before, after, true, (ignore) -> true, Replica::size);
   }
 
@@ -76,6 +84,69 @@ static Map<Integer, Long> replicaLeaderToRemove(ClusterInfo before, ClusterInfo
     return migratedChanged(before, after, false, Replica::isLeader, ignore -> 1L);
   }
 
+  /**
+   * @param before the ClusterInfo before migrated replicas
+   * @param after the ClusterInfo after migrated replicas
+   * @param clusterBean cluster metrics
+   * @return estimated migrated time required by all brokers (seconds)
+   */
+  public static Map<Integer, Long> brokerMigrationSecond(
+      ClusterInfo before, ClusterInfo after, ClusterBean clusterBean) {
+    var brokerInRate =
+        before.brokers().stream()
+            .collect(
+                Collectors.toMap(
+                    Broker::id,
+                    nodeInfo ->
+                        brokerMaxRate(
+                            nodeInfo.id(),
+                            clusterBean,
+                            PartitionMigrateTimeCost.MaxReplicationInRateBean.class)));
+    var brokerOutRate =
+        before.brokers().stream()
+            .collect(
+                Collectors.toMap(
+                    Broker::id,
+                    nodeInfo ->
+                        brokerMaxRate(
+                            nodeInfo.id(),
+                            clusterBean,
+                            PartitionMigrateTimeCost.MaxReplicationOutRateBean.class)));
+    var brokerMigrateInSecond =
+        MigrationCost.recordSizeToSync(before, after).entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    brokerSize ->
+                        brokerSize.getValue() / brokerInRate.get(brokerSize.getKey()).orElse(0)));
+    var brokerMigrateOutSecond =
+        MigrationCost.recordSizeToFetched(before, after).entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    brokerSize ->
+                        brokerSize.getValue() / brokerOutRate.get(brokerSize.getKey()).orElse(0)));
+    return Stream.concat(before.brokers().stream(), after.brokers().stream())
+        .map(Broker::id)
+        .distinct()
+        .collect(
+            Collectors.toMap(
+                nodeId -> nodeId,
+                nodeId ->
+                    (long)
+                        Math.max(
+                            brokerMigrateInSecond.get(nodeId),
+                            brokerMigrateOutSecond.get(nodeId))));
+  }
+
+  static OptionalDouble brokerMaxRate(
+      int identity, ClusterBean clusterBean, Class<? extends HasBeanObject> statisticMetrics) {
+    return clusterBean
+        .brokerMetrics(identity, statisticMetrics)
+        .mapToDouble(b -> ((HasMaxRate) b).maxRate())
+        .max();
+  }
+
   /**
    * @param before the ClusterInfo before migrated replicas
    * @param after the ClusterInfo after migrated replicas
@@ -99,7 +170,14 @@ private static Map<Integer, Long> migratedChanged(
                 p ->
                     dest.replicas(p).stream()
                         .filter(predicate)
-                        .filter(r -> !source.replicas(p).contains(r)))
+                        .filter(
+                            r ->
+                                source.replicas(p).stream()
+                                    .noneMatch(
+                                        sourceReplica ->
+                                            sourceReplica
+                                                .topicPartitionReplica()
+                                                .equals(r.topicPartitionReplica()))))
             .map(
                 r -> {
                   if (migrateOut) return dest.replicaLeader(r.topicPartition()).orElse(r);