[COST] add MigrateTimeCost

app/src/main/java/org/astraea/app/publisher/MetricPublisher.java

@@ -59,6 +59,7 @@ static void execute(Arguments arguments) {
                                             NodeInfo::id,
                                             node ->
                                                 MBeanClient.jndi(
+                                                    node.id(),
                                                     node.host(),
                                                     arguments.idToJmxPort().apply(node.id()))))))
             .fetchBeanDelay(arguments.period)

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

@@ -78,7 +78,9 @@ class BalancerHandler implements Handler, AutoCloseable {
                             .collect(
                                 Collectors.toUnmodifiableMap(
                                     NodeInfo::id,
-                                    b -> MBeanClient.jndi(b.host(), jmxPortMapper.apply(b.id())))));
+                                    b ->
+                                        MBeanClient.jndi(
+                                            b.id(), b.host(), jmxPortMapper.apply(b.id())))));
     this.metricStore =
         MetricStore.builder()
             .beanExpiration(Duration.ofSeconds(90))

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

@@ -46,7 +46,7 @@ public CompletionStage<Response> get(Channel channel) {
                         .map(
                             b -> {
                               try (var client =
-                                  MBeanClient.jndi(b.host(), jmxPorts.apply(b.id()))) {
+                                  MBeanClient.jndi(b.id(), b.host(), jmxPorts.apply(b.id()))) {
                                 return new NodeBean(
                                     b.host(),
                                     client.beans(builder.build()).stream()

common/src/main/java/org/astraea/common/assignor/Assignor.java

@@ -163,7 +163,8 @@ public final void configure(Map<String, ?> configs) {
                       brokers.forEach(
                           b ->
                               map.put(
-                                  b.id(), MBeanClient.jndi(b.host(), jmxPortGetter.apply(b.id()))));
+                                  b.id(),
+                                  MBeanClient.jndi(b.id(), b.host(), jmxPortGetter.apply(b.id()))));
                       // add local client to fetch consumer metrics
                       map.put(-1, MBeanClient.local());
                       return Collections.unmodifiableMap(map);

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

@@ -16,20 +16,22 @@
  */
 package org.astraea.common.cost;
 
+import static org.astraea.common.cost.PartitionMigrateTimeCost.brokerMaxRate;
+
 import java.util.List;
 import java.util.Map;
 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.ClusterBean;
 import org.astraea.common.admin.ClusterInfo;
 import org.astraea.common.admin.NodeInfo;
 import org.astraea.common.admin.Replica;
 
 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)";
@@ -69,6 +71,60 @@ static Map<Integer, Long> replicaLeaderChanged(ClusterInfo before, ClusterInfo a
     return changedReplicaNumber(before, after, Replica::isLeader);
   }
 
+  /**
+   * @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> brokerMigrationTime(
+      ClusterInfo before, ClusterInfo after, ClusterBean clusterBean) {
+    var brokerInRate =
+        brokerMaxRate(before, clusterBean, PartitionMigrateTimeCost.MaxReplicationInRateBean.class);
+    var brokerOutRate =
+        brokerMaxRate(
+            before, clusterBean, PartitionMigrateTimeCost.MaxReplicationOutRateBean.class);
+    var brokerMigrateInTime =
+        MigrationCost.recordSizeToFetch(before, after).entrySet().stream()
+            .map(
+                brokerSize ->
+                    Map.entry(
+                        brokerSize.getKey(),
+                        brokerSize.getValue()
+                            / brokerInRate
+                                .get(brokerSize.getKey())
+                                .orElseThrow(
+                                    () ->
+                                        new RuntimeException(
+                                            "No any metric for broker" + brokerSize.getKey()))))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    var brokerMigrateOutTime =
+        MigrationCost.recordSizeToSync(before, after).entrySet().stream()
+            .map(
+                brokerSize ->
+                    Map.entry(
+                        brokerSize.getKey(),
+                        brokerSize.getValue()
+                            / brokerOutRate
+                                .get(brokerSize.getKey())
+                                .orElseThrow(
+                                    () ->
+                                        new RuntimeException(
+                                            "No any metric for broker" + brokerSize.getKey()))))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    return Stream.concat(before.nodes().stream(), after.nodes().stream())
+        .distinct()
+        .map(
+            nodeInfo ->
+                Map.entry(
+                    nodeInfo.id(),
+                    (long)
+                        Math.max(
+                            brokerMigrateInTime.get(nodeInfo.id()),
+                            brokerMigrateOutTime.get(nodeInfo.id()))))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
   /**
    * @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,164 @@
+/*
+ * 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 java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalDouble;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.astraea.common.Configuration;
+import org.astraea.common.admin.ClusterBean;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.metrics.BeanObject;
+import org.astraea.common.metrics.HasBeanObject;
+import org.astraea.common.metrics.Sensor;
+import org.astraea.common.metrics.broker.HasMeter;
+import org.astraea.common.metrics.broker.HasRate;
+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";
+  public static final String MAX_MIGRATE_TIME_KEY = "max.migrated.time.limit";
+  static final Map<Integer, Sensor<Double>> maxBrokerReplicationInRate = new HashMap<>();
+  static final Map<Integer, Sensor<Double>> maxBrokerReplicationOutRate = new HashMap<>();
+  // metrics windows size
+  private final long maxMigrateTime;
+
+  public PartitionMigrateTimeCost(Configuration config) {
+    this.maxMigrateTime =
+        config.string(MAX_MIGRATE_TIME_KEY).map(Long::parseLong).orElse(Long.MAX_VALUE);
+  }
+
+  @Override
+  public Optional<MetricSensor> metricSensor() {
+    return Optional.of(
+        (client, clusterBean) -> {
+          var metrics =
+              clusterBean.all().values().stream()
+                  .flatMap(Collection::stream)
+                  .distinct()
+                  .collect(Collectors.toCollection(ArrayList::new));
+          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 maxInRateSensor =
+              maxBrokerReplicationInRate.computeIfAbsent(
+                  client.identity(),
+                  ignore ->
+                      Sensor.builder().addStat(REPLICATION_IN_RATE, Max.<Double>of()).build());
+          var maxOutRateSensor =
+              maxBrokerReplicationOutRate.computeIfAbsent(
+                  client.identity(),
+                  ignore ->
+                      Sensor.builder().addStat(REPLICATION_OUT_RATE, Max.<Double>of()).build());
+          maxInRateSensor.record(newInMetrics.oneMinuteRate());
+          maxOutRateSensor.record(newOutMetrics.oneMinuteRate());
+          var inRate = maxInRateSensor.measure(REPLICATION_IN_RATE);
+          var outRate = maxOutRateSensor.measure(REPLICATION_OUT_RATE);
+          metrics.add(
+              (MaxReplicationInRateBean)
+                  () ->
+                      new BeanObject(
+                          newInMetrics.beanObject().domainName(),
+                          newInMetrics.beanObject().properties(),
+                          Map.of(HasRate.ONE_MIN_RATE_KEY, inRate),
+                          current.toMillis()));
+          metrics.add(
+              (MaxReplicationOutRateBean)
+                  () ->
+                      new BeanObject(
+                          newOutMetrics.beanObject().domainName(),
+                          newOutMetrics.beanObject().properties(),
+                          Map.of(HasRate.ONE_MIN_RATE_KEY, outRate),
+                          current.toMillis()));
+          return metrics;
+        });
+  }
+
+  public static Map<Integer, OptionalDouble> brokerMaxRate(
+      ClusterInfo clusterInfo,
+      ClusterBean clusterBean,
+      Class<? extends HasBeanObject> statisticMetrics) {
+    return clusterInfo.brokers().stream()
+        .map(
+            broker ->
+                Map.entry(
+                    broker.id(),
+                    clusterBean.all().getOrDefault(broker.id(), List.of()).stream()
+                        .filter(x -> statisticMetrics.isAssignableFrom(x.getClass()))
+                        .mapToDouble(x -> ((HasMeter) x).oneMinuteRate())
+                        .max()))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  public Map<Integer, Double> brokerMigratedTime(
+      Map<Integer, Long> needToMigrated, Map<Integer, OptionalDouble> brokerRate) {
+    return needToMigrated.entrySet().stream()
+        .map(
+            brokerSize ->
+                Map.entry(
+                    brokerSize.getKey(),
+                    brokerSize.getValue()
+                        / brokerRate
+                            .get(brokerSize.getKey())
+                            .orElseThrow(
+                                () ->
+                                    new NoSufficientMetricsException(
+                                        this,
+                                        Duration.ofSeconds(1),
+                                        "No metric for broker" + brokerSize.getKey()))))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  @Override
+  public MoveCost moveCost(ClusterInfo before, ClusterInfo after, ClusterBean clusterBean) {
+    var brokerInRate = brokerMaxRate(before, clusterBean, MaxReplicationInRateBean.class);
+    var brokerOutRate = brokerMaxRate(before, clusterBean, MaxReplicationOutRateBean.class);
+    var needToMigrateIn = MigrationCost.recordSizeToFetch(before, after);
+    var needToMigrateOut = MigrationCost.recordSizeToSync(before, after);
+
+    var brokerMigrateInTime = brokerMigratedTime(needToMigrateIn, brokerInRate);
+    var brokerMigrateOutTime = brokerMigratedTime(needToMigrateOut, brokerOutRate);
+    var maxMigrateTime =
+        Stream.concat(before.nodes().stream(), after.nodes().stream())
+            .distinct()
+            .map(
+                nodeInfo ->
+                    Math.max(
+                        brokerMigrateInTime.get(nodeInfo.id()),
+                        brokerMigrateOutTime.get(nodeInfo.id())))
+            .max(Comparator.comparing(Function.identity()))
+            .orElse(0.0);
+    return () -> maxMigrateTime > this.maxMigrateTime;
+  }
+
+  public interface MaxReplicationInRateBean extends HasMeter {}
+
+  public interface MaxReplicationOutRateBean extends HasMeter {}
+}

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

@@ -56,7 +56,7 @@
  * }</pre>
  */
 public interface MBeanClient extends AutoCloseable {
-  static MBeanClient of(Collection<BeanObject> objs) {
+  static MBeanClient of(int identity, Collection<BeanObject> objs) {
     return new MBeanClient() {
 
       @Override
@@ -95,6 +95,11 @@ public Collection<BeanObject> beans(
                                         .matches()))
             .collect(Collectors.toUnmodifiableList());
       }
+
+      @Override
+      public int identity() {
+        return identity;
+      }
     };
   }
 
@@ -103,9 +108,10 @@ public Collection<BeanObject> beans(
    * @param port the port of jmx server
    * @return a mbean client using JNDI to lookup metrics.
    */
-  static MBeanClient jndi(String host, int port) {
+  static MBeanClient jndi(int identity, String host, int port) {
     try {
       return of(
+          identity,
           new JMXServiceURL(
               String.format(
                   "service:jmx:rmi://%s:%s/jndi/rmi://%s:%s/jmxrmi", host, port, host, port)));
@@ -114,11 +120,12 @@ static MBeanClient jndi(String host, int port) {
     }
   }
 
-  static MBeanClient of(JMXServiceURL jmxServiceURL) {
+  static MBeanClient of(int identity, JMXServiceURL jmxServiceURL) {
     return Utils.packException(
         () -> {
           var jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
           return new BasicMBeanClient(
+              identity,
               jmxConnector.getMBeanServerConnection(),
               jmxServiceURL.getHost(),
               jmxServiceURL.getPort()) {
@@ -131,7 +138,8 @@ public void close() {
   }
 
   static MBeanClient local() {
-    return new BasicMBeanClient(ManagementFactory.getPlatformMBeanServer(), Utils.hostname(), -1);
+    return new BasicMBeanClient(
+        -1, ManagementFactory.getPlatformMBeanServer(), Utils.hostname(), -1);
   }
 
   /**
@@ -168,17 +176,21 @@ default Collection<BeanObject> beans(BeanQuery beanQuery) {
    */
   Collection<BeanObject> beans(BeanQuery beanQuery, Consumer<RuntimeException> errorHandle);
 
+  int identity();
+
   @Override
   default void close() {}
 
   class BasicMBeanClient implements MBeanClient {
 
     private final MBeanServerConnection connection;
+    final int identity;
     final String host;
 
     final int port;
 
-    BasicMBeanClient(MBeanServerConnection connection, String host, int port) {
+    BasicMBeanClient(int identity, MBeanServerConnection connection, String host, int port) {
+      this.identity = identity;
       this.connection = connection;
       this.host = host;
       this.port = port;
@@ -261,6 +273,11 @@ public Collection<BeanObject> beans(
                   .collect(Collectors.toUnmodifiableList()));
     }
 
+    @Override
+    public int identity() {
+      return identity;
+    }
+
     /**
      * Returns the list of domains in which any MBean is currently registered.
      *

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

@@ -20,6 +20,10 @@
 import org.astraea.common.metrics.HasBeanObject;
 
 public interface HasRate extends HasBeanObject {
+  String ONE_MIN_RATE_KEY = "OneMinuteRate";
+  String FIVE_MINUTE_RATE = "FiveMinuteRate";
+  String FIFTEEN_MINUTE_RATE = "FifteenMinuteRate";
+
   default double meanRate() {
     return (double) beanObject().attributes().get("MeanRate");
   }

common/src/main/java/org/astraea/common/metrics/collector/MetricStore.java

@@ -229,7 +229,7 @@ private MetricStoreImpl(
                 lastSensors = sensorsSupplier.get();
                 allBeans.forEach(
                     (id, bs) -> {
-                      var client = MBeanClient.of(bs);
+                      var client = MBeanClient.of(id, bs);
                       var clusterBean = clusterBean();
                       lastSensors.forEach(
                           (sensor, errorHandler) -> {