Repository: skiptests/astraea (java)

Issue:
`NetworkIngressCost` and `NetworkEgressCost` doesn't consider replication bandwidth
`NetworkCost` 的實作忘了把 replication 的流量也算進去

Pull request description:
Make `NetworkCost` replication bandwidth aware
Resolve #1326.

讓 `NetworkCost` 把 follower replica 造成的流量納入考量。

![image](https://user-images.githubusercontent.com/39105714/209189593-5d757795-e17e-48ea-8ee5-4fc2e1cccfd8.png)

* 黃色線是所有節點的網路卡看到的流量之平均
* 綠色線是所有節點的 BytesInPerSec + ReplicationBytesInPerSec 之平均
* 藍色線是 CostFunction 預計會看到的，所有節點的輸入流量 之平均

黃色線和綠色線之間有一個微妙約 20 MB/s 的差異，這個差異可能是來自 Protocol 的額外資料開銷，這部分應該是沒辦法直接用 JMX metrics 取得

## 測試

1. 兩個 Topic，各 100 個 Partition, Replication Factor 2，依照 Kafka 創建 topic 的邏輯擺放。
2. Partition 輸入流量速率固定
3. Partition 的流量依照 Zipfian 分配

![image](https://user-images.githubusercontent.com/39105714/209189174-809fa35c-26ee-4c86-a466-d12f032c6b9d.png)

上圖中的 `BrokerIngress` 和 `BrokerEgress` 是各節點的 `BytesInPerSec + ReplicationBytesInPerSec`

上圖的 Max-min time series 是各時間點最大流量和最小流量的差異，因為線太多、加上現在的情境複雜很難把流量都聚集成一條線，所以用他反映負載有沒有比較集中

Patch:
diff --git a/common/src/main/java/org/astraea/common/cost/NetworkCost.java b/common/src/main/java/org/astraea/common/cost/NetworkCost.java
index 2b204b2c..79cf7bda 100644
--- a/common/src/main/java/org/astraea/common/cost/NetworkCost.java
+++ b/common/src/main/java/org/astraea/common/cost/NetworkCost.java
@@ -24,6 +24,7 @@ import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import org.astraea.common.EnumInfo;
 import org.astraea.common.admin.BrokerTopic;
 import org.astraea.common.admin.ClusterBean;
 import org.astraea.common.admin.ClusterInfo;
@@ -47,13 +48,20 @@ import org.astraea.common.metrics.collector.Fetcher;
  *   <li>The network egress data rate of each partition is constant, it won't fluctuate over time.
  *   <li>No consumer or consumer group attempts to subscribe or read a subset of partitions. It must
  *       subscribe to the whole topic.
+ *   <li>This implementation assumes consumer won't fetch data from the closest replica. That is,
+ *       every consumer fetches data from the leader(which is the default behavior of Kafka). For
+ *       more detail about consumer rack awareness or how consumer can fetch data from the closest
+ *       replica, see <a href="https://cwiki.apache.org/confluence/x/go_zBQ">KIP-392<a>.
  * </ol>
  */
 public abstract class NetworkCost implements HasClusterCost {
 
   private final AtomicReference<ClusterInfo<Replica>> currentCluster = new AtomicReference<>();
+  private final BandwidthType bandwidthType;
 
-  abstract ServerMetrics.Topic useMetric();
+  NetworkCost(BandwidthType bandwidthType) {
+    this.bandwidthType = bandwidthType;
+  }
 
   void noMetricCheck(ClusterBean clusterBean) {
     var noMetricBrokers =
@@ -91,17 +99,42 @@ public abstract class NetworkCost implements HasClusterCost {
     noMetricCheck(clusterBean);
     updateCurrentCluster(clusterInfo, clusterBean, currentCluster);
 
-    var dataRate = estimateRate(currentCluster.get(), clusterBean, useMetric());
+    var ingressRate =
+        estimateRate(currentCluster.get(), clusterBean, ServerMetrics.Topic.BYTES_IN_PER_SEC);
+    var egressRate =
+        estimateRate(currentCluster.get(), clusterBean, ServerMetrics.Topic.BYTES_OUT_PER_SEC);
     var brokerRate =
         clusterInfo
             .replicaStream()
-            .filter(ReplicaInfo::isLeader)
             .filter(ReplicaInfo::isOnline)
             .collect(
                 Collectors.groupingBy(
                     replica -> clusterInfo.node(replica.nodeInfo().id()),
                     Collectors.mapping(
-                        replica -> notNull(dataRate.get(replica.topicPartition())),
+                        replica -> {
+                          var tp = replica.topicPartition();
+                          switch (bandwidthType) {
+                            case Ingress:
+                              // ingress might come from producer-send or follower-fetch.
+                              return notNull(ingressRate.get(tp));
+                            case Egress:
+                              // egress is composed of consumer-fetch and follower-fetch.
+                              // this implementation assumes no consumer rack awareness fetcher
+                              // enabled so all consumers fetch data from the leader only.
+                              return replica.isLeader()
+                                  ? notNull(egressRate.get(tp))
+                                      + notNull(ingressRate.get(tp))
+                                          // Multiply by the number of follower replicas. This
+                                          // number considers both online replicas and offline
+                                          // replicas since an offline replica is probably a
+                                          // transient behavior. So the offline state should get
+                                          // resolved in the near future, we count it in advance.
+                                          * (clusterInfo.replicas(tp).size() - 1)
+                                  : 0;
+                            default:
+                              throw new RuntimeException();
+                          }
+                        },
                         Collectors.summingDouble(x -> x))));
 
     var summary = brokerRate.values().stream().mapToDouble(x -> x).summaryStatistics();
@@ -121,7 +154,11 @@ public abstract class NetworkCost implements HasClusterCost {
     // TODO: We need a reliable way to access the actual current cluster info. To do that we need to
     //  obtain the replica info, so we intentionally sample log size but never use it.
     //  https://github.com/skiptests/astraea/pull/1240#discussion_r1044487473
-    return Fetcher.of(List.of(useMetric()::fetch, LogMetrics.Log.SIZE::fetch));
+    return Fetcher.of(
+        List.of(
+            ServerMetrics.Topic.BYTES_IN_PER_SEC::fetch,
+            ServerMetrics.Topic.BYTES_OUT_PER_SEC::fetch,
+            LogMetrics.Log.SIZE::fetch));
   }
 
   private Map<BrokerTopic, List<Replica>> mapLeaderAllocation(
@@ -188,4 +225,23 @@ public abstract class NetworkCost implements HasClusterCost {
   private <T> T fail(String reason) {
     throw new NoSufficientMetricsException(this, Duration.ofSeconds(1), reason);
   }
+
+  enum BandwidthType implements EnumInfo {
+    Ingress,
+    Egress;
+
+    static BandwidthType ofAlias(String alias) {
+      return EnumInfo.ignoreCaseEnum(BandwidthType.class, alias);
+    }
+
+    @Override
+    public String alias() {
+      return name();
+    }
+
+    @Override
+    public String toString() {
+      return alias();
+    }
+  }
 }
diff --git a/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java b/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java
index 4c9e3a69..a28fbc21 100644
--- a/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java
+++ b/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java
@@ -16,15 +16,12 @@
  */
 package org.astraea.common.cost;
 
-import org.astraea.common.metrics.broker.ServerMetrics;
-
 /**
  * A cost function to evaluate cluster load balance score in terms of message egress data rate. See
  * {@link NetworkCost} for further detail.
  */
 public class NetworkEgressCost extends NetworkCost {
-  @Override
-  ServerMetrics.Topic useMetric() {
-    return ServerMetrics.Topic.BYTES_OUT_PER_SEC;
+  public NetworkEgressCost() {
+    super(BandwidthType.Egress);
   }
 }
diff --git a/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java b/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java
index 641e33c8..3b387cea 100644
--- a/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java
+++ b/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java
@@ -16,15 +16,12 @@
  */
 package org.astraea.common.cost;
 
-import org.astraea.common.metrics.broker.ServerMetrics;
-
 /**
  * A cost function to evaluate cluster load balance score in terms of message ingress data rate. See
  * {@link NetworkCost} for further detail.
  */
 public class NetworkIngressCost extends NetworkCost {
-  @Override
-  ServerMetrics.Topic useMetric() {
-    return ServerMetrics.Topic.BYTES_IN_PER_SEC;
+  public NetworkIngressCost() {
+    super(BandwidthType.Ingress);
   }
 }
diff --git a/common/src/main/java/org/astraea/common/metrics/stats/Avg.java b/common/src/main/java/org/astraea/common/metrics/stats/Avg.java
index fee95be2..832eb13d 100644
--- a/common/src/main/java/org/astraea/common/metrics/stats/Avg.java
+++ b/common/src/main/java/org/astraea/common/metrics/stats/Avg.java
@@ -38,40 +38,6 @@ public class Avg {
     };
   }
 
-  /**
-   * Calculate the difference of the latest two ranged data.
-   *
-   * @param period Set the interval time for obtaining indicators. If multiple values are obtained
-   *     within the duration, it will be regarded as one
-   */
-  public static Stat<Double> rateByTime(Duration period) {
-    return new Stat<>() {
-      private double accumulate = 0.0;
-
-      private long count = 0;
-
-      private final Debounce<Double> debounce = Debounce.of(period);
-      ;
-
-      @Override
-      public synchronized void record(Double value) {
-        long current = System.currentTimeMillis();
-        debounce
-            .record(value, current)
-            .ifPresent(
-                debouncedValue -> {
-                  accumulate += debouncedValue;
-                  ++count;
-                });
-      }
-
-      @Override
-      public synchronized Double measure() {
-        return accumulate / count;
-      }
-    };
-  }
-
   public static Stat<Double> expWeightByTime(Duration period) {
     return expWeightByTime(period, 0.5);
   }
diff --git a/common/src/main/java/org/astraea/common/metrics/stats/Rate.java b/common/src/main/java/org/astraea/common/metrics/stats/Rate.java
index d171b09f..37a5a6f3 100644
--- a/common/src/main/java/org/astraea/common/metrics/stats/Rate.java
+++ b/common/src/main/java/org/astraea/common/metrics/stats/Rate.java
@@ -16,32 +16,30 @@
  */
 package org.astraea.common.metrics.stats;
 
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.time.Duration;
+import org.astraea.common.DataSize;
 
 /** By contrast to {@link Avg}, {@link Rate} measure the value by time instead of "count". */
-public interface Rate extends Stat<Long> {
+public interface Rate<T> extends Stat<T> {
   /**
-   * @param unit of rate
-   * @return sum of recorded values / (time of last record - start time)
+   * @return sum of recorded size / (time of last record - start time). Noted that the unit is
+   *     second
    */
-  static Rate of(TimeUnit unit) {
-    return new Rate() {
+  static Rate<DataSize> sizeRate() {
+    return new Rate<>() {
       private final long start = System.currentTimeMillis();
-      private volatile long last = System.currentTimeMillis();
-      private final AtomicLong sum = new AtomicLong();
+      private DataSize size = DataSize.Byte.of(0);
 
       @Override
-      public void record(Long value) {
-        sum.addAndGet(value);
-        last = System.currentTimeMillis();
+      public synchronized void record(DataSize value) {
+        size = size.add(value);
       }
 
       @Override
-      public Long measure() {
-        var diff = last - start;
-        if (diff <= 0) return 0L;
-        return (long) (sum.doubleValue() / unit.convert(diff, TimeUnit.MILLISECONDS));
+      public synchronized DataSize measure() {
+        var diff = System.currentTimeMillis() - start;
+        if (diff <= 0) return DataSize.Byte.of(0);
+        return size.dataRate(Duration.ofMillis(diff)).dataSize();
       }
     };
   }
diff --git a/common/src/main/java/org/astraea/common/metrics/stats/RateByTime.java b/common/src/main/java/org/astraea/common/metrics/stats/RateByTime.java
deleted file mode 100644
index 00b4c524..00000000
--- a/common/src/main/java/org/astraea/common/metrics/stats/RateByTime.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.stats;
-
-import java.time.Duration;
-
-/** Calculate the difference of the latest two ranged data. */
-public class RateByTime implements Stat<Double> {
-
-  private final double[] oldValue = new double[2];
-
-  private final Debounce<Double> debounce;
-
-  /**
-   * @param period Set the interval time for obtaining indicators. If multiple values are obtained
-   *     within the duration, it will be regarded as one
-   */
-  public RateByTime(Duration period) {
-    this.debounce = Debounce.of(period);
-  }
-
-  @Override
-  public synchronized void record(Double value) {
-    long current = System.currentTimeMillis();
-    // Update when a new record occurred
-    debounce
-        .record(value, current)
-        .ifPresent(
-            debouncedValue -> {
-              oldValue[0] = oldValue[1];
-              oldValue[1] = debouncedValue;
-            });
-  }
-
-  @Override
-  public synchronized Double measure() {
-    return oldValue[1] - oldValue[0];
-  }
-}
diff --git a/connector/src/main/java/org/astraea/connector/perf/PerfSource.java b/connector/src/main/java/org/astraea/connector/perf/PerfSource.java
index d9cf9dc5..dd843065 100644
--- a/connector/src/main/java/org/astraea/connector/perf/PerfSource.java
+++ b/connector/src/main/java/org/astraea/connector/perf/PerfSource.java
@@ -23,7 +23,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -34,7 +33,6 @@ import org.astraea.common.DataUnit;
 import org.astraea.common.DistributionType;
 import org.astraea.common.Utils;
 import org.astraea.common.metrics.stats.Rate;
-import org.astraea.common.producer.Metadata;
 import org.astraea.common.producer.Record;
 import org.astraea.connector.Definition;
 import org.astraea.connector.MetadataStorage;
@@ -169,9 +167,7 @@ public class PerfSource extends SourceConnector {
 
     Set<Integer> specifyPartitions = Set.of();
 
-    final Rate rate = Rate.of(TimeUnit.SECONDS);
-
-    long last = System.currentTimeMillis();
+    final Rate<DataSize> sizeRate = Rate.sizeRate();
 
     @Override
     protected void init(Configuration configuration, MetadataStorage storage) {
@@ -235,35 +231,41 @@ public class PerfSource extends SourceConnector {
           });
     }
 
-    @Override
-    protected Collection<Record<byte[], byte[]>> take() {
-      if (rate.measure() >= throughput.bytes()) return List.of();
-      try {
-        if (specifyPartitions.isEmpty())
-          return topics.stream()
-              .map(t -> Record.builder().topic(t).key(key()).value(value()).build())
-              .collect(Collectors.toList());
+    private Collection<Record<byte[], byte[]>> records() {
+      if (specifyPartitions.isEmpty())
         return topics.stream()
-            .flatMap(
-                t ->
-                    specifyPartitions.stream()
-                        .map(
-                            p ->
-                                Record.builder()
-                                    .topic(t)
-                                    .partition(p)
-                                    .key(key())
-                                    .value(value())
-                                    .build()))
+            .map(t -> Record.builder().topic(t).key(key()).value(value()).build())
             .collect(Collectors.toList());
-      } finally {
-        last = System.currentTimeMillis();
-      }
+      return topics.stream()
+          .flatMap(
+              t ->
+                  specifyPartitions.stream()
+                      .map(
+                          p ->
+                              Record.builder()
+                                  .topic(t)
+                                  .partition(p)
+                                  .key(key())
+                                  .value(value())
+                                  .build()))
+          .collect(Collectors.toList());
     }
 
     @Override
-    protected void commit(Metadata metadata) {
-      rate.record((long) (metadata.serializedValueSize() + metadata.serializedKeySize()));
+    protected Collection<Record<byte[], byte[]>> take() {
+      var size = sizeRate.measure();
+      if (size.greaterThan(throughput)) return List.of();
+      var records = records();
+      records.forEach(r -> sizeRate.record(DataSize.Byte.of(estimatedSize(r))));
+      return records;
+    }
+
+    private static int estimatedSize(Record<byte[], byte[]> record) {
+      return (record.key() == null ? 0 : record.key().length)
+          + (record.value() == null ? 0 : record.value().length)
+          + record.headers().stream()
+              .mapToInt(h -> h.key().length() + (h.value() == null ? 0 : record.value().length))
+              .sum();
     }
   }
 }


Test patch:
diff --git a/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java b/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java
index a050dc50..4d5a78ed 100644
--- a/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java
+++ b/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java
@@ -24,6 +24,7 @@ import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -92,10 +93,15 @@ class NetworkCostTest {
                 1,
                 List.of(
                     logSize(TopicPartition.of("Pipeline", 0), -1),
-                    bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Pipeline", 0))));
+                    bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Pipeline", 0),
+                    bandwidth(ServerMetrics.Topic.BYTES_OUT_PER_SEC, "Pipeline", 0))));
     var noLog =
         ClusterBean.of(
-            Map.of(1, List.of(bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Pipeline", 0))));
+            Map.of(
+                1,
+                List.of(
+                    bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Pipeline", 0),
+                    bandwidth(ServerMetrics.Topic.BYTES_OUT_PER_SEC, "Pipeline", 0))));
 
     Assertions.assertThrows(
         NoSufficientMetricsException.class, () -> cost.clusterCost(initial, ClusterBean.EMPTY));
@@ -120,7 +126,7 @@ class NetworkCostTest {
   }
 
   void testEstimatedRate(ServerMetrics.Topic metric) {
-    var t = new HandWrittenTestCase(metric);
+    var t = new HandWrittenTestCase();
     new NetworkIngressCost()
         .estimateRate(t.clusterInfo(), t.clusterBean(), metric)
         .forEach(
@@ -133,20 +139,17 @@ class NetworkCostTest {
   }
 
   static Stream<Arguments> testcases() {
-    var in = ServerMetrics.Topic.BYTES_IN_PER_SEC;
-    var out = ServerMetrics.Topic.BYTES_OUT_PER_SEC;
-
     return Stream.of(
-        Arguments.of(ingressCost(), new HandWrittenTestCase(in)),
-        Arguments.of(ingressCost(), new LargeTestCase(in, 2, 100, 0xfeedbabe)),
-        Arguments.of(ingressCost(), new LargeTestCase(in, 3, 100, 0xabcdef01)),
-        Arguments.of(ingressCost(), new LargeTestCase(in, 10, 200, 0xcafebabe)),
-        Arguments.of(ingressCost(), new LargeTestCase(in, 15, 300, 0xfee1dead)),
-        Arguments.of(egressCost(), new HandWrittenTestCase(out)),
-        Arguments.of(egressCost(), new LargeTestCase(out, 5, 100, 0xfa11fa11)),
-        Arguments.of(egressCost(), new LargeTestCase(out, 6, 100, 0xf001f001)),
-        Arguments.of(egressCost(), new LargeTestCase(out, 8, 200, 0xba1aba1a)),
-        Arguments.of(egressCost(), new LargeTestCase(out, 14, 300, 0xdd0000bb)));
+        Arguments.of(ingressCost(), new HandWrittenTestCase()),
+        Arguments.of(ingressCost(), new LargeTestCase(2, 100, 0xfeedbabe)),
+        Arguments.of(ingressCost(), new LargeTestCase(3, 100, 0xabcdef01)),
+        Arguments.of(ingressCost(), new LargeTestCase(10, 200, 0xcafebabe)),
+        Arguments.of(ingressCost(), new LargeTestCase(15, 300, 0xfee1dead)),
+        Arguments.of(egressCost(), new HandWrittenTestCase()),
+        Arguments.of(egressCost(), new LargeTestCase(5, 100, 0xfa11fa11)),
+        Arguments.of(egressCost(), new LargeTestCase(6, 100, 0xf001f001)),
+        Arguments.of(egressCost(), new LargeTestCase(8, 200, 0xba1aba1a)),
+        Arguments.of(egressCost(), new LargeTestCase(14, 300, 0xdd0000bb)));
   }
 
   @ParameterizedTest
@@ -171,13 +174,7 @@ class NetworkCostTest {
 
   @Test
   void testCompositeOptimization() {
-    var testCase =
-        new LargeTestCase(
-            ServerMetrics.Topic.BYTES_IN_PER_SEC,
-            ServerMetrics.Topic.BYTES_OUT_PER_SEC,
-            10,
-            300,
-            0xfeed);
+    var testCase = new LargeTestCase(10, 300, 0xfeed);
     var costFunction =
         HasClusterCost.of(
             Map.of(
@@ -186,6 +183,81 @@ class NetworkCostTest {
     testOptimization(costFunction, testCase);
   }
 
+  @Test
+  void testReplicationAware() {
+    var base =
+        ClusterInfoBuilder.builder()
+            .addNode(Set.of(1, 2, 3))
+            .addFolders(Map.of(1, Set.of("/folder0", "/folder1")))
+            .addFolders(Map.of(2, Set.of("/folder0", "/folder1")))
+            .addFolders(Map.of(3, Set.of("/folder0", "/folder1")))
+            .build();
+    var iter1 = List.of(1, 2).iterator();
+    var iter2 = List.of(true, false).iterator();
+    var iter3 = List.of(true, false).iterator();
+    var iter4 = List.of(3, 1).iterator();
+    var iter5 = List.of(true, false).iterator();
+    var iter6 = List.of(true, false).iterator();
+    var cluster =
+        ClusterInfoBuilder.builder(base)
+            .addTopic(
+                "Rain",
+                1,
+                (short) 2,
+                (r) ->
+                    Replica.builder(r)
+                        .nodeInfo(base.node(iter1.next()))
+                        .isPreferredLeader(iter2.next())
+                        .isLeader(iter3.next())
+                        .size(1)
+                        .build())
+            .addTopic(
+                "Drop",
+                1,
+                (short) 2,
+                (r) ->
+                    Replica.builder(r)
+                        .nodeInfo(base.node(iter4.next()))
+                        .isPreferredLeader(iter5.next())
+                        .isLeader(iter6.next())
+                        .size(1)
+                        .build())
+            .build();
+    var beans =
+        ClusterBean.of(
+            Map.of(
+                1,
+                    List.of(
+                        bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Rain", 100),
+                        bandwidth(ServerMetrics.Topic.BYTES_OUT_PER_SEC, "Rain", 300)),
+                2, List.of(noise(5566)),
+                3,
+                    List.of(
+                        bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Drop", 80),
+                        bandwidth(ServerMetrics.Topic.BYTES_OUT_PER_SEC, "Drop", 800))));
+
+    double error = 1e-9;
+    Function<Double, Predicate<Double>> around =
+        (expected) -> (value) -> (expected + error) >= value && value >= (expected - error);
+
+    double expectedIngress1 = 100 + 80;
+    double expectedIngress2 = 100;
+    double expectedIngress3 = 80;
+    double expectedEgress1 = 300 + 100;
+    double expectedEgress2 = 0;
+    double expectedEgress3 = 800 + 80 * 2;
+    double expectedIngressScore = (expectedIngress1 - expectedIngress3) / expectedIngress1;
+    double expectedEgressScore = (expectedEgress3 - expectedEgress2) / expectedEgress3;
+    double ingressScore = ingressCost().clusterCost(cluster, beans).value();
+    double egressScore = egressCost().clusterCost(cluster, beans).value();
+    Assertions.assertTrue(
+        around.apply(expectedIngressScore).test(ingressScore),
+        "Ingress score should be " + expectedIngressScore + " but it is " + ingressScore);
+    Assertions.assertTrue(
+        around.apply(expectedEgressScore).test(egressScore),
+        "Egress score should be " + expectedEgressScore + " but it is " + egressScore);
+  }
+
   @Test
   void testZeroBandwidth() {
     // if a partition has no produce/fetch occur, it won't have a BytesInPerSec or BytesOutPerSec
@@ -246,21 +318,51 @@ class NetworkCostTest {
    * </ul>
    */
   private static class HandWrittenTestCase implements TestCase {
-    private HandWrittenTestCase(ServerMetrics.Topic metric) {
+    private HandWrittenTestCase() {
       this.clusterBean =
           ClusterBean.of(
               Map.of(
                   1,
                   expectedBrokerTopicBandwidth.get(1).entrySet().stream()
-                      .map(e -> bandwidth(metric, e.getKey(), e.getValue()))
+                      .flatMap(
+                          e ->
+                              Stream.of(
+                                  bandwidth(
+                                      ServerMetrics.Topic.BYTES_IN_PER_SEC,
+                                      e.getKey(),
+                                      e.getValue()),
+                                  bandwidth(
+                                      ServerMetrics.Topic.BYTES_OUT_PER_SEC,
+                                      e.getKey(),
+                                      e.getValue())))
                       .collect(Collectors.toUnmodifiableList()),
                   2,
                   expectedBrokerTopicBandwidth.get(2).entrySet().stream()
-                      .map(e -> bandwidth(metric, e.getKey(), e.getValue()))
+                      .flatMap(
+                          e ->
+                              Stream.of(
+                                  bandwidth(
+                                      ServerMetrics.Topic.BYTES_IN_PER_SEC,
+                                      e.getKey(),
+                                      e.getValue()),
+                                  bandwidth(
+                                      ServerMetrics.Topic.BYTES_OUT_PER_SEC,
+                                      e.getKey(),
+                                      e.getValue())))
                       .collect(Collectors.toUnmodifiableList()),
                   3,
                   expectedBrokerTopicBandwidth.get(3).entrySet().stream()
-                      .map(e -> bandwidth(metric, e.getKey(), e.getValue()))
+                      .flatMap(
+                          e ->
+                              Stream.of(
+                                  bandwidth(
+                                      ServerMetrics.Topic.BYTES_IN_PER_SEC,
+                                      e.getKey(),
+                                      e.getValue()),
+                                  bandwidth(
+                                      ServerMetrics.Topic.BYTES_OUT_PER_SEC,
+                                      e.getKey(),
+                                      e.getValue())))
                       .collect(Collectors.toUnmodifiableList())));
     }
 
@@ -334,16 +436,7 @@ class NetworkCostTest {
     private final Map<TopicPartition, Long> rate;
     private final Supplier<DataRate> dataRateSupplier;
 
-    public LargeTestCase(ServerMetrics.Topic metric, int brokers, int partitions, int seed) {
-      this(metric, null, brokers, partitions, seed);
-    }
-
-    public LargeTestCase(
-        ServerMetrics.Topic metric0,
-        ServerMetrics.Topic metric1,
-        int brokers,
-        int partitions,
-        int seed) {
+    public LargeTestCase(int brokers, int partitions, int seed) {
       var random = new Random(seed);
       this.dataRateSupplier =
           () -> {
@@ -408,7 +501,7 @@ class NetworkCostTest {
                                   noise(random.nextInt()),
                                   noise(random.nextInt()),
                                   bandwidth(
-                                      metric0,
+                                      ServerMetrics.Topic.BYTES_IN_PER_SEC,
                                       "Pipeline",
                                       clusterInfo
                                           .replicaStream()
@@ -420,7 +513,7 @@ class NetworkCostTest {
                                   noise(random.nextInt()),
                                   noise(random.nextInt()),
                                   bandwidth(
-                                      metric1,
+                                      ServerMetrics.Topic.BYTES_OUT_PER_SEC,
                                       "Pipeline",
                                       clusterInfo
                                           .replicaStream()
diff --git a/common/src/test/java/org/astraea/common/metrics/stats/AvgTest.java b/common/src/test/java/org/astraea/common/metrics/stats/AvgTest.java
index 80b7bfe1..f85dd096 100644
--- a/common/src/test/java/org/astraea/common/metrics/stats/AvgTest.java
+++ b/common/src/test/java/org/astraea/common/metrics/stats/AvgTest.java
@@ -37,21 +37,6 @@ public class AvgTest {
     Assertions.assertThrows(RuntimeException.class, stat::measure);
   }
 
-  @Test
-  void testRateByTime() throws InterruptedException {
-    var rateByTime = Avg.rateByTime(Duration.ofSeconds(1));
-    rateByTime.record(10.0);
-    rateByTime.record(10.0);
-    Thread.sleep(1000);
-    rateByTime.record(50.0);
-
-    Assertions.assertEquals((10 + 50) / 2.0, rateByTime.measure());
-
-    rateByTime.record(50.0);
-
-    Assertions.assertEquals((10 + 50) / 2.0, rateByTime.measure());
-  }
-
   @Test
   void testExpWeightByTime() throws InterruptedException {
     var rateByTime = Avg.expWeightByTime(Duration.ofSeconds(1), 0.5);
diff --git a/common/src/test/java/org/astraea/common/metrics/stats/RateByTimeTest.java b/common/src/test/java/org/astraea/common/metrics/stats/RateTest.java
similarity index 70%
rename from common/src/test/java/org/astraea/common/metrics/stats/RateByTimeTest.java
rename to common/src/test/java/org/astraea/common/metrics/stats/RateTest.java
index 13eaa8e1..c7f60b03 100644
--- a/common/src/test/java/org/astraea/common/metrics/stats/RateByTimeTest.java
+++ b/common/src/test/java/org/astraea/common/metrics/stats/RateTest.java
@@ -17,22 +17,18 @@
 package org.astraea.common.metrics.stats;
 
 import java.time.Duration;
+import org.astraea.common.DataSize;
+import org.astraea.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class RateByTimeTest {
-  @Test
-  void testMeasure() throws InterruptedException {
-    var rateByTime = new RateByTime(Duration.ofSeconds(1));
-    rateByTime.record(10.0);
-    rateByTime.record(10.0);
-    Thread.sleep(1000);
-    rateByTime.record(50.0);
-
-    Assertions.assertEquals(40.0, rateByTime.measure());
+public class RateTest {
 
-    rateByTime.record(50.0);
-
-    Assertions.assertEquals(40.0, rateByTime.measure());
+  @Test
+  void testMeasure() {
+    var rate = Rate.sizeRate();
+    rate.record(DataSize.Byte.of(100L));
+    Utils.sleep(Duration.ofSeconds(1));
+    Assertions.assertTrue(rate.measure().bytes() < 100);
   }
 }
diff --git a/connector/src/test/java/org/astraea/connector/perf/PerfSourceTest.java b/connector/src/test/java/org/astraea/connector/perf/PerfSourceTest.java
index e718e8ac..a5c5de0a 100644
--- a/connector/src/test/java/org/astraea/connector/perf/PerfSourceTest.java
+++ b/connector/src/test/java/org/astraea/connector/perf/PerfSourceTest.java
@@ -113,16 +113,17 @@ public class PerfSourceTest extends RequireSingleWorkerCluster {
                 ConnectorConfigs.TOPICS_KEY,
                 topicName,
                 PerfSource.THROUGHPUT_DEF.name(),
-                "0Byte"))
+                "1Byte"))
         .toCompletableFuture()
         .join();
 
     Utils.sleep(Duration.ofSeconds(3));
 
     try (var admin = Admin.of(bootstrapServers())) {
-      // no records due to throughput is defined by 0Byte
-      Assertions.assertFalse(
-          admin.topicNames(false).toCompletableFuture().join().contains(topicName));
+      var offsets =
+          admin.latestOffsets(Set.of(TopicPartition.of(topicName, 0))).toCompletableFuture().join();
+      Assertions.assertEquals(1, offsets.size());
+      Assertions.assertEquals(1, offsets.get(TopicPartition.of(topicName, 0)));
     }
   }