Repository: skiptests/astraea (java)

Issue:
Proposing a way to evaluate the partition egress loading of brokers
這個 Issue 想要提出一個：衡量 Partition 負載分佈和 Broker 節點，在網路輸出上的負載量關係。
即假設有個 Broker b 他有 p0, p1, p2 這三個 Partition，則這三個 Partition 會給 b 帶來多少網路輸出上的開銷。
有了這個計算方法，我們就能實作出讓節點的輸出負載平衡的 Cost Function

這個問題需要特別討論的原因是：

1. Kafka 內提供的 metrics 很難去反映這個負載量和 Partition 的關係，這對 Balancer 產生計劃打分帶來一個問題。
2. 需要一個精準的方法來衡量這個負載量，如果衡量不準，那我們產生出來的負載優化計劃對叢集是否有幫助會是個問號。

以下會提出幾種我想到的衡量方法，每個版本有自己的特色，會再特別做討論

## Version 0 (這個版本實務上做不到)

![image](https://user-images.githubusercontent.com/39105714/204347552-f350ef54-a14d-4a4a-9309-7684fcce3726.png)

* 最理想的計算方法
* 想不到從 kafka metrics 取得 `tp 被撈取的速度` 的方法，所以這個**計算方法不可行**

## Version 1 (這個版本實務上做不到)

![image](https://user-images.githubusercontent.com/39105714/204347782-1585b934-d8aa-41a3-b061-25f204a2c2af.png)

* 這個方法引入 ConsumerGroup的角色，從 ConsumerGroup 去進一步揣測每個 Partition 的流量
  * 因為只顧慮到 ConsumerGroup，所以 **非 ConsumerGroup 的撈取是黑數**
* 想不到如何從 broker 取得 `group_fetch_speed` 的方法，所以這個**計算方法不可行**

## Version 2

![image](https://user-images.githubusercontent.com/39105714/204348222-a401e79d-3d4f-4014-86bd-174167182d55.png)

這個方法的直覺是，所有 ConsumerGroup 都在撈取最新的資料，所以資料的輸入理論上會等同**資料的輸出速度**(理想情況)

* `partition_ingress` 可以從 `log` 增長速度來取得
* 如果 ConsumerGroup 跳著撈資料，則這個方法不可行
  * ConsumerGroup 的一大目的就是要確保每個資料都被處理，所以我認為不常發生 ConsumerGroup 跳著撈資料的情況，由於觸發頻率不高，且這個 Cost 是要反映叢集長遠的效能表現，所以這個問題能被忽略。
* 如果 ConsumerGroup 以 below-ingress-speed 撈資料，則這個方法判斷的實際流量會失真
  * 我覺得這個情況會發生在下面幾個情況會發生
    1. ConsumerGroup 不足以應付負載量，這個其實不是 Balancer 的問題，應該請維護者擴充 Consumer 數量，讓 Balancer 處理這個問題的價值也不高，因為這個問題背後隱含的另外一個危險是資料來不及處理，這個問題比負載平衡還嚴重，應該先處理好他，如果沒處理好他，沒辦法談後續的負載平衡，如果這個問題處理好了，那 ConsumerGroup 將以 full-speed 撈資料，這個問題也就不存在了。
    4. 這個 Partition 有 Brust Load 的情況，也就是平常沒什麼資料，但資料一來就是大批的塞進來，通常這個情況，維護者也不在乎資料有沒有及時處理完，有處理完就好，且他有很大的信心一定會處理完，如果他沒信心的話應該添加 Consumer 的數量，因為這個 Brust Load 從進來到處理完都不是長遠的行為（我們在意的流量問題），所以不在這個 Cost 要處理的範疇內。
* 如果 ConsumerGroup 以 over-ingress-speed 撈資料，則這個方法判斷的實際流量會失真
  * 我覺得這個情況會在下面幾個情況會發生，多半和資料堆積有關：
    1. ConsumerGroup 因為某些原因暫停，過一下子繼續跑：這類的事件不常發生且影響小，Cost 要反映叢集長遠的效能表現，所以這個問題可以被忽略。
    2. ConsumerGroup 因為某些原因暫停，過了很久才繼續跑：這類的事件不常發生，如果發生了，只要叢集有資源就能慢慢追上，就 Cost 要反映長遠的效能表現這點，我覺得這個問題可以被忽略。
* 因為上面的討論都以長遠的表現來反駁那些問題，所以這個做法在**長期**看來，我們的負載計算方法是正確的。但如果從**短期**來看，我們一定會看到負載不平衡的現象，但那些應該都一段時間的事情，這些很隨機的事件很難從 Cost Function 的角度去處理，所以只能這樣...
* 非 ConsumerGroup 的撈取依然是黑數

## Version 3

![image](https://user-images.githubusercontent.com/39105714/204351105-2206b1a9-dc2c-421a-af76-bfddb6c5f5b1.png)

這個是 Ver2 的延伸，這個版本要嘗試解決非 ConsumerGroup 撈取是黑數的問題，但方法是臆測，基本上是從 `BrokerTopicMetrics#BytesOutPerSec(topic)` 這個 metrics 中萃取出非 ConsumerGroup 的那部分，然後從中以各個 Partition 的輸入資料量來刮分這個部分，因為真的不知道非 ConsumerGroup 對各個 partition 的撈取狀況，我們只能假設這個撈取的量和其輸入速度大小有關，**但這個直覺有可能會不正確**。

* `BrokerEgressVer2` 是上面 Ver2 的公式
* `BrokerBytesOutNonGroup(b, tp)` == `BrokerTopicMetrics#BytesOutPerSec(topic)` - `consumer group 的那部分撈取`，後面 consumer gorup 那部分的撈取是已知數，我們 Ver2 中有求出來。
* 最後那個分數是對應 partition 在輸入速度方面所佔據的比例

其實非 ConsumerGroup 的撈取有可能代表他不在乎有沒有 catch 上進度，做 random access/ old data sequential access 是一種可能，這邊的衡量方法是從目前的輸入速度下去猜， random access 的量和 log size 的關聯比較高，或許應該從整體的 log size 去做從揣測，而非從輸入速度去猜。


Pull request:
Implementation of `NetworkIngressCost` and `NetworkEgressCost`
Resolve #1181, #1148.

根據 #1181 最後的討論實作出的節點網路IO Cost Function，內部透過 `ServerMetrics.Topic.BYTES_IN_PER_SEC`  和 `ServerMetrics.Topic.BYTES_OUT_PER_SEC` 得到 Topic 在特定節點的輸入輸出流量，後面透過每個 Replica 的大小去均分這個輸入輸出流量。這個機制在每個 Topic/Partition 流入的資料量固定或變動不大時效果不錯。

## Experiment Result

* Skewed Load 情境，每個 Topic/Partition 輸入吞吐量不太一樣
* 情境中有兩個 Topic，其中一個 Topic 的 Consumer Fanout 為 2

Balancer Config

```json
{
  "timeout": "60s",
  "balancer": "org.astraea.common.balancer.algorithms.GreedyBalancer",
  "balancer-config": {
    "shuffle.plan.generator.min.step": "1",
    "shuffle.plan.generator.max.step": "5"
  },
  "costWeights": [
    {
      "cost": "org.astraea.common.cost.NetworkIngressCost",
      "weight": 1
    },
    {
      "cost": "org.astraea.common.cost.NetworkEgressCost",
      "weight": 1
    }
  ]
}
```

![image](https://user-images.githubusercontent.com/39105714/206643460-808c8370-e97e-42ca-9c47-12357a4c2559.png)


Patch:
diff --git a/common/src/main/java/org/astraea/common/connector/Builder.java b/common/src/main/java/org/astraea/common/connector/Builder.java
index 2b946565..d424a22e 100644
--- a/common/src/main/java/org/astraea/common/connector/Builder.java
+++ b/common/src/main/java/org/astraea/common/connector/Builder.java
@@ -16,9 +16,11 @@
  */
 package org.astraea.common.connector;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -216,20 +218,44 @@ public class Builder {
       }
 
       @Override
-      public CompletionStage<Validation> validate(String name, Map<String, String> configs) {
+      public CompletionStage<Validation> validate(String className, Map<String, String> configs) {
+        var finalConfigs = configs;
+        if (!finalConfigs.containsKey(ConnectorConfigs.CONNECTOR_CLASS_KEY)) {
+          finalConfigs = new HashMap<>(configs);
+          finalConfigs.put(ConnectorConfigs.CONNECTOR_CLASS_KEY, className);
+        }
         return httpExecutor
             .put(
-                getURL("/connector-plugins") + "/" + name + "/config/validate",
-                configs,
+                getURL("/connector-plugins") + "/" + className + "/config/validate",
+                finalConfigs,
                 TypeRef.of(Validation.class))
             .thenApply(Response::body);
       }
 
       @Override
-      public CompletionStage<Set<PluginInfo>> plugins() {
+      public CompletionStage<List<PluginInfo>> plugins() {
         return httpExecutor
-            .get(getURL("/connector-plugins"), TypeRef.set(PluginInfo.class))
-            .thenApply(Response::body);
+            .get(getURL("/connector-plugins"), TypeRef.set(KafkaPluginInfo.class))
+            .thenApply(Response::body)
+            .thenCompose(
+                plugins ->
+                    FutureUtils.sequence(
+                        plugins.stream()
+                            .map(
+                                p ->
+                                    // "Must configure one of topics or topics.regex" by kafka
+                                    validate(
+                                            p.className,
+                                            Map.of(ConnectorConfigs.TOPICS_KEY, "random"))
+                                        .thenApply(
+                                            ds ->
+                                                new PluginInfo(
+                                                    p.className,
+                                                    ds.configs().stream()
+                                                        .map(Config::definition)
+                                                        .collect(Collectors.toList())))
+                                        .toCompletableFuture())
+                            .collect(Collectors.toList())));
       }
 
       private String getURL(String path) {
@@ -275,4 +301,9 @@ public class Builder {
 
     private Optional<String> trace = Optional.empty();
   }
+
+  private static class KafkaPluginInfo {
+    @JsonProperty("class")
+    private String className;
+  }
 }
diff --git a/common/src/main/java/org/astraea/common/connector/ConnectorClient.java b/common/src/main/java/org/astraea/common/connector/ConnectorClient.java
index 522e7ec3..f9457bd5 100644
--- a/common/src/main/java/org/astraea/common/connector/ConnectorClient.java
+++ b/common/src/main/java/org/astraea/common/connector/ConnectorClient.java
@@ -57,9 +57,9 @@ public interface ConnectorClient {
 
   CompletionStage<Void> deleteConnector(String name);
 
-  CompletionStage<Validation> validate(String name, Map<String, String> configs);
+  CompletionStage<Validation> validate(String className, Map<String, String> configs);
 
-  CompletionStage<Set<PluginInfo>> plugins();
+  CompletionStage<List<PluginInfo>> plugins();
 
   default CompletionStage<Boolean> waitConnectorInfo(
       String connectName, Predicate<ConnectorStatus> predicate, Duration timeout) {
diff --git a/common/src/main/java/org/astraea/common/connector/PluginInfo.java b/common/src/main/java/org/astraea/common/connector/PluginInfo.java
index 396cdb1a..7a81fb1c 100644
--- a/common/src/main/java/org/astraea/common/connector/PluginInfo.java
+++ b/common/src/main/java/org/astraea/common/connector/PluginInfo.java
@@ -16,17 +16,24 @@
  */
 package org.astraea.common.connector;
 
-import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
 
+/** this is not a kind of json response from kafka. We compose it with definition. */
 public class PluginInfo {
+  private final String className;
 
-  /** connector plugin class name */
-  @JsonProperty("class")
-  private String className;
+  private final List<Definition> definitions;
 
-  public PluginInfo() {}
+  public PluginInfo(String className, List<Definition> definitions) {
+    this.className = className;
+    this.definitions = definitions;
+  }
 
   public String className() {
     return className;
   }
+
+  public List<Definition> definitions() {
+    return definitions;
+  }
 }
diff --git a/common/src/main/java/org/astraea/common/cost/NetworkCost.java b/common/src/main/java/org/astraea/common/cost/NetworkCost.java
new file mode 100644
index 00000000..28a40ed1
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/cost/NetworkCost.java
@@ -0,0 +1,175 @@
+/*
+ * 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.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.astraea.common.admin.BrokerTopic;
+import org.astraea.common.admin.ClusterBean;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.admin.Replica;
+import org.astraea.common.admin.ReplicaInfo;
+import org.astraea.common.admin.TopicPartition;
+import org.astraea.common.metrics.HasBeanObject;
+import org.astraea.common.metrics.broker.HasRate;
+import org.astraea.common.metrics.broker.LogMetrics;
+import org.astraea.common.metrics.broker.ServerMetrics;
+import org.astraea.common.metrics.collector.Fetcher;
+
+/**
+ * This cost function calculate the load balance score in terms of network ingress or network
+ * egress. The implementation is work under these assumptions:
+ *
+ * <ol>
+ *   <li>The loading between each partition can be different. That is, there might be skewed load
+ *       behavior between each partition.
+ *   <li>The network ingress data rate of each partition is constant, it won't fluctuate over time.
+ *   <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.
+ * </ol>
+ */
+public abstract class NetworkCost implements HasClusterCost {
+
+  private final AtomicReference<ClusterInfo<Replica>> currentCluster = new AtomicReference<>();
+
+  abstract ServerMetrics.Topic useMetric();
+
+  void updateCurrentCluster(
+      ClusterInfo<Replica> clusterInfo,
+      ClusterBean clusterBean,
+      AtomicReference<ClusterInfo<Replica>> ref) {
+    // TODO: We need a reliable way to access the actual current cluster info. The following method
+    //  try to compare the equality of cluster info and cluster bean in terms of replica set. But it
+    //  didn't consider the data folder info. See the full discussion:
+    //  https://github.com/skiptests/astraea/pull/1240#discussion_r1044487473
+    var metricReplicas = clusterBean.replicas();
+    var mismatchSet =
+        clusterInfo.topicPartitionReplicas().stream()
+            .filter(tpr -> !metricReplicas.contains(tpr))
+            .collect(Collectors.toUnmodifiableSet());
+    if (mismatchSet.isEmpty()) ref.set(clusterInfo);
+    if (ref.get() == null)
+      fail("Initial clusterInfo required, the following replicas are mismatch: " + mismatchSet);
+  }
+
+  @Override
+  public ClusterCost clusterCost(ClusterInfo<Replica> clusterInfo, ClusterBean clusterBean) {
+    updateCurrentCluster(clusterInfo, clusterBean, currentCluster);
+
+    var dataRate = estimateRate(currentCluster.get(), clusterBean, useMetric());
+    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())),
+                        Collectors.summingDouble(x -> x))));
+
+    var summary = brokerRate.values().stream().mapToDouble(x -> x).summaryStatistics();
+    if (summary.getMax() < 0)
+      throw new IllegalStateException(
+          "Corrupted max rate: " + summary.getMax() + ", brokers: " + brokerRate);
+    if (summary.getMin() < 0)
+      throw new IllegalStateException(
+          "Corrupted min rate: " + summary.getMin() + ", brokers: " + brokerRate);
+    if (summary.getMax() == 0) return () -> 0; // edge case to avoid divided by zero error
+    double score = (summary.getMax() - summary.getMin()) / (summary.getMax());
+    return () -> score;
+  }
+
+  @Override
+  public Optional<Fetcher> fetcher() {
+    // 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));
+  }
+
+  private Map<BrokerTopic, List<Replica>> mapLeaderAllocation(
+      ClusterInfo<? extends Replica> clusterInfo) {
+    return clusterInfo
+        .replicaStream()
+        .filter(ReplicaInfo::isOnline)
+        .filter(ReplicaInfo::isLeader)
+        .map(r -> Map.entry(BrokerTopic.of(r.nodeInfo().id(), r.topic()), r))
+        .collect(
+            Collectors.groupingBy(
+                Map.Entry::getKey,
+                Collectors.mapping(Map.Entry::getValue, Collectors.toUnmodifiableList())));
+  }
+
+  Map<TopicPartition, Long> estimateRate(
+      ClusterInfo<? extends Replica> clusterInfo,
+      ClusterBean clusterBean,
+      ServerMetrics.Topic metric) {
+    return mapLeaderAllocation(clusterInfo).entrySet().stream()
+        .flatMap(
+            e -> {
+              var bt = e.getKey();
+              var totalSize = e.getValue().stream().mapToLong(Replica::size).sum();
+              var totalShare =
+                  (double)
+                      clusterBean
+                          .brokerTopicMetrics(bt, ServerMetrics.Topic.Meter.class)
+                          .filter(bean -> bean.type().equals(metric))
+                          .max(Comparator.comparingLong(HasBeanObject::createdTimestamp))
+                          .map(HasRate::fifteenMinuteRate)
+                          .orElseThrow(
+                              () ->
+                                  new NoSufficientMetricsException(
+                                      this, Duration.ofSeconds(1), "No metric for " + bt));
+              if (Double.isNaN(totalShare) || totalShare < 0)
+                throw new NoSufficientMetricsException(
+                    this,
+                    Duration.ofSeconds(1),
+                    "Illegal load value " + totalShare + " for broker-topic: " + bt);
+              var calculateShare =
+                  (Function<Replica, Long>)
+                      (replica) ->
+                          totalSize > 0
+                              ? (long) ((totalShare * replica.size()) / totalSize)
+                              : totalSize == 0
+                                  ? 0L
+                                  : fail("Illegal replica with negative size: " + replica);
+
+              return e.getValue().stream()
+                  .map(r -> Map.entry(r.topicPartition(), calculateShare.apply(r)));
+            })
+        .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  private <T> T notNull(T value) {
+    if (value == null)
+      throw new NoSufficientMetricsException(this, Duration.ofSeconds(1), "No metric");
+    return value;
+  }
+
+  private <T> T fail(String reason) {
+    throw new NoSufficientMetricsException(this, Duration.ofSeconds(1), reason);
+  }
+}
diff --git a/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java b/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java
new file mode 100644
index 00000000..4c9e3a69
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/cost/NetworkEgressCost.java
@@ -0,0 +1,30 @@
+/*
+ * 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 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;
+  }
+}
diff --git a/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java b/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java
new file mode 100644
index 00000000..641e33c8
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/cost/NetworkIngressCost.java
@@ -0,0 +1,30 @@
+/*
+ * 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 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;
+  }
+}
diff --git a/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java b/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java
index 4b39b8a8..8fd5e4b7 100644
--- a/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java
@@ -147,6 +147,12 @@ public final class LogMetrics {
                   .property("topic", "*")
                   .property("partition", "*")
                   .property("name", metricName)
+                  // Due to a Kafka bug. This log metrics might come with an `is-future` property
+                  // with it.
+                  // And the property is never removed even if the folder migration is done.
+                  // We use the BeanQuery property list pattern to work around this issue.
+                  // See https://github.com/apache/kafka/pull/12979
+                  .usePropertyListPattern()
                   .build())
           .stream()
           .map(Gauge::new)
diff --git a/connector/src/main/java/org/astraea/connector/MetadataStorage.java b/connector/src/main/java/org/astraea/connector/MetadataStorage.java
new file mode 100644
index 00000000..29af1342
--- /dev/null
+++ b/connector/src/main/java/org/astraea/connector/MetadataStorage.java
@@ -0,0 +1,38 @@
+/*
+ * 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.connector;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.kafka.connect.storage.OffsetStorageReader;
+import org.astraea.common.json.JsonConverter;
+
+public interface MetadataStorage {
+
+  JsonConverter CONVERTER = JsonConverter.defaultConverter();
+
+  static MetadataStorage of(OffsetStorageReader reader) {
+    return index -> {
+      var v = reader.offset(index);
+      if (v == null) return Map.of();
+      return v.entrySet().stream()
+          .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> e.getValue().toString()));
+    };
+  }
+
+  Map<String, String> metadata(Map<String, String> index);
+}
diff --git a/connector/src/main/java/org/astraea/connector/SourceConnector.java b/connector/src/main/java/org/astraea/connector/SourceConnector.java
index 672d3d23..bed8ed44 100644
--- a/connector/src/main/java/org/astraea/connector/SourceConnector.java
+++ b/connector/src/main/java/org/astraea/connector/SourceConnector.java
@@ -26,7 +26,7 @@ import org.astraea.common.VersionUtils;
 public abstract class SourceConnector extends org.apache.kafka.connect.source.SourceConnector {
   public static String TOPICS_KEY = "topics";
 
-  protected abstract void init(Configuration configuration);
+  protected abstract void init(Configuration configuration, MetadataStorage storage);
 
   protected abstract Class<? extends SourceTask> task();
 
@@ -41,7 +41,7 @@ public abstract class SourceConnector extends org.apache.kafka.connect.source.So
   // -------------------------[final]-------------------------//
   @Override
   public final void start(Map<String, String> props) {
-    init(Configuration.of(props));
+    init(Configuration.of(props), MetadataStorage.of(context().offsetStorageReader()));
   }
 
   @Override
diff --git a/connector/src/main/java/org/astraea/connector/SourceRecord.java b/connector/src/main/java/org/astraea/connector/SourceRecord.java
index a9009c3d..88e9c2ad 100644
--- a/connector/src/main/java/org/astraea/connector/SourceRecord.java
+++ b/connector/src/main/java/org/astraea/connector/SourceRecord.java
@@ -21,8 +21,6 @@ import java.util.Map;
 import java.util.Optional;
 import org.astraea.common.Header;
 import org.astraea.common.admin.TopicPartition;
-import org.astraea.common.json.JsonConverter;
-import org.astraea.common.json.TypeRef;
 import org.astraea.common.producer.Record;
 
 public class SourceRecord implements Record<byte[], byte[]> {
@@ -31,8 +29,6 @@ public class SourceRecord implements Record<byte[], byte[]> {
     return new Builder();
   }
 
-  private static final JsonConverter CONVERTER = JsonConverter.defaultConverter();
-
   private final Record<byte[], byte[]> record;
   private final Map<String, String> metadataIndex;
   private final Map<String, String> metadata;
@@ -132,16 +128,13 @@ public class SourceRecord implements Record<byte[], byte[]> {
       return this;
     }
 
-    public Builder metadataIndex(Object metadataIndex) {
-      if (metadataIndex != null)
-        this.metadataIndex =
-            CONVERTER.fromJson(CONVERTER.toJson(metadataIndex), TypeRef.map(String.class));
+    public Builder metadataIndex(Map<String, String> metadataIndex) {
+      this.metadataIndex = metadataIndex;
       return this;
     }
 
-    public Builder metadata(Object metadata) {
-      if (metadata != null)
-        this.metadata = CONVERTER.fromJson(CONVERTER.toJson(metadata), TypeRef.map(String.class));
+    public Builder metadata(Map<String, String> metadata) {
+      this.metadata = metadata;
       return this;
     }
 
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 6304bd0e..cb14a9e6 100644
--- a/connector/src/main/java/org/astraea/connector/perf/PerfSource.java
+++ b/connector/src/main/java/org/astraea/connector/perf/PerfSource.java
@@ -24,6 +24,7 @@ import org.astraea.common.DataSize;
 import org.astraea.common.DistributionType;
 import org.astraea.common.Utils;
 import org.astraea.connector.Definition;
+import org.astraea.connector.MetadataStorage;
 import org.astraea.connector.SourceConnector;
 import org.astraea.connector.SourceTask;
 
@@ -73,7 +74,7 @@ public class PerfSource extends SourceConnector {
   private Configuration config;
 
   @Override
-  protected void init(Configuration configuration) {
+  protected void init(Configuration configuration, MetadataStorage storage) {
     this.config = configuration;
   }
 
diff --git a/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java b/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java
index 27a5cfcd..f2d3e0dd 100644
--- a/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java
+++ b/gui/src/main/java/org/astraea/gui/tab/ConnectorNode.java
@@ -16,6 +16,7 @@
  */
 package org.astraea.gui.tab;
 
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -25,7 +26,10 @@ import javafx.scene.Node;
 import org.astraea.common.FutureUtils;
 import org.astraea.common.MapUtils;
 import org.astraea.common.connector.ConnectorConfigs;
+import org.astraea.common.connector.Definition;
 import org.astraea.common.connector.WorkerStatus;
+import org.astraea.common.json.JsonConverter;
+import org.astraea.common.json.TypeRef;
 import org.astraea.common.metrics.connector.ConnectorMetrics;
 import org.astraea.gui.Context;
 import org.astraea.gui.pane.PaneBuilder;
@@ -74,10 +78,14 @@ public class ConnectorNode {
                 TextInput.required(ConnectorConfigs.TOPICS_KEY, EditableText.singleLine().build()),
                 TextInput.required(
                     ConnectorConfigs.TASK_MAX_KEY, EditableText.singleLine().onlyNumber().build()),
-                TextInput.of("configs", EditableText.multiline().build())),
+                TextInput.of("configs", EditableText.multiline().hint("{\"a\":\"b\"}").build())),
             "CREATE",
             (argument, logger) -> {
               var req = new HashMap<>(argument.nonEmptyTexts());
+              var configs = req.remove("configs");
+              if (configs != null)
+                req.putAll(
+                    JsonConverter.defaultConverter().fromJson(configs, TypeRef.map(String.class)));
               return context
                   .connectorClient()
                   .createConnector(req.remove(ConnectorConfigs.NAME_KEY), req)
@@ -264,7 +272,11 @@ public class ConnectorNode {
                                 .map(
                                     pluginInfo -> {
                                       var map = new LinkedHashMap<String, Object>();
-                                      map.put("class name", pluginInfo.className());
+                                      map.put("class", pluginInfo.className());
+                                      pluginInfo.definitions().stream()
+                                          .sorted(Comparator.comparing(Definition::name))
+                                          .forEach(
+                                              d -> map.put(d.name(), d.defaultValue().orElse("")));
                                       return map;
                                     })
                                 .collect(Collectors.toList())))


Test patch:
diff --git a/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java b/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java
index 872cce3e..6bb2bb1e 100644
--- a/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java
+++ b/common/src/test/java/org/astraea/common/connector/ConnectorClientTest.java
@@ -51,10 +51,7 @@ class ConnectorClientTest extends RequireWorkerCluster {
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
     var validation =
         connectorClient
-            .validate(
-                "TestTextSourceConnector",
-                Map.of(
-                    ConnectorConfigs.CONNECTOR_CLASS_KEY, TestTextSourceConnector.class.getName()))
+            .validate(TestTextSourceConnector.class.getName(), Map.of())
             .toCompletableFuture()
             .join();
     assertNotEquals(0, validation.errorCount());
@@ -220,9 +217,11 @@ class ConnectorClientTest extends RequireWorkerCluster {
   void testPlugin() {
     var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
     var plugins = connectorClient.plugins().toCompletableFuture().join();
+    assertNotEquals(0, plugins.size());
     assertTrue(
         plugins.stream()
             .anyMatch(x -> TestTextSourceConnector.class.getName().equals(x.className())));
+    plugins.forEach(p -> assertNotEquals(0, p.definitions().size()));
   }
 
   @Test
diff --git a/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java b/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java
new file mode 100644
index 00000000..311d0920
--- /dev/null
+++ b/common/src/test/java/org/astraea/common/cost/NetworkCostTest.java
@@ -0,0 +1,435 @@
+/*
+ * 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.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.astraea.common.DataRate;
+import org.astraea.common.admin.ClusterBean;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.admin.ClusterInfoBuilder;
+import org.astraea.common.admin.Replica;
+import org.astraea.common.admin.ReplicaInfo;
+import org.astraea.common.admin.TopicPartition;
+import org.astraea.common.balancer.Balancer;
+import org.astraea.common.balancer.algorithms.AlgorithmConfig;
+import org.astraea.common.balancer.tweakers.ShuffleTweaker;
+import org.astraea.common.metrics.BeanObject;
+import org.astraea.common.metrics.broker.LogMetrics;
+import org.astraea.common.metrics.broker.ServerMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class NetworkCostTest {
+
+  private static NetworkIngressCost ingressCost() {
+    return new NetworkIngressCost() {
+      @Override
+      void updateCurrentCluster(
+          ClusterInfo<Replica> clusterInfo,
+          ClusterBean clusterBean,
+          AtomicReference<ClusterInfo<Replica>> ref) {
+        ref.compareAndSet(null, clusterInfo);
+      }
+    };
+  }
+
+  private static NetworkEgressCost egressCost() {
+    return new NetworkEgressCost() {
+      @Override
+      void updateCurrentCluster(
+          ClusterInfo<Replica> clusterInfo,
+          ClusterBean clusterBean,
+          AtomicReference<ClusterInfo<Replica>> ref) {
+        ref.compareAndSet(null, clusterInfo);
+      }
+    };
+  }
+
+  @Test
+  void testInitialClusterInfo() {
+    var cost = new NetworkIngressCost();
+    var initial =
+        ClusterInfoBuilder.builder()
+            .addNode(Set.of(1))
+            .addFolders(Map.of(1, Set.of("/folder0", "/folder1")))
+            .addTopic("Pipeline", 1, (short) 1)
+            .addNode(Set.of(2))
+            .addFolders(Map.of(2, Set.of("/folder0", "/folder1")))
+            .build();
+    var modified = new ShuffleTweaker(10, 30).generate(initial).findFirst().orElseThrow();
+    var withLog =
+        ClusterBean.of(
+            Map.of(
+                1,
+                List.of(
+                    logSize(TopicPartition.of("Pipeline", 0), -1),
+                    bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Pipeline", 0))));
+    var noLog =
+        ClusterBean.of(
+            Map.of(1, List.of(bandwidth(ServerMetrics.Topic.BYTES_IN_PER_SEC, "Pipeline", 0))));
+
+    Assertions.assertThrows(
+        NoSufficientMetricsException.class, () -> cost.clusterCost(initial, ClusterBean.EMPTY));
+    Assertions.assertThrows(
+        NoSufficientMetricsException.class, () -> cost.clusterCost(modified, ClusterBean.EMPTY));
+    Assertions.assertThrows(
+        NoSufficientMetricsException.class, () -> cost.clusterCost(initial, noLog));
+    Assertions.assertThrows(
+        NoSufficientMetricsException.class, () -> cost.clusterCost(modified, noLog));
+    Assertions.assertDoesNotThrow(() -> cost.clusterCost(initial, withLog));
+    Assertions.assertDoesNotThrow(() -> cost.clusterCost(modified, noLog));
+  }
+
+  @Test
+  void testEstimatedIngressRate() {
+    testEstimatedRate(ServerMetrics.Topic.BYTES_IN_PER_SEC);
+  }
+
+  @Test
+  void testEstimatedEgressRate() {
+    testEstimatedRate(ServerMetrics.Topic.BYTES_OUT_PER_SEC);
+  }
+
+  void testEstimatedRate(ServerMetrics.Topic metric) {
+    var t = new HandWrittenTestCase(metric);
+    new NetworkIngressCost()
+        .estimateRate(t.clusterInfo(), t.clusterBean(), metric)
+        .forEach(
+            (tp, actual) -> {
+              var expected = t.expectedRate().get(tp);
+              var test = (expected - 3) + " < " + actual + " < " + (expected + 3);
+              System.out.println(test);
+              Assertions.assertTrue((expected - 3) < actual && actual < (expected + 3), test);
+            });
+  }
+
+  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)));
+  }
+
+  @ParameterizedTest
+  @MethodSource("testcases")
+  @DisplayName("Run with Balancer")
+  void testOptimization(HasClusterCost costFunction, TestCase testcase) {
+    var newPlan =
+        Balancer.Official.Greedy.create(
+                AlgorithmConfig.builder()
+                    .clusterCost(costFunction)
+                    .metricSource(testcase::clusterBean)
+                    .build())
+            .offer(testcase.clusterInfo(), Duration.ofSeconds(1));
+
+    Assertions.assertTrue(newPlan.isPresent());
+    System.out.println("Initial cost: " + newPlan.get().initialClusterCost().value());
+    System.out.println("New cost: " + newPlan.get().proposalClusterCost().value());
+    Assertions.assertTrue(
+        newPlan.get().initialClusterCost().value() > newPlan.get().proposalClusterCost().value());
+  }
+
+  @Test
+  void testCompositeOptimization() {
+    var testCase =
+        new LargeTestCase(
+            ServerMetrics.Topic.BYTES_IN_PER_SEC,
+            ServerMetrics.Topic.BYTES_OUT_PER_SEC,
+            10,
+            300,
+            0xfeed);
+    var costFunction =
+        HasClusterCost.of(
+            Map.of(
+                ingressCost(), 1.0,
+                egressCost(), 1.0));
+    testOptimization(costFunction, testCase);
+  }
+
+  interface TestCase {
+
+    ClusterInfo<Replica> clusterInfo();
+
+    ClusterBean clusterBean();
+  }
+
+  /**
+   * A manually crafted cluster & metrics
+   *
+   * <ul>
+   *   <li>3 Brokers.
+   *   <li>Broker 3 has no load at all.
+   * </ul>
+   */
+  private static class HandWrittenTestCase implements TestCase {
+    private HandWrittenTestCase(ServerMetrics.Topic metric) {
+      this.clusterBean =
+          ClusterBean.of(
+              Map.of(
+                  1,
+                  expectedBrokerTopicBandwidth.get(1).entrySet().stream()
+                      .map(e -> bandwidth(metric, e.getKey(), e.getValue()))
+                      .collect(Collectors.toUnmodifiableList()),
+                  2,
+                  expectedBrokerTopicBandwidth.get(2).entrySet().stream()
+                      .map(e -> bandwidth(metric, e.getKey(), e.getValue()))
+                      .collect(Collectors.toUnmodifiableList()),
+                  3,
+                  expectedBrokerTopicBandwidth.get(3).entrySet().stream()
+                      .map(e -> bandwidth(metric, e.getKey(), e.getValue()))
+                      .collect(Collectors.toUnmodifiableList())));
+    }
+
+    final ClusterInfo<Replica> base =
+        ClusterInfoBuilder.builder()
+            .addNode(Set.of(1, 2, 3))
+            .addFolders(Map.of(1, Set.of("/ssd1", "/ssd2", "/ssd3")))
+            .addFolders(Map.of(2, Set.of("/ssd1", "/ssd2", "/ssd3")))
+            .addFolders(Map.of(3, Set.of("/ssd1", "/ssd2", "/ssd3")))
+            .build();
+    final Map<TopicPartition, Long> expectedRate =
+        Map.of(
+            TopicPartition.of("Beef", 0), ThreadLocalRandom.current().nextLong(10000),
+            TopicPartition.of("Beef", 1), ThreadLocalRandom.current().nextLong(10000),
+            TopicPartition.of("Beef", 2), 0L,
+            TopicPartition.of("Beef", 3), ThreadLocalRandom.current().nextLong(10000),
+            TopicPartition.of("Pork", 0), ThreadLocalRandom.current().nextLong(10000),
+            TopicPartition.of("Pork", 1), ThreadLocalRandom.current().nextLong(10000),
+            TopicPartition.of("Pork", 2), 0L,
+            TopicPartition.of("Pork", 3), ThreadLocalRandom.current().nextLong(10000));
+
+    final Map<Integer, Map<String, Long>> expectedBrokerTopicBandwidth =
+        Stream.of(0, 1, 2)
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    index -> index + 1,
+                    index ->
+                        Stream.of("Beef", "Pork")
+                            .collect(
+                                Collectors.toUnmodifiableMap(
+                                    topic -> topic,
+                                    topic ->
+                                        expectedRate.entrySet().stream()
+                                            .filter(e -> e.getKey().topic().equals(topic))
+                                            .filter(e -> e.getKey().partition() % 3 == index)
+                                            .mapToLong(Map.Entry::getValue)
+                                            .sum()))));
+    final Function<Replica, Replica> modPlacement =
+        replica ->
+            Replica.builder(replica)
+                .size(expectedRate.get(replica.topicPartition()) * 100)
+                .nodeInfo(base.node(1 + replica.partition() % 3))
+                .build();
+    final ClusterInfo<Replica> clusterInfo =
+        ClusterInfoBuilder.builder(base)
+            .addTopic("Beef", 4, (short) 1, modPlacement)
+            .addTopic("Pork", 4, (short) 1, modPlacement)
+            .build();
+    final ClusterBean clusterBean;
+
+    public Map<TopicPartition, Long> expectedRate() {
+      return expectedRate;
+    }
+
+    @Override
+    public ClusterInfo<Replica> clusterInfo() {
+      return clusterInfo;
+    }
+
+    @Override
+    public ClusterBean clusterBean() {
+      return clusterBean;
+    }
+  }
+
+  /** A large cluster */
+  private static class LargeTestCase implements TestCase {
+
+    private final ClusterInfo<Replica> clusterInfo;
+    private final ClusterBean clusterBean;
+    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) {
+      var random = new Random(seed);
+      this.dataRateSupplier =
+          () -> {
+            switch (1 + random.nextInt(8)) {
+              case 1:
+              case 2:
+              case 3:
+              case 4:
+              case 5:
+                return DataRate.MB.of((long) (random.nextInt(50))).perSecond();
+              case 6:
+              case 7:
+                return DataRate.MB.of(3 * (long) (random.nextInt(50))).perSecond();
+              case 8:
+                return DataRate.MB.of(5 * (long) (random.nextInt(50))).perSecond();
+              default:
+                throw new RuntimeException();
+            }
+          };
+      this.clusterInfo =
+          ClusterInfoBuilder.builder()
+              .addNode(IntStream.range(0, brokers).boxed().collect(Collectors.toUnmodifiableSet()))
+              .addFolders(
+                  IntStream.range(0, brokers)
+                      .boxed()
+                      .collect(
+                          Collectors.toUnmodifiableMap(
+                              b -> b,
+                              b -> Set.of("/broker" + b + "/ssd1", "/broker" + b + "/ssd2"))))
+              .addTopic(
+                  "Pipeline",
+                  partitions,
+                  (short) 1,
+                  replica ->
+                      Replica.builder(replica)
+                          .size((long) (24 * 60 * 60 * dataRateSupplier.get().byteRate()))
+                          .build())
+              .build();
+      this.rate =
+          clusterInfo.topicPartitions().stream()
+              .collect(
+                  Collectors.toUnmodifiableMap(
+                      tp -> tp,
+                      tp -> clusterInfo.replicaLeader(tp).orElseThrow().size() / 24 / 60 / 60));
+      var consumerFanout =
+          IntStream.range(0, partitions)
+              .boxed()
+              .collect(
+                  Collectors.toUnmodifiableMap(
+                      p -> TopicPartition.of("Pipeline", p), p -> random.nextInt(10)));
+      this.clusterBean =
+          ClusterBean.of(
+              IntStream.range(0, brokers)
+                  .boxed()
+                  .collect(
+                      Collectors.toUnmodifiableMap(
+                          id -> id,
+                          id ->
+                              List.of(
+                                  noise(random.nextInt()),
+                                  noise(random.nextInt()),
+                                  noise(random.nextInt()),
+                                  noise(random.nextInt()),
+                                  bandwidth(
+                                      metric0,
+                                      "Pipeline",
+                                      clusterInfo
+                                          .replicaStream()
+                                          .filter(r -> r.nodeInfo().id() == id)
+                                          .filter(ReplicaInfo::isLeader)
+                                          .filter(ReplicaInfo::isOnline)
+                                          .mapToLong(r -> rate.get(r.topicPartition()))
+                                          .sum()),
+                                  noise(random.nextInt()),
+                                  noise(random.nextInt()),
+                                  bandwidth(
+                                      metric1,
+                                      "Pipeline",
+                                      clusterInfo
+                                          .replicaStream()
+                                          .filter(r -> r.nodeInfo().id() == id)
+                                          .filter(ReplicaInfo::isLeader)
+                                          .filter(ReplicaInfo::isOnline)
+                                          .mapToLong(
+                                              r ->
+                                                  rate.get(r.topicPartition())
+                                                      * consumerFanout.get(r.topicPartition()))
+                                          .sum()),
+                                  noise(random.nextInt())))));
+    }
+
+    @Override
+    public ClusterInfo<Replica> clusterInfo() {
+      return clusterInfo;
+    }
+
+    @Override
+    public ClusterBean clusterBean() {
+      return clusterBean;
+    }
+  }
+
+  static ServerMetrics.Topic.Meter noise(int seed) {
+    var random = new Random(seed);
+    return bandwidth(
+        random.nextInt() > 0
+            ? ServerMetrics.Topic.TOTAL_PRODUCE_REQUESTS_PER_SEC
+            : ServerMetrics.Topic.TOTAL_FETCH_REQUESTS_PER_SEC,
+        "noise_" + random.nextInt(),
+        random.nextDouble());
+  }
+
+  static ServerMetrics.Topic.Meter bandwidth(
+      ServerMetrics.Topic metric, String topic, double fifteenRate) {
+    if (metric == null) return noise(0);
+    var domainName = "kafka.server";
+    var properties =
+        Map.of("type", "BrokerTopicMetric", "topic", topic, "name", metric.metricName());
+    var attributes = Map.<String, Object>of("FifteenMinuteRate", fifteenRate);
+    return new ServerMetrics.Topic.Meter(new BeanObject(domainName, properties, attributes));
+  }
+
+  static LogMetrics.Log.Gauge logSize(TopicPartition topicPartition, long size) {
+    var domainName = LogMetrics.DOMAIN_NAME;
+    var properties =
+        Map.of(
+            "type", "BrokerTopicMetric",
+            "topic", topicPartition.topic(),
+            "partition", String.valueOf(topicPartition.partition()),
+            "name", LogMetrics.Log.SIZE.metricName());
+    var attributes = Map.<String, Object>of("value", size);
+    return new LogMetrics.Log.Gauge(new BeanObject(domainName, properties, attributes));
+  }
+}
diff --git a/connector/src/test/java/org/astraea/connector/ConnectorTest.java b/connector/src/test/java/org/astraea/connector/ConnectorTest.java
index 41e444db..46b55ab8 100644
--- a/connector/src/test/java/org/astraea/connector/ConnectorTest.java
+++ b/connector/src/test/java/org/astraea/connector/ConnectorTest.java
@@ -16,6 +16,8 @@
  */
 package org.astraea.connector;
 
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
 import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
@@ -34,6 +36,14 @@ import org.junit.jupiter.api.Test;
 
 public class ConnectorTest extends RequireWorkerCluster {
 
+  @Test
+  void testPlugin() {
+    var connectorClient = ConnectorClient.builder().url(workerUrl()).build();
+    var plugins = connectorClient.plugins().toCompletableFuture().join();
+    assertNotEquals(0, plugins.size());
+    plugins.forEach(p -> assertNotEquals(0, p.definitions().size()));
+  }
+
   @Test
   void testCustomConnectorPlugins() {
     var client = ConnectorClient.builder().url(workerUrl()).build();
@@ -91,7 +101,7 @@ public class ConnectorTest extends RequireWorkerCluster {
     private static final AtomicInteger CLOSE_COUNT = new AtomicInteger(0);
 
     @Override
-    protected void init(Configuration configuration) {
+    protected void init(Configuration configuration, MetadataStorage storage) {
       INIT_COUNT.incrementAndGet();
     }
 
diff --git a/connector/src/test/java/org/astraea/connector/MetadataStorageTest.java b/connector/src/test/java/org/astraea/connector/MetadataStorageTest.java
new file mode 100644
index 00000000..d9c87307
--- /dev/null
+++ b/connector/src/test/java/org/astraea/connector/MetadataStorageTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.connector;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.common.Configuration;
+import org.astraea.common.Utils;
+import org.astraea.common.connector.ConnectorClient;
+import org.astraea.common.connector.ConnectorConfigs;
+import org.astraea.common.producer.Record;
+import org.astraea.it.RequireSingleWorkerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class MetadataStorageTest extends RequireSingleWorkerCluster {
+
+  private static final Map<String, String> KEY = Map.of("a", "b");
+  private static final Map<String, String> VALUE = Map.of("c", "d");
+
+  @Test
+  void test() {
+    var client = ConnectorClient.builder().url(workerUrl()).build();
+    var name = Utils.randomString();
+    var topic = Utils.randomString();
+    client
+        .createConnector(
+            name,
+            Map.of(
+                ConnectorConfigs.CONNECTOR_CLASS_KEY,
+                MySource.class.getName(),
+                ConnectorConfigs.TASK_MAX_KEY,
+                "1",
+                ConnectorConfigs.TOPICS_KEY,
+                topic))
+        .toCompletableFuture()
+        .join();
+
+    Utils.sleep(Duration.ofSeconds(3));
+
+    var status = client.connectorStatus(name).toCompletableFuture().join();
+    Assertions.assertEquals("RUNNING", status.state());
+    Assertions.assertEquals(1, status.tasks().size());
+    status
+        .tasks()
+        .forEach(t -> Assertions.assertEquals("RUNNING", t.state(), t.error().orElse("no error")));
+
+    // delete connector to check metadata storage later
+    client.deleteConnector(name).toCompletableFuture().join();
+    Utils.sleep(Duration.ofSeconds(3));
+
+    client
+        .createConnector(
+            name,
+            Map.of(
+                ConnectorConfigs.CONNECTOR_CLASS_KEY,
+                MySource.class.getName(),
+                ConnectorConfigs.TASK_MAX_KEY,
+                "1",
+                ConnectorConfigs.TOPICS_KEY,
+                topic))
+        .toCompletableFuture()
+        .join();
+    Utils.sleep(Duration.ofSeconds(3));
+    Assertions.assertNotEquals(0, MySource.FETCHED_METADATA.size());
+    Assertions.assertEquals("d", MySource.FETCHED_METADATA.get("c"));
+  }
+
+  public static class MySource extends SourceConnector {
+
+    private static volatile Map<String, String> FETCHED_METADATA = Map.of();
+
+    private Configuration configuration;
+
+    @Override
+    protected void init(Configuration configuration, MetadataStorage storage) {
+      this.configuration = configuration;
+      FETCHED_METADATA = storage.metadata(KEY);
+    }
+
+    @Override
+    protected Class<? extends SourceTask> task() {
+      return MyTask.class;
+    }
+
+    @Override
+    protected List<Configuration> takeConfiguration(int maxTasks) {
+      return IntStream.range(0, maxTasks).mapToObj(i -> configuration).collect(Collectors.toList());
+    }
+
+    @Override
+    protected List<Definition> definitions() {
+      return List.of();
+    }
+  }
+
+  public static class MyTask extends SourceTask {
+
+    private boolean isDone = false;
+    private Set<String> topics = Set.of();
+
+    @Override
+    protected void init(Configuration configuration) {
+      topics = Set.copyOf(configuration.list(ConnectorConfigs.TOPICS_KEY, ","));
+    }
+
+    @Override
+    protected Collection<Record<byte[], byte[]>> take() {
+      if (isDone) return List.of();
+      isDone = true;
+      return topics.stream()
+          .map(
+              t ->
+                  SourceRecord.builder()
+                      .topic(t)
+                      .key(new byte[100])
+                      .metadataIndex(KEY)
+                      .metadata(VALUE)
+                      .build())
+          .collect(Collectors.toList());
+    }
+  }
+}
diff --git a/connector/src/test/java/org/astraea/connector/SourceTest.java b/connector/src/test/java/org/astraea/connector/SourceTest.java
index d38ba879..00491f03 100644
--- a/connector/src/test/java/org/astraea/connector/SourceTest.java
+++ b/connector/src/test/java/org/astraea/connector/SourceTest.java
@@ -101,7 +101,7 @@ public class SourceTest extends RequireSingleWorkerCluster {
     private Configuration configuration = Configuration.EMPTY;
 
     @Override
-    protected void init(Configuration configuration) {
+    protected void init(Configuration configuration, MetadataStorage storage) {
       this.configuration = configuration;
     }