[COST] add MigrateTimeCost

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

@@ -193,7 +193,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/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

@@ -179,6 +179,7 @@ public Optional<Plan> offer(AlgorithmConfig config) {
                 .map(
                     newAllocation ->
                         new Plan(
+                            config.clusterBean(),
                             config.clusterInfo(),
                             initialCost,
                             newAllocation,

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

@@ -104,6 +104,7 @@ public Optional<Plan> offer(AlgorithmConfig config) {
         .map(
             newAllocation ->
                 new Plan(
+                    config.clusterBean(),
                     config.clusterInfo(),
                     currentCost,
                     newAllocation,

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

@@ -18,26 +18,31 @@
 
 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.ClusterInfo;
 import org.astraea.common.admin.NodeInfo;
 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 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 migrateReplicaNum = replicaNumChanged(before, after);
@@ -46,6 +51,9 @@ public static List<MigrationCost> migrationCosts(ClusterInfo before, ClusterInfo
     return List.of(
         new MigrationCost(TO_SYNC_BYTES, migrateInBytes),
         new MigrationCost(TO_FETCH_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));
@@ -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.nodes().stream()
+            .collect(
+                Collectors.toMap(
+                    NodeInfo::id,
+                    nodeInfo ->
+                        brokerMaxRate(
+                            nodeInfo.id(),
+                            clusterBean,
+                            PartitionMigrateTimeCost.MaxReplicationInRateBean.class)));
+    var brokerOutRate =
+        before.nodes().stream()
+            .collect(
+                Collectors.toMap(
+                    NodeInfo::id,
+                    nodeInfo ->
+                        brokerMaxRate(
+                            nodeInfo.id(),
+                            clusterBean,
+                            PartitionMigrateTimeCost.MaxReplicationOutRateBean.class)));
+    var brokerMigrateInSecond =
+        MigrationCost.recordSizeToFetch(before, after).entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    brokerSize ->
+                        brokerSize.getValue() / brokerInRate.get(brokerSize.getKey()).orElse(0)));
+    var brokerMigrateOutSecond =
+        MigrationCost.recordSizeToSync(before, after).entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    brokerSize ->
+                        brokerSize.getValue() / brokerOutRate.get(brokerSize.getKey()).orElse(0)));
+    return Stream.concat(before.nodes().stream(), after.nodes().stream())
+        .map(NodeInfo::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

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

@@ -0,0 +1,109 @@
+/*
+ * 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.astraea.common.cost;
+
+import static org.astraea.common.cost.MigrationCost.brokerMaxRate;
+
+import java.time.Duration;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.astraea.common.Configuration;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.metrics.BeanObject;
+import org.astraea.common.metrics.ClusterBean;
+import org.astraea.common.metrics.Sensor;
+import org.astraea.common.metrics.broker.HasMaxRate;
+import org.astraea.common.metrics.broker.ServerMetrics;
+import org.astraea.common.metrics.collector.MetricSensor;
+import org.astraea.common.metrics.stats.Max;
+
+/** MoveCost: more max write rate change -> higher migrate cost. */
+public class PartitionMigrateTimeCost implements HasMoveCost {
+  private static final String REPLICATION_IN_RATE = "replication_in_rate";
+  private static final String REPLICATION_OUT_RATE = "replication_out_rate";
+  static final String MAX_MIGRATE_TIME_KEY = "max.migrated.time.limit";
+  public static final String STATISTICS_RATE_KEY = "statistics.rate.key";
+
+  // metrics windows size
+  private final Duration maxMigrateTime;
+
+  public PartitionMigrateTimeCost(Configuration config) {
+    this.maxMigrateTime =
+        config.duration(MAX_MIGRATE_TIME_KEY).orElse(Duration.ofSeconds(Long.MAX_VALUE));
+  }
+
+  @Override
+  public MetricSensor metricSensor() {
+    return (client, clusterBean) -> {
+      var oldInRate =
+          brokerMaxRate(
+              client.identity(),
+              clusterBean,
+              PartitionMigrateTimeCost.MaxReplicationInRateBean.class);
+      var oldOutRate =
+          brokerMaxRate(
+              client.identity(),
+              clusterBean,
+              PartitionMigrateTimeCost.MaxReplicationOutRateBean.class);
+      var newInMetrics = ServerMetrics.BrokerTopic.REPLICATION_BYTES_IN_PER_SEC.fetch(client);
+      var newOutMetrics = ServerMetrics.BrokerTopic.REPLICATION_BYTES_OUT_PER_SEC.fetch(client);
+      var current = Duration.ofMillis(System.currentTimeMillis());
+      var maxRateSensor =
+          Sensor.builder()
+              .addStats(
+                  Map.of(
+                      REPLICATION_IN_RATE, Max.<Double>of(),
+                      REPLICATION_OUT_RATE, Max.of()))
+              .build();
+      maxRateSensor.record(REPLICATION_IN_RATE, newInMetrics.oneMinuteRate());
+      maxRateSensor.record(REPLICATION_OUT_RATE, newOutMetrics.oneMinuteRate());
+      var inRate = maxRateSensor.measure(REPLICATION_IN_RATE);
+      var outRate = maxRateSensor.measure(REPLICATION_OUT_RATE);
+      return List.of(
+          (MaxReplicationInRateBean)
+              () ->
+                  new BeanObject(
+                      newInMetrics.beanObject().domainName(),
+                      newInMetrics.beanObject().properties(),
+                      Map.of(STATISTICS_RATE_KEY, Math.max(oldInRate.orElse(0), inRate)),
+                      current.toMillis()),
+          (MaxReplicationOutRateBean)
+              () ->
+                  new BeanObject(
+                      newOutMetrics.beanObject().domainName(),
+                      newOutMetrics.beanObject().properties(),
+                      Map.of(STATISTICS_RATE_KEY, Math.max(oldOutRate.orElse(0), outRate)),
+                      current.toMillis()));
+    };
+  }
+
+  @Override
+  public MoveCost moveCost(ClusterInfo before, ClusterInfo after, ClusterBean clusterBean) {
+    var brokerMigrateSecond = MigrationCost.brokerMigrationSecond(before, after, clusterBean);
+    var planMigrateSecond =
+        brokerMigrateSecond.values().stream()
+            .max(Comparator.comparing(Function.identity()))
+            .orElse(Long.MAX_VALUE);
+    return () -> planMigrateSecond > this.maxMigrateTime.getSeconds();
+  }
+
+  public interface MaxReplicationInRateBean extends HasMaxRate {}
+
+  public interface MaxReplicationOutRateBean extends HasMaxRate {}
+}

common/src/main/java/org/astraea/common/metrics/Sensor.java

@@ -30,6 +30,9 @@ static Builder<Long> builder() {
   /** Record the new get data */
   void record(V value);
 
+  /** Record the new get data for given `metricName` */
+  void record(String statName, V value);
+
   /**
    * Get the statistic by the given `metricName`.
    *
@@ -69,6 +72,11 @@ public synchronized void record(V value) {
           stats.values().forEach(stat -> stat.record(value));
         }
 
+        @Override
+        public void record(String statName, V value) {
+          stats.get(statName).record(value);
+        }
+
         @Override
         public V measure(String statName) {
           return stats.get(statName).measure();

common/src/main/java/org/astraea/common/metrics/broker/HasMaxRate.java

@@ -0,0 +1,25 @@
+/*
+ * 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.astraea.common.metrics.broker;
+
+import org.astraea.common.metrics.HasBeanObject;
+
+public interface HasMaxRate extends HasBeanObject {
+  default double maxRate() {
+    return (double) beanObject().attributes().get("statistics.rate.key");
+  }
+}

common/src/test/java/org/astraea/common/balancer/BalancerConsoleTest.java

@@ -400,7 +400,12 @@ public Optional<Plan> offer(AlgorithmConfig config) {
                   costFunction,
                   Duration.ofMillis(sampleTimeMs - (System.currentTimeMillis() - startMs)));
             return Optional.of(
-                new Plan(config.clusterInfo(), () -> 0, config.clusterInfo(), () -> 0));
+                new Plan(
+                    config.clusterBean(),
+                    config.clusterInfo(),
+                    () -> 0,
+                    config.clusterInfo(),
+                    () -> 0));
           }
         };