Repository: skiptests/astraea
Pull request #315 (2022-05-02 17:05:46)

Smooth weight Round Robin Dispatcher
#298 中提到的第三階段

實做了smooth weight，在`SmoothWeightRoundRobinDispatcher`中加入了`unusedPartitions`deque用來存放調用卻未被真正發送的target partition，解決之前的結果沒有每次都被調用這一問題。
對`Periodic`中的`tryUpdate`方法進行了一些小的修改，爲其添加了second parameter。因爲有需求是需要等待一段時間收集數據再做計算調整。所以時間上改成了可以自己定義。
進一步完全捨棄了隨機性，當確認了target broker後。對於選擇partition這一件事現在會在Broker內做round robin。

我做了一些測試，它的數據結果與之前的差不多，證明其有在正常運作。
不過我注意到如果一個producer instance要同時服務於多個topic，那它可能不太能正常運作。這個問題我會再發個issue討論一下修改方向，然後在下一個pr中更新。

wycccccc: 另外還有個問題想要請教下學長，如果我一個叢集由三臺機器構成。然後我用三臺機器向這個叢集打資料，打資料的topic分佈都是均勻的，使用的是Kafka default partitioner。那麼是不是就可以說，得到的throughput就是三打三這一情況下Kafka所能達到的throughput能力的上限。
chia7712: > 那麼是不是就可以說，得到的throughput就是三打三這一情況下Kafka排除partitioner的影響，所能達到的throughput能力的上限。

可以從伺服器端來看是否有把網路頻寬填滿
wycccccc: @chia7712  有一點點，我把衝突修了。

單節點throughput上限問題
我測試了一下在Broker上本機打本機。效能並沒有改善，還是只有750MB/s。
我還觀察了一下Kafka logs都是很平常的log，沒有異樣。

我有跑幾次jmc在Broker端，收集到的數據樣本都差不多，以下是其中一次的數據大體報告。說實話我對Broker端的接受數據這段不是很熟可能還需要讀下書。。。
看上去物理內存相當吃盡，但我記得kafka好像會把消息從producer到consumer儘可能在內存中完成。可能是這個原因造成的？data-plane-kafka-request-handler似乎經常block，或許原因應該出在這裏，不過我不太清楚是哪個原因造成了block，可能需要學長點撥一下。
![image](https://user-images.githubusercontent.com/43372856/167303095-98391486-7277-4097-8142-40032f0ec2b3.png)

![image](https://user-images.githubusercontent.com/43372856/167303029-e59f825b-dccb-429b-8be6-e583a061b7a8.png)

如果上述內容都還算正常的話，那可能還需要再找找定位一下問題了。
chia7712: > 我測試了一下在Broker上本機打本機。效能並沒有改善，還是只有750MB/s。

老樣子，要先區分是不是Kafka這個服務不給力、還是硬碟不給力，麻煩原地打一下硬碟看一下長時間的吞吐量行不行

> data-plane-kafka-request-handler似乎經常block

這是正常的，因為它在等request

> 如果上述內容都還算正常的話，那可能還需要再找找定位一下問題了。

假如上面說的硬碟問題確認完了，接下來確認一下是不是我們的效能工具打一個節點有效能問題，可以試試看原地開兩個performance tool打資料，看看兩個加總起來的吞吐量有沒有突破750MB
wycccccc: @chia7712 
我先去開個issue討論這個問題，感覺這些內容與這個pr沒有太大關係。後續這個問題的更新我會放在#319
chia7712: `SmoothWeightRoundRobinDispatcher`和`SmoothWeightDispatcher`有什麼區別？有明確的理由要引入新的`dispatcher`嗎？包含這隻PR，專案裡面會有三個`Dispatcher`也就是三種不同用cost functions的邏輯，我們必須要有明確的理由才能引入更多，否則會淪為為做而做，這樣不是一個好的方向
wycccccc: 他們是一樣的，實現的是同一個功能。`SmoothWeightDispatcher`應該是要被移除的，可能是解決衝突的時候被添加了回來。這是那時候學弟修改的，但實作上其實與真正的smoothWeight有一些差別，我就把他全部打掉按smoothWeight的邏輯修改了。我再把`SmoothWeightDispatcher`移除。
chia7712: > 他們是一樣的，實現的是同一個功能。SmoothWeightDispatcher應該是要被移除的，可能是解決衝突的時候被添加了回來。這是那時候學弟修改的，但實作上其實與真正的smoothWeight有一些差別，我就把他全部打掉按smoothWeight的邏輯修改了。我再把SmoothWeightDispatcher移除。

ok，那記得要完成後續清理不必要程式碼的工作

另外有機會在數據上或任何方式來呈現`SmoothWeightRoundRobinDispatcher`能帶來比簡單的strict dispatcher的好處嗎？我會這樣問是因為`SmoothWeightRoundRobinDispatcher`帶來了比較複雜的邏輯，因此也要有對應的好處才能證明他的價值
wycccccc: > 另外有機會在數據上或任何方式來呈現SmoothWeightRoundRobinDispatcher能帶來比簡單的strict dispatcher的好處嗎？我會這樣問是因為SmoothWeightRoundRobinDispatcher帶來了比較複雜的邏輯，因此也要有對應的好處才能證明他的價值

我看了一下strict dispatcher的邏輯，它似乎每次都會選擇負載分數最小的節點。這代表在分數發生變化前所有資料都會發往一個Broker。這種發送方式直觀上感覺就不太健康。要數據對比的話我再跑跑看strict dispatcher。

wycccccc: 每秒一次，應該不只有送到同一節點這一問題。只打一個可能會造成block，然後隨着節點的增加可能要過好幾秒才有機會臨幸到某一個Broker。新的dispatch的smoothweight不會出現AAAAABBBBBCCCCCC這種情況，哪怕權重較小，例如C權重比較小也會變成ABABCABABC這種形式，而不會是AAAABBBBCC。


chia7712: > 新的dispatch的smoothweight不會出現AAAAABBBBBCCCCCC這種情況，哪怕權重較小，例如C權重比較小也會變成ABABCABABC這種形式，而不會是AAAABBBBCC。

我先確定一下我腦袋裡對這個問題的背景有沒有跟你同步。現在故事是來自dispatcher不管是從cluster info來的資訊、或是從metrics來的資訊都不會總是最新，因此strict dispatcher可能會有一段時間都是拿“同樣”的資訊在評估成本，進而導致你說的狀況 - AAAAABBBBBCCCCCC

如果上述正確的話，那麼smooth dispatcher也有一樣的窘境，也就是手上拿的資訊”也不是最新的“，因此是如何來”猜“此時可能最適合的節點？在一個小範圍（例如成本最低的前三名）裡面做亂數選擇嗎？
chia7712: > 不斷重複第二步，調整權重類似於一步一步改善的感覺。避免出現那種這十秒你分數很差我不給你打，結果後十秒因為我不給你打你分數又變很好我就狂給你打，這種循環中。

我想問的是這個設計的“核心概念”是基於什麼原理？當然我們都知道要避免你後面這兩句話，但具體來說參考的依據是什麼？就在本地透過“某種”方式微調權重進而改變排序嗎？如果只參考本地的資訊，同時那個“某種”方法在不同的producers身上都是一樣邏輯的話，會不會出現每個producers都在某個時刻“一起”轉向，例如每個producer同時產生ABABCABABC這種組合，以區域（自身）來看，好像避免了上述某個時間都打同一個節點的問題，但以全局來看，反而陷入更糟糕大家“都”打同一個節點的hot-spot
chia7712: > 認為一個叢集傳輸消息的業務應該是會運行一段時間的，不會是我工作1min就休息1min那種。而叢集中會有很多的業務，會有新業務啟動和停止，但叢集中的負載狀況應該會保持一個相對比較穩定但可能會緩慢變化的狀況

這個想法很符合我看到的狀況。用我的角度來想partitioner就是要設計出一個“能在各個節點的負載不斷起起落落的瞬間，找到最適合（通常是最閒置）的節點來擺放資料。
chia7712: > 不斷重複第二步，調整權重類似於一步一步改善的感覺

這句應該是關鍵，這個“調整”是基於什麼原理？如果不帶有隨機的概念的話，要怎麼避免我前面說每個producer同時轉向的問題？我會講隨機是因為這個方法很常用來處理一個情境：大家各自躲在房間裡算分，但彼此不知道彼此的狀況，因為算分的邏輯都一樣，因此很容易出現某段時間大家算出來的分數都一樣（因為一開始都是基於叢集來算，算法一樣，就容易走到一樣的結果）
wycccccc: > 我想問的是這個設計的“核心概念”是基於什麼原理？當然我們都知道要避免你後面這兩句話，但具體來說參考的依據是什麼？就在本地透過“某種”方式微調權重進而改變排序嗎？如果只參考本地的資訊，同時那個“某種”方法在不同的producers身上都是一樣邏輯的話，會不會出現每個producers都在某個時刻“一起”轉向，例如每個producer同時產生ABABCABABC這種組合，以區域（自身）來看，好像避免了上述某個時間都打同一個節點的問題，但以全局來看，反而陷入更糟糕大家“都”打同一個節點的hot-spot

確實會整體轉向，並且我認為除非我能在producer間建立通訊不然根據分數來判斷就會發生這種現象。之前的根據實時負載分數作為分數就有這一問題。
目前的做法是我想到的一個比較可行的解法，既然轉向無法避免，那麼我就減少轉向的幅度。
如果負載分數與負載分數均值差距較大那麼轉向的幅度就大，相反就小。

smooth dispatch的核心理念在於讓每一臺機器間的硬體資源消耗量竟可能接近，也就是整個叢集的負載分數標準差最小。
維持在標準差很小的情況下，讓他們在很小的幅度內來回轉向我認為是可以的結果。
chia7712: > 並且我認為除非我能在producer間建立通訊不然根據分數來判斷就會發生這種現象

這件事應該不太可能，因為這會回到客戶端收集資訊的通訊問題和量的問題

> 目前的做法是我想到的一個比較可行的解法，既然轉向無法避免，那麼我就減少轉向的幅度。
如果負載分數與負載分數均值差距較大那麼轉向的幅度就大，相反就小。
每次轉向都是建立在上一次轉向的結果上的

因次這裡有兩個故事要處理，一個是偶爾要轉向以避免某個時刻都打同個節點、另一個是要避免大家同時轉向又同時打同一個節點。現在這個解法聽起來降低轉向的幅度，但大家一樣有可能選擇一樣的幅度吧？還是有其他原理可以解決第二個問題？

> 讓每一臺機器間的硬體資源消耗量竟可能接近，也就是整個叢集的負載分數標準差最小。
根據負載分數不斷的轉向，每次轉向的幅度應該就會不斷變小。
最後收斂在標準差很小的情況下，讓他們在很小的幅度內來回轉向我認為是可以的結果。

轉向幅度小是否就是代表在分數前幾名的節點上選一個嗎？如果是的話那會怎麼選？
wycccccc: > 現在這個解法聽起來降低轉向的幅度，但大家一樣有可能選擇一樣的幅度吧？

是會同樣幅度轉向，但我的轉向可能只是讓producer每秒向轉向的節點多打1%資料。這樣感覺也沒有問題吧。

轉向是指轉向下圖中的effective weight，轉向只是在原有的effective weight上進行一定量的變化，變化的幅度取決於broker負載分數與平均分輸的差異。
而實際起到命中哪個節點的是下圖中的current weight的大小。current weight最高的會被命中。
然後每次要挑選節點時都是先讓current weight加上各自的effective weight，然後選中的節點會減去所有effective weight的綜合。

> 一個是偶爾要轉向以避免某個時刻都打同個節點

我不知道這個問題要同時刻打這一時間精度需要精確到哪個量級。如果是1秒的話我認為這一套體系就不會發生這個問題，如果說同一瞬間三台機器這筆資料都會選擇同一個節點，那麼我感覺應該是要降低這一個事件的概率，那麼smooth weight可以說能夠避免這件事連續發生。（因為他較低Effective weight 同樣有機會被選到）
* <p>||========================================||============================================||
 * ||------------ Broker in cluster ------------||------------- Effective weight -------------||
 * ||------------------Broker1------------------||-------------------- 5 ---------------------||
 * ||------------------Broker2------------------||-------------------- 1 ---------------------||
 * ||------------------Broker3------------------||-------------------- 1 ---------------------||
 * ||===========================================||============================================||
 *
 * <p>||===================||=======================||===============||======================||
 * ||--- Request Number ---|| Before current weight || Target Broker || After current weight ||
 * ||----------1-----------||------ {5, 1, 1} ------||----Broker1----||----- {-2, 1, 1} -----||
 * ||----------2-----------||------ {3, 2, 2} ------||----Broker1----||----- {-4, 2, 2} -----||
 * ||----------3-----------||------ {1, 3, 3} ------||----Broker2----||----- { 1,-4, 3} -----||
 * ||----------4-----------||------ {6,-3, 4} ------||----Broker1----||----- {-1,-3, 4} -----||
 * ||----------5-----------||------ {4,-2, 5} ------||----Broker3----||----- { 4,-2,-2} -----||
 * ||----------6-----------||------ {9,-1,-1} ------||----Broker1----||----- { 2,-1,-1} -----||
 * ||----------7-----------||------ {7, 0, 0} ------||----Broker1----||----- { 0, 0, 0} -----||
 * ||======================||=======================||===============||======================||
wycccccc: @chia7712 衝突已解決，這支還有其他議題需要討論嗎。

Patch:
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
index 68ada882..4fc5f55d 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -11,17 +11,29 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 public final class Utils {
 
+  private static Throwable unpack(Throwable exception) {
+    Throwable current = exception;
+    while (current instanceof ExecutionException) {
+      current = current.getCause();
+    }
+    return current;
+  }
+
   public static <R> R handleException(Getter<R> getter) {
     try {
       return getter.get();
-    } catch (Exception e) {
-      throw new RuntimeException(e);
+    } catch (Throwable exception) {
+      var current = unpack(exception);
+      if (current instanceof RuntimeException) throw (RuntimeException) current;
+      if (current == null) throw new RuntimeException("unknown error");
+      throw new RuntimeException(current);
     }
   }
 
@@ -137,12 +149,15 @@ public final class Utils {
     return value;
   }
 
-  public static boolean overSecond(long lastTime, int second) {
-    return (lastTime + Duration.ofSeconds(second).toMillis()) < System.currentTimeMillis();
-  }
-
-  public static boolean overSecond(long lastTime, Duration second) {
-    return (lastTime + second.toMillis()) < System.currentTimeMillis();
+  /**
+   * Check if the time is expired.
+   *
+   * @param lastTime Check time.
+   * @param interval Interval.
+   * @return Is expired.
+   */
+  public static boolean isExpired(long lastTime, Duration interval) {
+    return (lastTime + interval.toMillis()) < System.currentTimeMillis();
   }
 
   public static void sleep(Duration duration) {
diff --git a/app/src/main/java/org/astraea/admin/Builder.java b/app/src/main/java/org/astraea/admin/Builder.java
index 7eeec61d..8906b422 100644
--- a/app/src/main/java/org/astraea/admin/Builder.java
+++ b/app/src/main/java/org/astraea/admin/Builder.java
@@ -618,6 +618,7 @@ public class Builder {
 
         @Override
         public void create() {
+          if (connectionRate == Integer.MAX_VALUE) return;
           Utils.handleException(
               () ->
                   admin
@@ -627,7 +628,8 @@ public class Builder {
                                   new ClientQuotaEntity(Map.of(ClientQuotaEntity.IP, ip)),
                                   List.of(
                                       new ClientQuotaAlteration.Op(
-                                          "connection_creation_rate", (double) connectionRate)))))
+                                          Quota.Limit.IP_CONNECTION_RATE.nameOfKafka(),
+                                          (double) connectionRate)))))
                       .all()
                       .get());
         }
@@ -654,20 +656,26 @@ public class Builder {
 
         @Override
         public void create() {
-          Utils.handleException(
-              () ->
-                  admin
-                      .alterClientQuotas(
-                          List.of(
-                              new ClientQuotaAlteration(
-                                  new ClientQuotaEntity(Map.of(ClientQuotaEntity.CLIENT_ID, id)),
-                                  List.of(
-                                      new ClientQuotaAlteration.Op(
-                                          "producer_byte_rate", (double) produceRate),
-                                      new ClientQuotaAlteration.Op(
-                                          "consumer_byte_rate", (double) consumeRate)))))
-                      .all()
-                      .get());
+          var q = new ArrayList<ClientQuotaAlteration.Op>();
+          if (produceRate != Integer.MAX_VALUE)
+            q.add(
+                new ClientQuotaAlteration.Op(
+                    Quota.Limit.PRODUCER_BYTE_RATE.nameOfKafka(), (double) produceRate));
+          if (consumeRate != Integer.MAX_VALUE)
+            q.add(
+                new ClientQuotaAlteration.Op(
+                    Quota.Limit.CONSUMER_BYTE_RATE.nameOfKafka(), (double) consumeRate));
+          if (!q.isEmpty())
+            Utils.handleException(
+                () ->
+                    admin
+                        .alterClientQuotas(
+                            List.of(
+                                new ClientQuotaAlteration(
+                                    new ClientQuotaEntity(Map.of(ClientQuotaEntity.CLIENT_ID, id)),
+                                    q)))
+                        .all()
+                        .get());
         }
       };
     }
diff --git a/app/src/main/java/org/astraea/admin/Quota.java b/app/src/main/java/org/astraea/admin/Quota.java
index e955ab02..ae9fb62c 100644
--- a/app/src/main/java/org/astraea/admin/Quota.java
+++ b/app/src/main/java/org/astraea/admin/Quota.java
@@ -22,7 +22,7 @@ public class Quota {
                                         new Quota(
                                             target(stringStringEntry.getKey()),
                                             stringStringEntry.getValue(),
-                                            action(v.getKey()),
+                                            limit(v.getKey()),
                                             v.getValue()))))
         .collect(Collectors.toUnmodifiableList());
   }
@@ -34,14 +34,14 @@ public class Quota {
         .orElseThrow(() -> new RuntimeException("unknown: " + value));
   }
 
-  static Action action(String value) {
-    return Arrays.stream(Action.values())
+  static Limit limit(String value) {
+    return Arrays.stream(Limit.values())
         .filter(p -> p.nameOfKafka.equalsIgnoreCase(value))
         .findFirst()
         .orElseThrow(() -> new RuntimeException("unknown: " + value));
   }
 
-  enum Target {
+  public enum Target {
     USER(ClientQuotaEntity.USER),
     CLIENT_ID(ClientQuotaEntity.CLIENT_ID),
     IP(ClientQuotaEntity.IP);
@@ -51,12 +51,12 @@ public class Quota {
       this.nameOfKafka = nameOfKafka;
     }
 
-    String nameOfKafka() {
+    public String nameOfKafka() {
       return nameOfKafka;
     }
   }
 
-  enum Action {
+  public enum Limit {
     PRODUCER_BYTE_RATE("producer_byte_rate"),
     CONSUMER_BYTE_RATE("consumer_byte_rate"),
     REQUEST_PERCENTAGE("request_percentage"),
@@ -64,11 +64,11 @@ public class Quota {
     IP_CONNECTION_RATE("connection_creation_rate");
     private final String nameOfKafka;
 
-    Action(String nameOfKafka) {
+    Limit(String nameOfKafka) {
       this.nameOfKafka = nameOfKafka;
     }
 
-    String nameOfKafka() {
+    public String nameOfKafka() {
       return nameOfKafka;
     }
   }
@@ -76,14 +76,14 @@ public class Quota {
   private final Target target;
 
   private final String targetValue;
-  private final Action action;
-  private final double actionValue;
+  private final Limit limit;
+  private final double limitValue;
 
-  public Quota(Target target, String targetValue, Action action, double actionValue) {
+  public Quota(Target target, String targetValue, Limit limit, double limitValue) {
     this.target = target;
     this.targetValue = targetValue;
-    this.action = action;
-    this.actionValue = actionValue;
+    this.limit = limit;
+    this.limitValue = limitValue;
   }
 
   public Target target() {
@@ -94,12 +94,12 @@ public class Quota {
     return targetValue;
   }
 
-  public Action action() {
-    return action;
+  public Limit limit() {
+    return limit;
   }
 
-  public double actionValue() {
-    return actionValue;
+  public double limitValue() {
+    return limitValue;
   }
 
   @Override
@@ -107,15 +107,15 @@ public class Quota {
     if (this == o) return true;
     if (o == null || getClass() != o.getClass()) return false;
     Quota quota = (Quota) o;
-    return Double.compare(quota.actionValue, actionValue) == 0
+    return Double.compare(quota.limitValue, limitValue) == 0
         && target == quota.target
         && Objects.equals(targetValue, quota.targetValue)
-        && action == quota.action;
+        && limit == quota.limit;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(target, targetValue, action, actionValue);
+    return Objects.hash(target, targetValue, limit, limitValue);
   }
 
   @Override
@@ -126,10 +126,10 @@ public class Quota {
         + ", targetValue='"
         + targetValue
         + '\''
-        + ", action="
-        + action
-        + ", actionValue="
-        + actionValue
+        + ", limit="
+        + limit
+        + ", limitValue="
+        + limitValue
         + '}';
   }
 }
diff --git a/app/src/main/java/org/astraea/consumer/Builder.java b/app/src/main/java/org/astraea/consumer/Builder.java
index c609657b..cb17c699 100644
--- a/app/src/main/java/org/astraea/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/consumer/Builder.java
@@ -20,6 +20,7 @@ public class Builder<Key, Value> {
   private Deserializer<?> valueDeserializer = Deserializer.BYTE_ARRAY;
   private final Set<String> topics = new HashSet<>();
   private ConsumerRebalanceListener listener = ignore -> {};
+  private int distanceFromLatest = -1;
 
   Builder() {}
 
@@ -45,6 +46,18 @@ public class Builder<Key, Value> {
     return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
   }
 
+  /**
+   * set the offset to read from the latest offset. For example, the end offset is 5, and you set
+   * distanceFromLatest to 2, then you will read data from offset: 3
+   *
+   * @param distanceFromLatest the distance from the latest offset
+   * @return this builder
+   */
+  public Builder<Key, Value> distanceFromLatest(int distanceFromLatest) {
+    this.distanceFromLatest = distanceFromLatest;
+    return this;
+  }
+
   public Builder<Key, Value> topics(Set<String> topics) {
     this.topics.addAll(topics);
     return this;
@@ -95,6 +108,18 @@ public class Builder<Key, Value> {
             Deserializer.of((Deserializer<Key>) keyDeserializer),
             Deserializer.of((Deserializer<Value>) valueDeserializer));
     kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
+
+    // this mode is not supported by kafka, so we have to calculate the offset first
+    if (distanceFromLatest > 0) {
+      // 1) poll data until the assignment is completed
+      while (kafkaConsumer.assignment().isEmpty()) kafkaConsumer.poll(Duration.ofMillis(500));
+      var partitions = kafkaConsumer.assignment();
+      // 2) get the end offsets from all subscribed partitions
+      var endOffsets = kafkaConsumer.endOffsets(partitions);
+      // 3) calculate and then seek to the correct offset (end offset - recent offset)
+      endOffsets.forEach(
+          (tp, latest) -> kafkaConsumer.seek(tp, Math.max(0, latest - distanceFromLatest)));
+    }
     return new Consumer<>() {
       @Override
       public Collection<Record<Key, Value>> poll(int recordCount, Duration timeout) {
diff --git a/app/src/main/java/org/astraea/cost/MemoryWarningCost.java b/app/src/main/java/org/astraea/cost/MemoryWarningCost.java
deleted file mode 100644
index 51521315..00000000
--- a/app/src/main/java/org/astraea/cost/MemoryWarningCost.java
+++ /dev/null
@@ -1,48 +0,0 @@
-package org.astraea.cost;
-
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-import org.astraea.Utils;
-import org.astraea.metrics.collector.Fetcher;
-import org.astraea.metrics.java.HasJvmMemory;
-import org.astraea.metrics.kafka.KafkaMetrics;
-
-/** Warning on heap usage >= 80%. */
-public class MemoryWarningCost implements HasBrokerCost {
-  private final long constructTime = System.currentTimeMillis();
-
-  /** @return 1 if the heap usage >= 80%, 0 otherwise. */
-  @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var brokerScore =
-        clusterInfo.allBeans().entrySet().stream()
-            .map(
-                e ->
-                    Map.entry(
-                        e.getKey(),
-                        e.getValue().stream()
-                            .filter(beanObject -> beanObject instanceof HasJvmMemory)
-                            .map(beanObject -> (HasJvmMemory) beanObject)
-                            .map(
-                                hasJvmMemory -> {
-                                  if (Utils.overSecond(constructTime, 10)
-                                      && (hasJvmMemory.heapMemoryUsage().getUsed() + 0.0)
-                                              / (hasJvmMemory.heapMemoryUsage().getMax() + 1)
-                                          >= 0.8) {
-                                    return 1.0;
-                                  } else {
-                                    return 0.0;
-                                  }
-                                })
-                            .findAny()
-                            .orElseThrow()))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    return () -> brokerScore;
-  }
-
-  @Override
-  public Fetcher fetcher() {
-    return client -> List.of(KafkaMetrics.Host.jvmMemory(client));
-  }
-}
diff --git a/app/src/main/java/org/astraea/cost/Periodic.java b/app/src/main/java/org/astraea/cost/Periodic.java
index 34f33b65..03814ff8 100644
--- a/app/src/main/java/org/astraea/cost/Periodic.java
+++ b/app/src/main/java/org/astraea/cost/Periodic.java
@@ -10,43 +10,31 @@ public abstract class Periodic<Value> {
   private Value value;
 
   /**
-   * Updates the value each second.
+   * Updates the value interval second.
    *
    * @param updater Methods that need to be updated regularly.
+   * @param interval Required interval.
    * @return an object of type Value created from the parameter value.
    */
-  protected Value tryUpdate(Supplier<Value> updater) {
-    if (Utils.overSecond(lastUpdate, 1)) {
+  protected Value tryUpdate(Supplier<Value> updater, Duration interval) {
+    if (Utils.isExpired(lastUpdate, interval)) {
       value = updater.get();
       lastUpdate = currentTime();
     }
     return value;
   }
 
-  protected long currentTime() {
-    return System.currentTimeMillis();
-  }
-
   /**
-   * Updates the value interval second.
+   * Updates the value each second.
    *
    * @param updater Methods that need to be updated regularly.
-   * @param interval Required interval.
    * @return an object of type Value created from the parameter value.
    */
-  protected Value tryUpdate(Supplier<Value> updater, int interval) {
-    if (Utils.overSecond(lastUpdate, interval)) {
-      value = updater.get();
-      lastUpdate = currentTime();
-    }
-    return value;
+  protected Value tryUpdateAfterOneSecond(Supplier<Value> updater) {
+    return tryUpdate(updater, Duration.ofSeconds(1));
   }
 
-  protected Value tryUpdate(Supplier<Value> updater, Duration interval) {
-    if (Utils.overSecond(lastUpdate, interval)) {
-      value = updater.get();
-      lastUpdate = currentTime();
-    }
-    return value;
+  protected long currentTime() {
+    return System.currentTimeMillis();
   }
 }
diff --git a/app/src/main/java/org/astraea/cost/broker/BrokerInputCost.java b/app/src/main/java/org/astraea/cost/broker/BrokerInputCost.java
index 6e3ef9ee..721a8ad2 100644
--- a/app/src/main/java/org/astraea/cost/broker/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/cost/broker/BrokerInputCost.java
@@ -10,7 +10,6 @@ import java.util.stream.IntStream;
 import org.astraea.cost.BrokerCost;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.HasBrokerCost;
-import org.astraea.cost.Periodic;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
 import org.astraea.metrics.kafka.KafkaMetrics;
@@ -26,44 +25,44 @@ import org.astraea.metrics.kafka.KafkaMetrics;
  *   <li>The final result is the average of the ten-second data.
  * </ol>
  */
-public class BrokerInputCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+public class BrokerInputCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var brokerScore =
-        tryUpdate(
-            () -> {
-              var costMetrics =
-                  clusterInfo.allBeans().entrySet().stream()
-                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
-              clusterInfo
-                  .allBeans()
-                  .forEach(
-                      (brokerID, value) -> {
-                        if (!brokersMetric.containsKey(brokerID)) {
-                          brokersMetric.put(brokerID, new BrokerMetric());
-                        }
-                        value.stream()
+    var costMetrics =
+        clusterInfo.allBeans().entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry ->
+                        entry.getValue().stream()
                             .filter(
                                 hasBeanObject ->
                                     KafkaMetrics.BrokerTopic.BytesInPerSec.metricName()
                                         .equals(
                                             hasBeanObject.beanObject().getProperties().get("name")))
-                            .forEach(
-                                hasBeanObject -> {
-                                  var broker = brokersMetric.get(brokerID);
-                                  var inBean = (BrokerTopicMetricsResult) hasBeanObject;
-                                  costMetrics.put(
-                                      brokerID, (double) (inBean.count() - broker.accumulateCount));
-                                  broker.accumulateCount = inBean.count();
-                                });
-                      });
-              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+                            .findAny()
+                            .orElseThrow()))
+            .entrySet()
+            .stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry -> {
+                      if (!brokersMetric.containsKey(entry.getKey())) {
+                        brokersMetric.put(entry.getKey(), new BrokerMetric());
+                      }
+                      var broker = brokersMetric.get(entry.getKey());
+                      var inBean = (BrokerTopicMetricsResult) entry.getValue();
+                      var count = (double) (inBean.count() - broker.accumulateCount);
+                      broker.accumulateCount = inBean.count();
+                      return count;
+                    }));
 
-              return computeLoad();
-            });
-    return () -> brokerScore;
+    TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+
+    return this::computeLoad;
   }
 
   Map<Integer, Double> computeLoad() {
diff --git a/app/src/main/java/org/astraea/cost/broker/BrokerOutputCost.java b/app/src/main/java/org/astraea/cost/broker/BrokerOutputCost.java
index a175cdd9..58bc6888 100644
--- a/app/src/main/java/org/astraea/cost/broker/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/cost/broker/BrokerOutputCost.java
@@ -10,7 +10,6 @@ import java.util.stream.IntStream;
 import org.astraea.cost.BrokerCost;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.HasBrokerCost;
-import org.astraea.cost.Periodic;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
 import org.astraea.metrics.kafka.KafkaMetrics;
@@ -26,44 +25,44 @@ import org.astraea.metrics.kafka.KafkaMetrics;
  *   <li>The final result is the average of the ten-second data.
  * </ol>
  */
-public class BrokerOutputCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+public class BrokerOutputCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var brokerScore =
-        tryUpdate(
-            () -> {
-              var costMetrics =
-                  clusterInfo.allBeans().entrySet().stream()
-                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
-              clusterInfo
-                  .allBeans()
-                  .forEach(
-                      (brokerID, value) -> {
-                        if (!brokersMetric.containsKey(brokerID)) {
-                          brokersMetric.put(brokerID, new BrokerMetric());
-                        }
-                        value.stream()
+    var costMetrics =
+        clusterInfo.allBeans().entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry ->
+                        entry.getValue().stream()
                             .filter(
                                 hasBeanObject ->
                                     KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName()
                                         .equals(
                                             hasBeanObject.beanObject().getProperties().get("name")))
-                            .forEach(
-                                hasBeanObject -> {
-                                  var broker = brokersMetric.get(brokerID);
-                                  var outBean = (BrokerTopicMetricsResult) hasBeanObject;
-                                  costMetrics.put(
-                                      brokerID,
-                                      (double) (outBean.count() - broker.accumulateCount));
-                                  broker.accumulateCount = outBean.count();
-                                });
-                      });
-              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
-              return computeLoad();
-            });
-    return () -> brokerScore;
+                            .findAny()
+                            .orElseThrow()))
+            .entrySet()
+            .stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry -> {
+                      if (!brokersMetric.containsKey(entry.getKey())) {
+                        brokersMetric.put(entry.getKey(), new BrokerMetric());
+                      }
+                      var broker = brokersMetric.get(entry.getKey());
+                      var inBean = (BrokerTopicMetricsResult) entry.getValue();
+                      var count = (double) (inBean.count() - broker.accumulateCount);
+                      broker.accumulateCount = inBean.count();
+                      return count;
+                    }));
+
+    TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+
+    return this::computeLoad;
   }
 
   Map<Integer, Double> computeLoad() {
diff --git a/app/src/main/java/org/astraea/cost/broker/CpuCost.java b/app/src/main/java/org/astraea/cost/broker/CpuCost.java
index 7639f6f8..fa45fe8e 100644
--- a/app/src/main/java/org/astraea/cost/broker/CpuCost.java
+++ b/app/src/main/java/org/astraea/cost/broker/CpuCost.java
@@ -10,7 +10,6 @@ import java.util.stream.IntStream;
 import org.astraea.cost.BrokerCost;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.HasBrokerCost;
-import org.astraea.cost.Periodic;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.java.OperatingSystemInfo;
 import org.astraea.metrics.kafka.KafkaMetrics;
@@ -26,36 +25,37 @@ import org.astraea.metrics.kafka.KafkaMetrics;
  *   <li>The final result is the average of the ten-second data.
  * </ol>
  */
-public class CpuCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+public class CpuCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var brokerScore =
-        tryUpdate(
-            () -> {
-              var costMetrics =
-                  clusterInfo.allBeans().entrySet().stream()
-                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
-              clusterInfo
-                  .allBeans()
-                  .forEach(
-                      (brokerID, value) -> {
-                        if (!brokersMetric.containsKey(brokerID)) {
-                          brokersMetric.put(brokerID, new BrokerMetric());
-                        }
-                        value.stream()
-                            .filter(beanObject -> beanObject instanceof OperatingSystemInfo)
-                            .forEach(
-                                hasBeanObject -> {
-                                  var cpuBean = (OperatingSystemInfo) hasBeanObject;
-                                  costMetrics.put(brokerID, cpuBean.systemCpuLoad());
-                                });
-                      });
-              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
-              return computeLoad();
-            });
-    return () -> brokerScore;
+    var costMetrics =
+        clusterInfo.allBeans().entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry ->
+                        entry.getValue().stream()
+                            .filter(hasBeanObject -> hasBeanObject instanceof OperatingSystemInfo)
+                            .findAny()
+                            .orElseThrow()))
+            .entrySet()
+            .stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry -> {
+                      if (!brokersMetric.containsKey(entry.getKey())) {
+                        brokersMetric.put(entry.getKey(), new BrokerMetric());
+                      }
+                      var cpuBean = (OperatingSystemInfo) entry.getValue();
+                      return cpuBean.systemCpuLoad();
+                    }));
+
+    TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+
+    return this::computeLoad;
   }
 
   Map<Integer, Double> computeLoad() {
diff --git a/app/src/main/java/org/astraea/cost/broker/MemoryCost.java b/app/src/main/java/org/astraea/cost/broker/MemoryCost.java
index 8d18ae49..91b05219 100644
--- a/app/src/main/java/org/astraea/cost/broker/MemoryCost.java
+++ b/app/src/main/java/org/astraea/cost/broker/MemoryCost.java
@@ -31,34 +31,33 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
    */
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var brokerScore =
-        tryUpdate(
-            () -> {
-              var costMetrics =
-                  clusterInfo.allBeans().entrySet().stream()
-                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
-              clusterInfo
-                  .allBeans()
-                  .forEach(
-                      (brokerID, value) -> {
-                        if (!brokersMetric.containsKey(brokerID)) {
-                          brokersMetric.put(brokerID, new BrokerMetric());
-                        }
-                        value.stream()
-                            .filter(beanObject -> beanObject instanceof HasJvmMemory)
-                            .forEach(
-                                hasBeanObject -> {
-                                  var jvmBean = (HasJvmMemory) hasBeanObject;
-                                  costMetrics.put(
-                                      brokerID,
-                                      ((jvmBean.heapMemoryUsage().getUsed() + 0.0)
-                                          / (jvmBean.heapMemoryUsage().getMax() + 1)));
-                                });
-                      });
-              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
-              return computeLoad();
-            });
-    return () -> brokerScore;
+    var costMetrics =
+        clusterInfo.allBeans().entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry ->
+                        entry.getValue().stream()
+                            .filter(hasBeanObject -> hasBeanObject instanceof HasJvmMemory)
+                            .findAny()
+                            .orElseThrow()))
+            .entrySet()
+            .stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry -> {
+                      if (!brokersMetric.containsKey(entry.getKey())) {
+                        brokersMetric.put(entry.getKey(), new BrokerMetric());
+                      }
+                      var jvmBean = (HasJvmMemory) entry.getValue();
+                      return (jvmBean.heapMemoryUsage().getUsed() + 0.0)
+                          / (jvmBean.heapMemoryUsage().getMax() + 1);
+                    }));
+
+    TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+
+    return this::computeLoad;
   }
 
   Map<Integer, Double> computeLoad() {
diff --git a/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java
index c01f708b..10ebba84 100644
--- a/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java
@@ -9,7 +9,6 @@ import org.astraea.cost.BrokerCost;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.CostFunction;
 import org.astraea.cost.HasBrokerCost;
-import org.astraea.cost.Periodic;
 import org.astraea.metrics.collector.Fetcher;
 
 /**
@@ -19,7 +18,7 @@ import org.astraea.metrics.collector.Fetcher;
  * Because AHP is a subjective evaluation method and entropy method is an objective evaluation
  * method, so I named it NeutralIntegratedCost.
  */
-public class NeutralIntegratedCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+public class NeutralIntegratedCost implements HasBrokerCost {
   private final List<HasBrokerCost> metricsCost =
       List.of(new BrokerInputCost(), new BrokerOutputCost(), new CpuCost(), new MemoryCost());
   private final Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
@@ -28,75 +27,64 @@ public class NeutralIntegratedCost extends Periodic<Map<Integer, Double>> implem
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var score =
-        tryUpdate(
-            () -> {
-              var costMetrics =
-                  clusterInfo.allBeans().entrySet().stream()
-                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
-              costMetrics.forEach(
-                  (key, value) -> {
-                    if (!brokersMetric.containsKey(key)) {
-                      brokersMetric.put(key, new BrokerMetrics());
-                    }
-                  });
-
-              metricsCost.forEach(
-                  hasBrokerCost -> {
-                    if (hasBrokerCost instanceof BrokerInputCost) {
-                      hasBrokerCost
-                          .brokerCost(clusterInfo)
-                          .value()
-                          .forEach(
-                              (brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
-                    } else if (hasBrokerCost instanceof BrokerOutputCost) {
-                      hasBrokerCost
-                          .brokerCost(clusterInfo)
-                          .value()
-                          .forEach(
-                              (brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
-                    } else if (hasBrokerCost instanceof CpuCost) {
-                      hasBrokerCost
-                          .brokerCost(clusterInfo)
-                          .value()
-                          .forEach(
-                              (brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
-                    } else if (hasBrokerCost instanceof MemoryCost) {
-                      hasBrokerCost
-                          .brokerCost(clusterInfo)
-                          .value()
-                          .forEach(
-                              (brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
-                    }
-                  });
-
-              var integratedEmpowerment =
-                  entropyEmpowerment.empowerment(brokersMetric).entrySet().stream()
-                      .collect(
-                          Collectors.toMap(
-                              Map.Entry::getKey,
-                              entry ->
-                                  entry.getValue() * 0.5
-                                      + ahpEmpowerment.empowerment().get(entry.getKey()) * 0.5));
-
-              return brokersMetric.entrySet().stream()
-                  .collect(
-                      Collectors.toMap(
-                          Map.Entry::getKey,
-                          entry ->
-                              entry.getValue().inputScore
-                                      * integratedEmpowerment.get(
-                                          Metrics.inputThroughput.metricName)
-                                  + entry.getValue().outputScore
-                                      * integratedEmpowerment.get(
-                                          Metrics.outputThroughput.metricName)
-                                  + entry.getValue().cpuScore
-                                      * integratedEmpowerment.get(Metrics.cpu.metricName)
-                                  + entry.getValue().memoryScore
-                                      * integratedEmpowerment.get(Metrics.memory.metricName)));
-            });
-
-    return () -> score;
+    var costMetrics =
+        clusterInfo.allBeans().entrySet().stream()
+            .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
+    costMetrics.forEach(
+        (key, value) -> {
+          if (!brokersMetric.containsKey(key)) {
+            brokersMetric.put(key, new BrokerMetrics());
+          }
+        });
+
+    metricsCost.forEach(
+        hasBrokerCost -> {
+          if (hasBrokerCost instanceof BrokerInputCost) {
+            hasBrokerCost
+                .brokerCost(clusterInfo)
+                .value()
+                .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
+          } else if (hasBrokerCost instanceof BrokerOutputCost) {
+            hasBrokerCost
+                .brokerCost(clusterInfo)
+                .value()
+                .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
+          } else if (hasBrokerCost instanceof CpuCost) {
+            hasBrokerCost
+                .brokerCost(clusterInfo)
+                .value()
+                .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
+          } else if (hasBrokerCost instanceof MemoryCost) {
+            hasBrokerCost
+                .brokerCost(clusterInfo)
+                .value()
+                .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
+          }
+        });
+
+    var integratedEmpowerment =
+        entropyEmpowerment.empowerment(brokersMetric).entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry ->
+                        entry.getValue() * 0.5
+                            + ahpEmpowerment.empowerment().get(entry.getKey()) * 0.5));
+
+    return () ->
+        brokersMetric.entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry ->
+                        entry.getValue().inputScore
+                                * integratedEmpowerment.get(Metrics.inputThroughput.metricName)
+                            + entry.getValue().outputScore
+                                * integratedEmpowerment.get(Metrics.outputThroughput.metricName)
+                            + entry.getValue().cpuScore
+                                * integratedEmpowerment.get(Metrics.cpu.metricName)
+                            + entry.getValue().memoryScore
+                                * integratedEmpowerment.get(Metrics.memory.metricName)));
   }
 
   @Override
diff --git a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeight.java b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeight.java
deleted file mode 100644
index e049650b..00000000
--- a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeight.java
+++ /dev/null
@@ -1,47 +0,0 @@
-package org.astraea.partitioner.smooth;
-
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-
-/**
- * Given initial key-score pair, it will output a preferred key with higher score. The score of the
- * chosen key will decrease. And the other score will increment by its original score. It may result
- * in "higher score with higher chosen rate".
- */
-public final class SmoothWeight {
-  private Map<Integer, Integer> load;
-  private int loadSum;
-  public Map<Integer, Integer> weight;
-
-  public SmoothWeight() {}
-
-  public SmoothWeight(Map<Integer, Integer> load) {
-    init(load);
-  }
-
-  public synchronized void init(Map<Integer, Integer> load) {
-    this.load = new HashMap<>(load);
-    this.weight = new HashMap<>(load);
-    this.loadSum = load.values().stream().mapToInt(d -> d).sum();
-  }
-
-  /**
-   * Get the preferred ID, and update the state.
-   *
-   * @return the preferred ID
-   */
-  public int getAndChoose() {
-    Objects.requireNonNull(this.weight);
-
-    this.weight.replaceAll((ID, value) -> value + this.load.get(ID));
-    var maxID =
-        this.weight.entrySet().stream()
-            .max(Comparator.comparingDouble(Map.Entry::getValue))
-            .orElseGet(() -> Map.entry(0, 0))
-            .getKey();
-    this.weight.computeIfPresent(maxID, (ID, value) -> value - loadSum);
-    return maxID;
-  }
-}
diff --git a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightDispatcher.java b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightDispatcher.java
deleted file mode 100644
index a3247359..00000000
--- a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightDispatcher.java
+++ /dev/null
@@ -1,159 +0,0 @@
-package org.astraea.partitioner.smooth;
-
-import java.time.Duration;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.Optional;
-import java.util.Random;
-import java.util.TreeMap;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.stream.Collectors;
-import org.astraea.Utils;
-import org.astraea.cost.ClusterInfo;
-import org.astraea.cost.CostFunction;
-import org.astraea.cost.HasBrokerCost;
-import org.astraea.cost.LoadCost;
-import org.astraea.cost.MemoryWarningCost;
-import org.astraea.metrics.collector.BeanCollector;
-import org.astraea.metrics.collector.Fetcher;
-import org.astraea.metrics.collector.Receiver;
-import org.astraea.partitioner.Configuration;
-import org.astraea.partitioner.Dispatcher;
-
-/**
- * Based on the jmx metrics obtained from Kafka, it records the load status of the node over a
- * period of time. Predict the future status of each node through the poisson of the load status.
- * Finally, the result of poisson is used as the weight to perform smooth weighted RoundRobin.
- */
-public class SmoothWeightDispatcher implements Dispatcher {
-  public static final String JMX_PORT = "jmx.port";
-  private final BeanCollector beanCollector =
-      BeanCollector.builder().interval(Duration.ofSeconds(1)).build();
-  private Optional<Integer> jmxPortDefault = Optional.empty();
-  private final Map<Integer, Integer> jmxPorts = new TreeMap<>();
-  private final Map<Integer, Receiver> receivers = new TreeMap<>();
-
-  private final Collection<CostFunction> functions =
-      List.of(new LoadCost(), new MemoryWarningCost());
-  // Memory warning can dominate the score.
-  private final Map<Object, Double> costFraction =
-      Map.of(LoadCost.class, 1.0, MemoryWarningCost.class, -100.0);
-
-  private final SmoothWeight smoothWeightCal = new SmoothWeight();
-  private final Random random = new Random();
-  // Fetch data every n seconds
-  private long lastFetchTime = 0L;
-  private final Lock lock = new ReentrantLock();
-
-  @Override
-  public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
-    var partitionLeaders = clusterInfo.availablePartitionLeaders(topic);
-    // just return first partition if there is no available partitions
-    if (partitionLeaders.isEmpty()) return 0;
-
-    // just return the only one available partition
-    if (partitionLeaders.size() == 1) return partitionLeaders.iterator().next().partition();
-
-    if (Utils.overSecond(lastFetchTime, 1) && lock.tryLock()) {
-      try {
-        // add new receivers for new brokers
-        partitionLeaders.stream()
-            .filter(p -> !receivers.containsKey(p.nodeInfo().id()))
-            .forEach(
-                p ->
-                    receivers.put(
-                        p.nodeInfo().id(),
-                        receiver(p.nodeInfo().host(), jmxPort(p.nodeInfo().id()))));
-
-        // fetch the latest beans for each node
-        var beans =
-            receivers.entrySet().stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().current()));
-
-        var compoundScore =
-            functions.stream()
-                .filter(f -> f instanceof HasBrokerCost)
-                .map(f -> (HasBrokerCost) f)
-                .map(
-                    f -> {
-                      var map = f.brokerCost(ClusterInfo.of(clusterInfo, beans)).value();
-                      map.replaceAll((k, v) -> v * costFraction.get(f.getClass()));
-                      return map;
-                    })
-                .reduce(new HashMap<>(), SmoothWeightDispatcher::costCompound)
-                .entrySet()
-                .stream()
-                .map(e -> Map.entry(e.getKey(), (int) (double) e.getValue()))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
-        smoothWeightCal.init(compoundScore);
-      } finally {
-        lastFetchTime = System.currentTimeMillis();
-        lock.unlock();
-      }
-    }
-
-    // Make "smooth weight choosing" on the score.
-    var targetBroker = smoothWeightCal.getAndChoose();
-    var targetPartitions =
-        partitionLeaders.stream()
-            .filter(p -> p.nodeInfo().id() == targetBroker)
-            .collect(Collectors.toList());
-
-    return targetPartitions.get(random.nextInt(targetPartitions.size())).partition();
-  }
-
-  // Just add all cost function score together
-  static Map<Integer, Double> costCompound(
-      Map<Integer, Double> identity, Map<Integer, Double> cost) {
-    cost.forEach((id, value) -> identity.computeIfPresent(id, (k, v) -> v + value));
-    cost.forEach(identity::putIfAbsent);
-    return identity;
-  }
-
-  Receiver receiver(String host, int port) {
-    return beanCollector
-        .register()
-        .host(host)
-        .port(port)
-        .fetcher(
-            Fetcher.of(
-                functions.stream()
-                    .map(CostFunction::fetcher)
-                    .collect(Collectors.toUnmodifiableList())))
-        .build();
-  }
-
-  @Override
-  public void configure(Configuration config) {
-    jmxPortDefault = config.integer(JMX_PORT);
-
-    // seeks for custom jmx ports.
-    config.entrySet().stream()
-        .filter(e -> e.getKey().startsWith("broker."))
-        .filter(e -> e.getKey().endsWith(JMX_PORT))
-        .map(
-            e ->
-                Map.entry(
-                    e.getKey().replaceAll("^broker[.]", "").replaceAll("[.]" + JMX_PORT + "$", ""),
-                    e.getValue()))
-        .forEach(e -> jmxPorts.put(Integer.parseInt(e.getKey()), Integer.parseInt(e.getValue())));
-  }
-
-  // visible for testing
-  int jmxPort(int id) {
-    if (jmxPorts.containsKey(id)) return jmxPorts.get(id);
-    return jmxPortDefault.orElseThrow(
-        () -> new NoSuchElementException("broker: " + id + " does not have jmx port"));
-  }
-
-  @Override
-  public void close() {
-    receivers.values().forEach(Utils::close);
-    receivers.clear();
-  }
-}
diff --git a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
new file mode 100644
index 00000000..3dff8429
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
@@ -0,0 +1,175 @@
+package org.astraea.partitioner.smooth;
+
+import static org.astraea.partitioner.PartitionerUtils.partitionerConfig;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.kafka.common.Cluster;
+import org.astraea.Utils;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.NodeInfo;
+import org.astraea.cost.Periodic;
+import org.astraea.cost.ReplicaInfo;
+import org.astraea.cost.broker.NeutralIntegratedCost;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.collector.BeanCollector;
+import org.astraea.metrics.collector.Receiver;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.partitioner.Configuration;
+import org.astraea.partitioner.Dispatcher;
+
+public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Double>>
+    implements Dispatcher {
+  private final ConcurrentLinkedDeque<Integer> unusedPartitions = new ConcurrentLinkedDeque<>();
+  private final ConcurrentMap<String, BrokerNextCounter> topicCounter = new ConcurrentHashMap<>();
+  private final BeanCollector beanCollector =
+      BeanCollector.builder()
+          .interval(Duration.ofSeconds(1))
+          .numberOfObjectsPerNode(1)
+          .clientCreator(MBeanClient::jndi)
+          .build();
+  private final Optional<Integer> jmxPortDefault = Optional.empty();
+  private final Map<Integer, Integer> jmxPorts = new TreeMap<>();
+  private final Map<Integer, Receiver> receivers = new TreeMap<>();
+
+  private final Map<Integer, List<Integer>> hasPartitions = new TreeMap<>();
+
+  private SmoothWeightRoundRobin smoothWeightRoundRobinCal;
+
+  private final NeutralIntegratedCost neutralIntegratedCost = new NeutralIntegratedCost();
+
+  private Map<Integer, Collection<HasBeanObject>> beans;
+  private List<ReplicaInfo> partitions;
+
+  public static final String JMX_PORT = "jmx.port";
+
+  @Override
+  public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+    var targetPartition = unusedPartitions.poll();
+    tryUpdateAfterOneSecond(
+        () -> {
+          refreshPartitionMetaData(clusterInfo, topic);
+          // fetch the latest beans for each node
+          beans =
+              receivers.entrySet().stream()
+                  .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().current()));
+
+          var compoundScore =
+              neutralIntegratedCost.brokerCost(ClusterInfo.of(clusterInfo, beans)).value();
+
+          if (smoothWeightRoundRobinCal == null) {
+            smoothWeightRoundRobinCal = new SmoothWeightRoundRobin(compoundScore);
+          }
+          smoothWeightRoundRobinCal.init(compoundScore);
+
+          return compoundScore;
+        });
+    // just return first partition if there is no available partitions
+    if (partitions.isEmpty()) return 0;
+
+    // just return the only one available partition
+    if (partitions.size() == 1) return partitions.iterator().next().partition();
+
+    if (targetPartition == null) {
+      var targetBroker = smoothWeightRoundRobinCal.getAndChoose();
+      targetPartition =
+          hasPartitions
+              .get(targetBroker)
+              .get(
+                  nextValue(topic, clusterInfo, targetBroker)
+                      % hasPartitions.get(targetBroker).size());
+    }
+
+    return targetPartition;
+  }
+
+  @Override
+  public void close() {
+    receivers.values().forEach(Utils::close);
+    receivers.clear();
+  }
+
+  @Override
+  public void configure(Map<String, ?> configs) {
+    var properties = partitionerConfig(configs);
+    var config =
+        Configuration.of(
+            properties.entrySet().stream()
+                .collect(
+                    Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())));
+
+    // seeks for custom jmx ports.
+    config.entrySet().stream()
+        .filter(e -> e.getKey().startsWith("broker."))
+        .filter(e -> e.getKey().endsWith(JMX_PORT))
+        .forEach(
+            e ->
+                jmxPorts.put(
+                    Integer.parseInt(e.getKey().split("\\.")[1]), Integer.parseInt(e.getValue())));
+  }
+
+  @Override
+  public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+    unusedPartitions.add(prevPartition);
+  }
+
+  int jmxPort(int id) {
+    if (jmxPorts.containsKey(id)) return jmxPorts.get(id);
+    return jmxPortDefault.orElseThrow(
+        () -> new NoSuchElementException("broker: " + id + " does not have jmx port"));
+  }
+
+  Receiver receiver(String host, int port) {
+    return beanCollector
+        .register()
+        .host(host)
+        .port(port)
+        .fetcher(neutralIntegratedCost.fetcher())
+        .build();
+  }
+
+  private int nextValue(String topic, ClusterInfo clusterInfo, int targetBroker) {
+    return topicCounter
+        .computeIfAbsent(topic, k -> new BrokerNextCounter(clusterInfo))
+        .brokerCounter
+        .get(targetBroker)
+        .getAndIncrement();
+  }
+
+  private void refreshPartitionMetaData(ClusterInfo clusterInfo, String topic) {
+    partitions = clusterInfo.availablePartitions(topic);
+    partitions.forEach(
+        p ->
+            hasPartitions
+                .computeIfAbsent(p.nodeInfo().id(), k -> new ArrayList<>())
+                .add(p.partition()));
+
+    partitions.stream()
+        .filter(p -> !receivers.containsKey(p.nodeInfo().id()))
+        .forEach(
+            p ->
+                receivers.put(
+                    p.nodeInfo().id(), receiver(p.nodeInfo().host(), jmxPort(p.nodeInfo().id()))));
+  }
+
+  private static class BrokerNextCounter {
+    private final Map<Integer, AtomicInteger> brokerCounter;
+
+    BrokerNextCounter(ClusterInfo clusterInfo) {
+      brokerCounter =
+          clusterInfo.nodes().stream()
+              .collect(Collectors.toMap(NodeInfo::id, node -> new AtomicInteger(0)));
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/PostRequest.java b/app/src/main/java/org/astraea/web/PostRequest.java
index 962f0cc9..e3254cd2 100644
--- a/app/src/main/java/org/astraea/web/PostRequest.java
+++ b/app/src/main/java/org/astraea/web/PostRequest.java
@@ -6,40 +6,71 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 public interface PostRequest {
 
-  @SuppressWarnings("unchecked")
   static PostRequest of(HttpExchange exchange) throws IOException {
-    var jsonString = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
-    var raw =
-        ((Map<String, Object>) new Gson().fromJson(jsonString, Map.class))
+    return of(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8));
+  }
+
+  @SuppressWarnings("unchecked")
+  static PostRequest of(String json) {
+    return of(
+        ((Map<String, Object>) new Gson().fromJson(json, Map.class))
             .entrySet().stream()
                 .filter(e -> e.getValue() != null)
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toString()));
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> handleDouble(e.getValue()))));
+  }
+
+  static String handleDouble(Object obj) {
+    // the number in GSON is always DOUBLE>
+    if (obj instanceof Double) {
+      var value = (double) obj;
+      if (value - Math.floor(value) == 0) return String.valueOf((long) Math.floor(value));
+    }
+    return obj.toString();
+  }
+
+  static PostRequest of(Map<String, String> raw) {
     return () -> raw;
   }
 
   /** @return body represented by key-value */
   Map<String, String> raw();
 
+  default Optional<String> get(String key) {
+    return Optional.ofNullable(raw().get(key));
+  }
+
   default String value(String key) {
     var value = raw().get(key);
     if (value == null) throw new NoSuchElementException("the value for " + key + " is nonexistent");
     return value;
   }
 
+  default double doubleValue(String key, double defaultValue) {
+    return get(key).map(Double::parseDouble).orElse(defaultValue);
+  }
+
   default double doubleValue(String key) {
-    try {
-      return Double.parseDouble(value(key));
-    } catch (NumberFormatException e) {
-      throw new IllegalArgumentException(e);
-    }
+    return Double.parseDouble(value(key));
+  }
+
+  default int intValue(String key, int defaultValue) {
+    return get(key).map(Integer::parseInt).orElse(defaultValue);
   }
 
   default int intValue(String key) {
-    // the number in GSON is always DOUBLE>
-    return (int) doubleValue(key);
+    return Integer.parseInt(value(key));
+  }
+
+  default short shortValue(String key, short defaultValue) {
+    return get(key).map(Short::parseShort).orElse(defaultValue);
+  }
+
+  default short shortValue(String key) {
+    return Short.parseShort(value(key));
   }
 }
diff --git a/app/src/main/java/org/astraea/web/QuotaHandler.java b/app/src/main/java/org/astraea/web/QuotaHandler.java
new file mode 100644
index 00000000..fd7c73f9
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/QuotaHandler.java
@@ -0,0 +1,106 @@
+package org.astraea.web;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.astraea.admin.Admin;
+import org.astraea.admin.Quota;
+
+public class QuotaHandler implements Handler {
+
+  static final String IP_KEY = org.astraea.admin.Quota.Target.IP.nameOfKafka();
+  static final String CLIENT_ID_KEY = org.astraea.admin.Quota.Target.CLIENT_ID.nameOfKafka();
+  static final String CONNECTION_RATE_KEY =
+      org.astraea.admin.Quota.Limit.IP_CONNECTION_RATE.nameOfKafka();
+  static final String PRODUCE_RATE_KEY =
+      org.astraea.admin.Quota.Limit.PRODUCER_BYTE_RATE.nameOfKafka();
+  static final String CONSUME_RATE_KEY =
+      org.astraea.admin.Quota.Limit.CONSUMER_BYTE_RATE.nameOfKafka();
+
+  private final Admin admin;
+
+  QuotaHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  @Override
+  public Quotas get(Optional<String> target, Map<String, String> queries) {
+    if (queries.containsKey(IP_KEY))
+      return new Quotas(admin.quotas(org.astraea.admin.Quota.Target.IP, queries.get(IP_KEY)));
+    if (queries.containsKey(CLIENT_ID_KEY))
+      return new Quotas(
+          admin.quotas(org.astraea.admin.Quota.Target.CLIENT_ID, queries.get(CLIENT_ID_KEY)));
+    return new Quotas(admin.quotas());
+  }
+
+  @Override
+  public JsonObject post(PostRequest request) {
+    if (request.get(IP_KEY).isPresent()) {
+      admin
+          .quotaCreator()
+          .ip(request.value(IP_KEY))
+          .connectionRate(request.intValue(CONNECTION_RATE_KEY, Integer.MAX_VALUE))
+          .create();
+      return new Quotas(admin.quotas(org.astraea.admin.Quota.Target.IP, request.value(IP_KEY)));
+    }
+    if (request.get(CLIENT_ID_KEY).isPresent()) {
+      admin
+          .quotaCreator()
+          .clientId(request.value(CLIENT_ID_KEY))
+          .produceRate(request.intValue(PRODUCE_RATE_KEY, Integer.MAX_VALUE))
+          .consumeRate(request.intValue(CONSUME_RATE_KEY, Integer.MAX_VALUE))
+          .create();
+      return new Quotas(
+          admin.quotas(org.astraea.admin.Quota.Target.CLIENT_ID, request.value(CLIENT_ID_KEY)));
+    }
+    return ErrorObject.for404("You must define either " + CLIENT_ID_KEY + " or " + IP_KEY);
+  }
+
+  static class Target implements JsonObject {
+    final String name;
+    final String value;
+
+    Target(String name, String value) {
+      this.name = name;
+      this.value = value;
+    }
+  }
+
+  static class Limit implements JsonObject {
+    final String name;
+    final double value;
+
+    Limit(String name, double value) {
+      this.name = name;
+      this.value = value;
+    }
+  }
+
+  static class Quota implements JsonObject {
+    final Target target;
+    final Limit limit;
+
+    public Quota(org.astraea.admin.Quota quota) {
+      this(
+          quota.target().nameOfKafka(),
+          quota.targetValue(),
+          quota.limit().nameOfKafka(),
+          quota.limitValue());
+    }
+
+    public Quota(String target, String targetValue, String limit, double limitValue) {
+      this.target = new Target(target, targetValue);
+      this.limit = new Limit(limit, limitValue);
+    }
+  }
+
+  static class Quotas implements JsonObject {
+    final List<Quota> quotas;
+
+    Quotas(Collection<org.astraea.admin.Quota> quotas) {
+      this.quotas = quotas.stream().map(Quota::new).collect(Collectors.toUnmodifiableList());
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/TopicHandler.java b/app/src/main/java/org/astraea/web/TopicHandler.java
index 210123e0..44e59c00 100644
--- a/app/src/main/java/org/astraea/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/web/TopicHandler.java
@@ -1,17 +1,23 @@
 package org.astraea.web;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.astraea.admin.Admin;
 import org.astraea.admin.Config;
 
 class TopicHandler implements Handler {
 
+  static final String TOPIC_NAME_KEY = "name";
+  static final String NUMBER_OF_PARTITIONS_KEY = "partitions";
+  static final String NUMBER_OF_REPLICAS_KEY = "replicas";
+
   private final Admin admin;
 
   TopicHandler(Admin admin) {
@@ -24,7 +30,11 @@ class TopicHandler implements Handler {
 
   @Override
   public JsonObject get(Optional<String> target, Map<String, String> queries) {
-    var topics = admin.topics(topicNames(target));
+    return get(topicNames(target));
+  }
+
+  private JsonObject get(Set<String> topicNames) {
+    var topics = admin.topics(topicNames);
     var replicas = admin.replicas(topics.keySet());
     var partitions =
         admin.offsets(topics.keySet()).entrySet().stream()
@@ -47,10 +57,39 @@ class TopicHandler implements Handler {
             .map(p -> new TopicInfo(p.getKey(), partitions.get(p.getKey()), p.getValue()))
             .collect(Collectors.toUnmodifiableList());
 
-    if (target.isPresent() && topicInfos.size() == 1) return topicInfos.get(0);
+    if (topicNames.size() == 1 && topicInfos.size() == 1) return topicInfos.get(0);
     return new Topics(topicInfos);
   }
 
+  static Map<String, String> remainingConfigs(PostRequest request) {
+    var configs = new HashMap<>(request.raw());
+    configs.remove(TOPIC_NAME_KEY);
+    configs.remove(NUMBER_OF_PARTITIONS_KEY);
+    configs.remove(NUMBER_OF_REPLICAS_KEY);
+    return configs;
+  }
+
+  @Override
+  public JsonObject post(PostRequest request) {
+    admin
+        .creator()
+        .topic(request.value(TOPIC_NAME_KEY))
+        .numberOfPartitions(request.intValue(NUMBER_OF_PARTITIONS_KEY, 1))
+        .numberOfReplicas(request.shortValue(NUMBER_OF_REPLICAS_KEY, (short) 1))
+        .configs(remainingConfigs(request))
+        .create();
+    if (admin.topicNames().contains(request.value(TOPIC_NAME_KEY))) {
+      try {
+        // if the topic creation is synced, we return the details.
+        return get(Set.of(request.value(TOPIC_NAME_KEY)));
+      } catch (UnknownTopicOrPartitionException e) {
+        // swallow
+      }
+    }
+    // Otherwise, return only name
+    return new TopicInfo(request.value(TOPIC_NAME_KEY), List.of(), Map.of());
+  }
+
   static class Topics implements JsonObject {
     final Collection<TopicInfo> topics;
 
@@ -65,11 +104,17 @@ class TopicHandler implements Handler {
     final Map<String, String> configs;
 
     private TopicInfo(String name, List<Partition> partitions, Config configs) {
+      this(
+          name,
+          partitions,
+          StreamSupport.stream(configs.spliterator(), false)
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+    }
+
+    private TopicInfo(String name, List<Partition> partitions, Map<String, String> configs) {
       this.name = name;
       this.partitions = partitions;
-      this.configs =
-          StreamSupport.stream(configs.spliterator(), false)
-              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+      this.configs = configs;
     }
   }
 
@@ -88,13 +133,13 @@ class TopicHandler implements Handler {
   }
 
   static class Replica implements JsonObject {
-    private final int broker;
-    private final long lag;
-    private final long size;
-    private final boolean leader;
-    private final boolean inSync;
-    private final boolean isFuture;
-    private final String path;
+    final int broker;
+    final long lag;
+    final long size;
+    final boolean leader;
+    final boolean inSync;
+    final boolean isFuture;
+    final String path;
 
     Replica(org.astraea.admin.Replica replica) {
       this(
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
index a3dd08d6..d0821feb 100644
--- a/app/src/main/java/org/astraea/web/WebService.java
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -19,6 +19,7 @@ public class WebService {
     server.createContext("/groups", new GroupHandler(Admin.of(arg.configs())));
     server.createContext("/brokers", new BrokerHandler(Admin.of(arg.configs())));
     server.createContext("/producers", new ProducerHandler(Admin.of(arg.configs())));
+    server.createContext("/quotas", new QuotaHandler(Admin.of(arg.configs())));
     server.start();
   }
 
diff --git a/docker/start_kafka_tool.sh b/docker/start_kafka_tool.sh
index 8ea38bc4..1678b132 100755
--- a/docker/start_kafka_tool.sh
+++ b/docker/start_kafka_tool.sh
@@ -71,8 +71,10 @@ function runContainer() {
   echo "JMX address: $ADDRESS:$JMX_PORT"
 
   # web service needs to bind a port to expose Restful APIs, so we have to open a port of container
-  need_to_bind_web=""
+  local need_to_bind_web=""
+  local background=""
   if [[ "$args" == web* ]]; then
+    background="-d"
     sentence=($args)
     defined_port="false"
     # use random port by default
@@ -96,6 +98,7 @@ function runContainer() {
   fi
 
   docker run --rm --init \
+    $background \
     -p $JMX_PORT:$JMX_PORT \
     $need_to_bind_web \
     "$IMAGE_NAME" \


Test patch:
diff --git a/app/src/test/java/org/astraea/UtilsTest.java b/app/src/test/java/org/astraea/UtilsTest.java
new file mode 100644
index 00000000..57ef6415
--- /dev/null
+++ b/app/src/test/java/org/astraea/UtilsTest.java
@@ -0,0 +1,36 @@
+package org.astraea;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class UtilsTest {
+
+  @Test
+  void testHandleException() {
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            Utils.handleException(
+                () -> {
+                  throw new ExecutionException(new IllegalArgumentException());
+                }));
+
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            Utils.handleException(
+                () -> {
+                  throw new IllegalArgumentException();
+                }));
+
+    Assertions.assertThrows(
+        RuntimeException.class,
+        () ->
+            Utils.handleException(
+                () -> {
+                  throw new IOException();
+                }));
+  }
+}
diff --git a/app/src/test/java/org/astraea/admin/AdminTest.java b/app/src/test/java/org/astraea/admin/AdminTest.java
index 572ba7cd..cfef8718 100644
--- a/app/src/test/java/org/astraea/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/admin/AdminTest.java
@@ -428,10 +428,10 @@ public class AdminTest extends RequireBrokerCluster {
                 Set.of(Quota.Target.IP),
                 quotas.stream().map(Quota::target).collect(Collectors.toSet()));
             Assertions.assertEquals(
-                Set.of(Quota.Action.IP_CONNECTION_RATE),
-                quotas.stream().map(Quota::action).collect(Collectors.toSet()));
+                Set.of(Quota.Limit.IP_CONNECTION_RATE),
+                quotas.stream().map(Quota::limit).collect(Collectors.toSet()));
             Assertions.assertEquals("192.168.11.11", quotas.iterator().next().targetValue());
-            Assertions.assertEquals(10, quotas.iterator().next().actionValue());
+            Assertions.assertEquals(10, quotas.iterator().next().limitValue());
           };
 
       // only target
@@ -445,6 +445,17 @@ public class AdminTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testMultipleIpQuota() throws InterruptedException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.quotaCreator().ip("192.168.11.11").connectionRate(10).create();
+      admin.quotaCreator().ip("192.168.11.11").connectionRate(12).create();
+      admin.quotaCreator().ip("192.168.11.11").connectionRate(9).create();
+      TimeUnit.SECONDS.sleep(2);
+      Assertions.assertEquals(1, admin.quotas(Quota.Target.IP, "192.168.11.11").size());
+    }
+  }
+
   @Test
   void testClientQuota() throws InterruptedException {
     try (var admin = Admin.of(bootstrapServers())) {
@@ -458,33 +469,33 @@ public class AdminTest extends RequireBrokerCluster {
                 Set.of(Quota.Target.CLIENT_ID),
                 quotas.stream().map(Quota::target).collect(Collectors.toSet()));
             Assertions.assertEquals(
-                Set.of(Quota.Action.PRODUCER_BYTE_RATE, Quota.Action.CONSUMER_BYTE_RATE),
-                quotas.stream().map(Quota::action).collect(Collectors.toSet()));
+                Set.of(Quota.Limit.PRODUCER_BYTE_RATE, Quota.Limit.CONSUMER_BYTE_RATE),
+                quotas.stream().map(Quota::limit).collect(Collectors.toSet()));
             Assertions.assertEquals(
                 10,
                 quotas.stream()
-                    .filter(q -> q.action() == Quota.Action.PRODUCER_BYTE_RATE)
+                    .filter(q -> q.limit() == Quota.Limit.PRODUCER_BYTE_RATE)
                     .findFirst()
                     .get()
-                    .actionValue());
+                    .limitValue());
             Assertions.assertEquals(
                 "my-id",
                 quotas.stream()
-                    .filter(q -> q.action() == Quota.Action.PRODUCER_BYTE_RATE)
+                    .filter(q -> q.limit() == Quota.Limit.PRODUCER_BYTE_RATE)
                     .findFirst()
                     .get()
                     .targetValue());
             Assertions.assertEquals(
                 100,
                 quotas.stream()
-                    .filter(q -> q.action() == Quota.Action.CONSUMER_BYTE_RATE)
+                    .filter(q -> q.limit() == Quota.Limit.CONSUMER_BYTE_RATE)
                     .findFirst()
                     .get()
-                    .actionValue());
+                    .limitValue());
             Assertions.assertEquals(
                 "my-id",
                 quotas.stream()
-                    .filter(q -> q.action() == Quota.Action.CONSUMER_BYTE_RATE)
+                    .filter(q -> q.limit() == Quota.Limit.CONSUMER_BYTE_RATE)
                     .findFirst()
                     .get()
                     .targetValue());
@@ -500,4 +511,14 @@ public class AdminTest extends RequireBrokerCluster {
               .collect(Collectors.toUnmodifiableList()));
     }
   }
+
+  @Test
+  void testMultipleClientQuota() throws InterruptedException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.quotaCreator().clientId("my-id").consumeRate(100).create();
+      admin.quotaCreator().clientId("my-id").produceRate(999).create();
+      TimeUnit.SECONDS.sleep(2);
+      Assertions.assertEquals(2, admin.quotas(Quota.Target.CLIENT_ID, "my-id").size());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/admin/QuotaTest.java b/app/src/test/java/org/astraea/admin/QuotaTest.java
index 6ad7ee38..0a1ac33f 100644
--- a/app/src/test/java/org/astraea/admin/QuotaTest.java
+++ b/app/src/test/java/org/astraea/admin/QuotaTest.java
@@ -14,7 +14,7 @@ public class QuotaTest {
 
   @Test
   void testAction() {
-    Arrays.stream(Quota.Action.values())
-        .forEach(t -> Assertions.assertEquals(t, Quota.action(t.nameOfKafka())));
+    Arrays.stream(Quota.Limit.values())
+        .forEach(t -> Assertions.assertEquals(t, Quota.limit(t.nameOfKafka())));
   }
 }
diff --git a/app/src/test/java/org/astraea/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/consumer/ConsumerTest.java
index 52c66dfc..7f99c882 100644
--- a/app/src/test/java/org/astraea/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/consumer/ConsumerTest.java
@@ -7,6 +7,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
 import org.apache.kafka.common.errors.WakeupException;
+import org.astraea.Utils;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -117,4 +118,38 @@ public class ConsumerTest extends RequireBrokerCluster {
     // use different group id
     testConsumer.accept("another_group", 0);
   }
+
+  @Test
+  void testDistanceFromLatest() {
+    var count = 10;
+    var topic = Utils.randomString(10);
+    try (var producer = Producer.of(bootstrapServers())) {
+      IntStream.range(0, count)
+          .forEach(
+              i ->
+                  producer
+                      .sender()
+                      .topic(topic)
+                      .value(String.valueOf(count).getBytes(StandardCharsets.UTF_8))
+                      .run());
+      producer.flush();
+    }
+    try (var consumer =
+        Consumer.builder()
+            .bootstrapServers(bootstrapServers())
+            .topics(Set.of(topic))
+            .distanceFromLatest(3)
+            .build()) {
+      Assertions.assertEquals(3, consumer.poll(4, Duration.ofSeconds(5)).size());
+    }
+
+    try (var consumer =
+        Consumer.builder()
+            .bootstrapServers(bootstrapServers())
+            .topics(Set.of(topic))
+            .distanceFromLatest(1000)
+            .build()) {
+      Assertions.assertEquals(10, consumer.poll(11, Duration.ofSeconds(5)).size());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java b/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java
deleted file mode 100644
index 78cf5c16..00000000
--- a/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java
+++ /dev/null
@@ -1,67 +0,0 @@
-package org.astraea.cost;
-
-import java.lang.management.MemoryUsage;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import org.astraea.metrics.HasBeanObject;
-import org.astraea.metrics.java.HasJvmMemory;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-public class MemoryWarningCostTest {
-  @Test
-  void testCost() throws InterruptedException {
-    var jvmMemory1 = mockResult(50L, 100L);
-    var jvmMemory2 = mockResult(81L, 100L);
-    var jvmMemory3 = mockResult(80L, 150L);
-
-    Collection<HasBeanObject> broker1 = List.of(jvmMemory1);
-    Collection<HasBeanObject> broker2 = List.of(jvmMemory2);
-    Collection<HasBeanObject> broker3 = List.of(jvmMemory3);
-    ClusterInfo clusterInfo =
-        new FakeClusterInfo() {
-          @Override
-          public Map<Integer, Collection<HasBeanObject>> allBeans() {
-            return Map.of(1, broker1, 2, broker2, 3, broker3);
-          }
-
-          @Override
-          public Set<String> topics() {
-            return Set.of("t");
-          }
-
-          @Override
-          public List<ReplicaInfo> availablePartitions(String topic) {
-            return List.of(
-                ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
-                ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
-                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
-          }
-        };
-
-    var memoryWarning = new MemoryWarningCost();
-    var scores = memoryWarning.brokerCost(clusterInfo).value();
-    Assertions.assertEquals(0.0, scores.get(1));
-    Assertions.assertEquals(0.0, scores.get(2));
-    Assertions.assertEquals(0.0, scores.get(3));
-
-    Thread.sleep(10000);
-
-    scores = memoryWarning.brokerCost(clusterInfo).value();
-    Assertions.assertEquals(0.0, scores.get(1));
-    Assertions.assertEquals(1.0, scores.get(2));
-    Assertions.assertEquals(0.0, scores.get(3));
-  }
-
-  private HasJvmMemory mockResult(long used, long max) {
-    var jvmMemory = Mockito.mock(HasJvmMemory.class);
-    var memoryUsage = Mockito.mock(MemoryUsage.class);
-    Mockito.when(jvmMemory.heapMemoryUsage()).thenReturn(memoryUsage);
-    Mockito.when(memoryUsage.getUsed()).thenReturn(used);
-    Mockito.when(memoryUsage.getMax()).thenReturn(max);
-    return jvmMemory;
-  }
-}
diff --git a/app/src/test/java/org/astraea/cost/PeriodicTest.java b/app/src/test/java/org/astraea/cost/PeriodicTest.java
index 603fbe28..f071e4db 100644
--- a/app/src/test/java/org/astraea/cost/PeriodicTest.java
+++ b/app/src/test/java/org/astraea/cost/PeriodicTest.java
@@ -1,5 +1,6 @@
 package org.astraea.cost;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -10,13 +11,24 @@ public class PeriodicTest extends Periodic<Map<Integer, Double>> {
   private double brokerValue = 0.0;
 
   @Test
-  void testTryUpdate() {
-    var broker1 = tryUpdate(this::testMap);
+  void testTryUpdateAfterOneSecond() {
+    var broker1 = tryUpdateAfterOneSecond(this::testMap);
     Assertions.assertEquals(broker1.get(0), 0.0);
-    var broker2 = tryUpdate(this::testMap);
+    var broker2 = tryUpdateAfterOneSecond(this::testMap);
     Assertions.assertEquals(broker2.get(0), 0.0);
     sleep(1);
-    broker2 = tryUpdate(this::testMap);
+    broker2 = tryUpdateAfterOneSecond(this::testMap);
+    Assertions.assertEquals(broker2.get(0), 1.0);
+  }
+
+  @Test
+  void testTryUpdate() {
+    var broker1 = tryUpdate(this::testMap, Duration.ofSeconds(3));
+    Assertions.assertEquals(broker1.get(0), 0.0);
+    var broker2 = tryUpdate(this::testMap, Duration.ofSeconds(3));
+    Assertions.assertEquals(broker2.get(0), 0.0);
+    sleep(4);
+    broker2 = tryUpdate(this::testMap, Duration.ofSeconds(3));
     Assertions.assertEquals(broker2.get(0), 1.0);
   }
 
diff --git a/app/src/test/java/org/astraea/cost/broker/BrokerInputCostTest.java b/app/src/test/java/org/astraea/cost/broker/BrokerInputCostTest.java
index b1effd31..8302b4ae 100644
--- a/app/src/test/java/org/astraea/cost/broker/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/cost/broker/BrokerInputCostTest.java
@@ -15,7 +15,7 @@ import org.mockito.Mockito;
 
 public class BrokerInputCostTest {
   @Test
-  void testCost() throws InterruptedException {
+  void testCost() {
     ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
 
     var brokerInputCost = new BrokerInputCost();
@@ -23,8 +23,6 @@ public class BrokerInputCostTest {
     Assertions.assertEquals(0.47, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.39, scores.get(3));
-    var scores5 = brokerInputCost.brokerCost(clusterInfo).value();
-    Thread.sleep(1000);
 
     ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
     scores = brokerInputCost.brokerCost(clusterInfo2).value();
diff --git a/app/src/test/java/org/astraea/cost/broker/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/cost/broker/BrokerOutPutCostTest.java
index a8ca9a15..7f666090 100644
--- a/app/src/test/java/org/astraea/cost/broker/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/cost/broker/BrokerOutPutCostTest.java
@@ -24,8 +24,6 @@ public class BrokerOutPutCostTest {
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.39, scores.get(3));
 
-    Thread.sleep(1000);
-
     ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
     scores = brokerOutputCost.brokerCost(clusterInfo2).value();
     Assertions.assertEquals(0.55, scores.get(1));
diff --git a/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java b/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java
index 95760900..1415a87c 100644
--- a/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java
@@ -18,7 +18,7 @@ import org.mockito.Mockito;
 public class MemoryCostTest {
   @Test
   void testCost() throws InterruptedException {
-    var jvmMemory1 = mockResult(50L, 100L);
+    var jvmMemory1 = mockResult(40L, 100L);
     var jvmMemory2 = mockResult(81L, 100L);
     var jvmMemory3 = mockResult(80L, 150L);
 
@@ -48,12 +48,12 @@ public class MemoryCostTest {
 
     var memoryCost = new MemoryCost();
     var scores = memoryCost.brokerCost(clusterInfo).value();
-    Assertions.assertEquals(0.42, scores.get(1));
-    Assertions.assertEquals(0.64, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
+    Assertions.assertEquals(0.39, scores.get(1));
+    Assertions.assertEquals(0.63, scores.get(2));
+    Assertions.assertEquals(0.47, scores.get(3));
 
     Thread.sleep(1000);
-    jvmMemory1 = mockResult(50L, 50L);
+    jvmMemory1 = mockResult(50L, 150L);
     jvmMemory2 = mockResult(30L, 100L);
     jvmMemory3 = mockResult(80L, 150L);
 
@@ -81,9 +81,9 @@ public class MemoryCostTest {
           }
         };
     scores = memoryCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.53, scores.get(1));
+    Assertions.assertEquals(0.42, scores.get(1));
     Assertions.assertEquals(0.52, scores.get(2));
-    Assertions.assertEquals(0.45, scores.get(3));
+    Assertions.assertEquals(0.55, scores.get(3));
   }
 
   private HasJvmMemory mockResult(long used, long max) {
diff --git a/app/src/test/java/org/astraea/cost/broker/IntegratedScoreTest.java b/app/src/test/java/org/astraea/cost/broker/NeutralIntegratedCostTest.java
similarity index 95%
rename from app/src/test/java/org/astraea/cost/broker/IntegratedScoreTest.java
rename to app/src/test/java/org/astraea/cost/broker/NeutralIntegratedCostTest.java
index 079958bb..8145ed23 100644
--- a/app/src/test/java/org/astraea/cost/broker/IntegratedScoreTest.java
+++ b/app/src/test/java/org/astraea/cost/broker/NeutralIntegratedCostTest.java
@@ -4,7 +4,8 @@ import java.util.Map;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class IntegratedScoreTest {
+// TODO Add more test.
+public class NeutralIntegratedCostTest {
   @Test
   void testEntropyEmpowerment() {
     var neutralIntegratedCost = new NeutralIntegratedCost();
diff --git a/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
new file mode 100644
index 00000000..c0e84377
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
@@ -0,0 +1,286 @@
+package org.astraea.partitioner.smooth;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.astraea.admin.Admin;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
+import org.astraea.concurrent.ThreadPool;
+import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Deserializer;
+import org.astraea.consumer.Header;
+import org.astraea.producer.Producer;
+import org.astraea.producer.Serializer;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
+  private final String brokerList = bootstrapServers();
+  Admin admin = Admin.of(bootstrapServers());
+
+  private Properties initProConfig() {
+    Properties props = new Properties();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
+    props.put(
+        ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightRoundRobinDispatcher.class.getName());
+    props.put("producerID", 1);
+    var file =
+        new File(
+            SmoothWeightRoundRobinDispatchTest.class.getResource("").getPath()
+                + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write("jmx.port=" + jmxServiceURL().getPort() + "\n");
+      fileWriter.write("broker.0.jmx.port=" + jmxServiceURL().getPort() + "\n");
+      fileWriter.write("broker.1.jmx.port=" + jmxServiceURL().getPort() + "\n");
+      fileWriter.write("broker.2.jmx.port=" + jmxServiceURL().getPort());
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    props.put(
+        "partitioner.config",
+        SmoothWeightRoundRobinDispatchTest.class.getResource("").getPath()
+            + "PartitionerConfigTest");
+    return props;
+  }
+
+  @Test
+  void testPartitioner() {
+    var topicName = "address";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+    try (var producer =
+        Producer.builder()
+            .keySerializer(Serializer.STRING)
+            .configs(
+                initProConfig().entrySet().stream()
+                    .collect(
+                        Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())))
+            .build()) {
+      var i = 0;
+      while (i < 300) {
+        var metadata =
+            producer
+                .sender()
+                .topic(topicName)
+                .key(key)
+                .timestamp(timestamp)
+                .headers(List.of(header))
+                .run()
+                .toCompletableFuture()
+                .get();
+        assertEquals(topicName, metadata.topic());
+        assertEquals(timestamp, metadata.timestamp());
+        i++;
+      }
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+    sleep(1);
+    try (var consumer =
+        Consumer.builder()
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .topics(Set.of(topicName))
+            .keyDeserializer(Deserializer.STRING)
+            .build()) {
+      var records = consumer.poll(Duration.ofSeconds(20));
+      var recordsCount = records.size();
+      while (recordsCount < 300) {
+        recordsCount += consumer.poll(Duration.ofSeconds(20)).size();
+      }
+      assertEquals(300, recordsCount);
+      var record = records.iterator().next();
+      assertEquals(topicName, record.topic());
+      assertEquals("tainan", record.key());
+      assertEquals(1, record.headers().size());
+      var actualHeader = record.headers().iterator().next();
+      assertEquals(header.key(), actualHeader.key());
+      Assertions.assertArrayEquals(header.value(), actualHeader.value());
+    }
+  }
+
+  @Test
+  void testMultipleProducer() {
+    var topicName = "addr";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+    try (var threadPool =
+        ThreadPool.builder()
+            .executors(
+                IntStream.range(0, 10)
+                    .mapToObj(
+                        i ->
+                            producerExecutor(
+                                Producer.builder()
+                                    .keySerializer(Serializer.STRING)
+                                    .configs(
+                                        initProConfig().entrySet().stream()
+                                            .collect(
+                                                Collectors.toMap(
+                                                    e -> e.getKey().toString(),
+                                                    e -> e.getValue().toString())))
+                                    .build(),
+                                topicName,
+                                key,
+                                header,
+                                timestamp))
+                    .collect(Collectors.toUnmodifiableList()))
+            .build()) {
+      threadPool.waitAll();
+    }
+    try (var consumer =
+        Consumer.builder()
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .topics(Set.of(topicName))
+            .keyDeserializer(Deserializer.STRING)
+            .build()) {
+      var records = consumer.poll(Duration.ofSeconds(20));
+      var recordsCount = records.size();
+      while (recordsCount < 1000) {
+        recordsCount += consumer.poll(Duration.ofSeconds(20)).size();
+      }
+      assertEquals(1000, recordsCount);
+      var record = records.iterator().next();
+      assertEquals(topicName, record.topic());
+      assertEquals("tainan", record.key());
+      assertEquals(1, record.headers().size());
+      var actualHeader = record.headers().iterator().next();
+      assertEquals(header.key(), actualHeader.key());
+      Assertions.assertArrayEquals(header.value(), actualHeader.value());
+    }
+  }
+
+  @Test
+  void testJmxConfig() {
+    var props = initProConfig();
+    var file =
+        new File(
+            SmoothWeightRoundRobinDispatchTest.class.getResource("").getPath()
+                + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write(
+          "broker.0.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.1.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.2.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n");
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    var topicName = "addressN";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+
+    try (var producer =
+        Producer.builder()
+            .keySerializer(Serializer.STRING)
+            .configs(
+                props.entrySet().stream()
+                    .collect(
+                        Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())))
+            .build()) {
+      var metadata =
+          producer
+              .sender()
+              .topic(topicName)
+              .key(key)
+              .timestamp(timestamp)
+              .headers(List.of(header))
+              .run()
+              .toCompletableFuture()
+              .get();
+      assertEquals(topicName, metadata.topic());
+      assertEquals(timestamp, metadata.timestamp());
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+
+  private Executor producerExecutor(
+      Producer<String, byte[]> producer, String topic, String key, Header header, long timeStamp) {
+    return new Executor() {
+      int i = 0;
+
+      @Override
+      public State execute() throws InterruptedException {
+        if (i > 99) return State.DONE;
+        try {
+          producer
+              .sender()
+              .topic(topic)
+              .key(key)
+              .timestamp(timeStamp)
+              .headers(List.of(header))
+              .run()
+              .toCompletableFuture()
+              .get();
+        } catch (ExecutionException e) {
+          e.printStackTrace();
+        }
+        i++;
+        return State.RUNNING;
+      }
+
+      @Override
+      public void close() {
+        producer.close();
+      }
+    };
+  }
+
+  private static void sleep(int seconds) {
+    try {
+      TimeUnit.SECONDS.sleep(seconds);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Test
+  public void testGetAndChoose() {
+    var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0));
+
+    Assertions.assertEquals(1, smoothWeight.getAndChoose());
+    Assertions.assertEquals(2, smoothWeight.getAndChoose());
+    Assertions.assertEquals(3, smoothWeight.getAndChoose());
+    Assertions.assertEquals(1, smoothWeight.getAndChoose());
+    Assertions.assertEquals(2, smoothWeight.getAndChoose());
+    Assertions.assertEquals(3, smoothWeight.getAndChoose());
+    Assertions.assertEquals(1, smoothWeight.getAndChoose());
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightTest.java b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightTest.java
deleted file mode 100644
index 6f853448..00000000
--- a/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightTest.java
+++ /dev/null
@@ -1,20 +0,0 @@
-package org.astraea.partitioner.smooth;
-
-import java.util.Map;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-public class SmoothWeightTest {
-  @Test
-  public void testGetAndChoose() {
-    var smoothWeight = new SmoothWeight(Map.of(1, 5, 2, 3, 3, 1));
-
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
-    Assertions.assertEquals(2, smoothWeight.getAndChoose());
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
-    Assertions.assertEquals(2, smoothWeight.getAndChoose());
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
-    Assertions.assertEquals(3, smoothWeight.getAndChoose());
-  }
-}
diff --git a/app/src/test/java/org/astraea/web/PostRequestTest.java b/app/src/test/java/org/astraea/web/PostRequestTest.java
index cf3193be..6fe5dfaa 100644
--- a/app/src/test/java/org/astraea/web/PostRequestTest.java
+++ b/app/src/test/java/org/astraea/web/PostRequestTest.java
@@ -11,7 +11,7 @@ import org.mockito.Mockito;
 public class PostRequestTest {
 
   @Test
-  void testParse() throws IOException {
+  void testParseHttpExchange() throws IOException {
     var input =
         new ByteArrayInputStream("{\"a\":\"b\",\"c\":123}".getBytes(StandardCharsets.UTF_8));
     var exchange = Mockito.mock(HttpExchange.class);
@@ -22,4 +22,20 @@ public class PostRequestTest {
     Assertions.assertEquals("b", request.raw().get("a"));
     Assertions.assertEquals(123, request.intValue("c"));
   }
+
+  @Test
+  void testHandleDouble() {
+    Assertions.assertEquals("10", PostRequest.handleDouble(10.00));
+    Assertions.assertEquals("10.01", PostRequest.handleDouble(10.01));
+    Assertions.assertEquals("xxxx", PostRequest.handleDouble("xxxx"));
+  }
+
+  @Test
+  void testParseJson() {
+    var request = PostRequest.of("{\"a\":1234, \"b\":3.34}");
+    Assertions.assertEquals(1234, request.intValue("a"));
+    Assertions.assertEquals(1234, request.shortValue("a"));
+    Assertions.assertEquals(1234.0, request.shortValue("a"));
+    Assertions.assertEquals(3.34, request.doubleValue("b"));
+  }
 }
diff --git a/app/src/test/java/org/astraea/web/QuotaHandlerTest.java b/app/src/test/java/org/astraea/web/QuotaHandlerTest.java
new file mode 100644
index 00000000..3c15b375
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/QuotaHandlerTest.java
@@ -0,0 +1,76 @@
+package org.astraea.web;
+
+import java.util.Map;
+import java.util.Optional;
+import org.astraea.admin.Admin;
+import org.astraea.admin.Quota;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class QuotaHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testCreateQuota() {
+    var ip = "192.168.10.11";
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new QuotaHandler(admin);
+
+      var result =
+          Assertions.assertInstanceOf(
+              QuotaHandler.Quotas.class,
+              handler.post(
+                  PostRequest.of(
+                      Map.of(QuotaHandler.IP_KEY, ip, QuotaHandler.CONNECTION_RATE_KEY, "10"))));
+      Assertions.assertEquals(1, result.quotas.size());
+      Assertions.assertEquals(
+          Quota.Target.IP.nameOfKafka(), result.quotas.iterator().next().target.name);
+      Assertions.assertEquals(ip, result.quotas.iterator().next().target.value);
+      Assertions.assertEquals(
+          Quota.Limit.IP_CONNECTION_RATE.nameOfKafka(), result.quotas.iterator().next().limit.name);
+      Assertions.assertEquals(10, result.quotas.iterator().next().limit.value);
+    }
+  }
+
+  @Test
+  void testQuery() {
+    var ip0 = "192.168.10.11";
+    var ip1 = "192.168.10.12";
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new QuotaHandler(admin);
+
+      handler.post(
+          PostRequest.of(Map.of(QuotaHandler.IP_KEY, ip0, QuotaHandler.CONNECTION_RATE_KEY, "10")));
+      handler.post(
+          PostRequest.of(Map.of(QuotaHandler.IP_KEY, ip1, QuotaHandler.CONNECTION_RATE_KEY, "20")));
+      Assertions.assertEquals(
+          1, handler.get(Optional.empty(), Map.of(QuotaHandler.IP_KEY, ip0)).quotas.size());
+      Assertions.assertEquals(
+          1, handler.get(Optional.empty(), Map.of(QuotaHandler.IP_KEY, ip1)).quotas.size());
+    }
+  }
+
+  @Test
+  void testQueryNonexistentQuota() {
+    var ip = "192.168.10.11";
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new QuotaHandler(admin);
+      Assertions.assertEquals(
+          0,
+          Assertions.assertInstanceOf(
+                  QuotaHandler.Quotas.class,
+                  handler.get(Optional.empty(), Map.of(Quota.Target.IP.nameOfKafka(), "unknown")))
+              .quotas
+              .size());
+
+      Assertions.assertEquals(
+          0,
+          Assertions.assertInstanceOf(
+                  QuotaHandler.Quotas.class,
+                  handler.get(
+                      Optional.empty(), Map.of(Quota.Target.CLIENT_ID.nameOfKafka(), "unknown")))
+              .quotas
+              .size());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
index 0c563351..f5af95fb 100644
--- a/app/src/test/java/org/astraea/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
@@ -23,9 +23,16 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       var response =
           Assertions.assertInstanceOf(
               TopicHandler.Topics.class, handler.get(Optional.empty(), Map.of()));
-      Assertions.assertEquals(1, response.topics.size());
-      Assertions.assertEquals(topicName, response.topics.iterator().next().name);
-      Assertions.assertNotEquals(0, response.topics.iterator().next().configs.size());
+      Assertions.assertEquals(
+          1, response.topics.stream().filter(t -> t.name.equals(topicName)).count());
+      Assertions.assertNotEquals(
+          0,
+          response.topics.stream()
+              .filter(t -> t.name.equals(topicName))
+              .findFirst()
+              .get()
+              .configs
+              .size());
     }
   }
 
@@ -67,4 +74,75 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       Assertions.assertTrue(handler.topicNames(Optional.empty()).contains(topicName));
     }
   }
+
+  @Test
+  void testCreateTopic() {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new TopicHandler(admin);
+      var topicInfo =
+          Assertions.assertInstanceOf(
+              TopicHandler.TopicInfo.class,
+              handler.post(PostRequest.of(Map.of(TopicHandler.TOPIC_NAME_KEY, topicName))));
+      Assertions.assertEquals(topicName, topicInfo.name);
+    }
+  }
+
+  @Test
+  void testCreateTopicWithReplicas() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new TopicHandler(admin);
+      var topicInfo =
+          Assertions.assertInstanceOf(
+              TopicHandler.TopicInfo.class,
+              handler.post(
+                  PostRequest.of(
+                      Map.of(
+                          TopicHandler.TOPIC_NAME_KEY,
+                          topicName,
+                          TopicHandler.NUMBER_OF_PARTITIONS_KEY,
+                          "2",
+                          TopicHandler.NUMBER_OF_REPLICAS_KEY,
+                          "2",
+                          "segment.ms",
+                          "3000"))));
+      Assertions.assertEquals(topicName, topicInfo.name);
+
+      // the topic creation is not synced, so we have to wait the creation.
+      if (topicInfo.partitions.isEmpty()) {
+        TimeUnit.SECONDS.sleep(2);
+        var result = admin.replicas(Set.of(topicName));
+        Assertions.assertEquals(2, result.size());
+        result.values().forEach(replicas -> Assertions.assertEquals(2, replicas.size()));
+      } else {
+        Assertions.assertEquals(2, topicInfo.partitions.size());
+        Assertions.assertEquals(2, topicInfo.partitions.iterator().next().replicas.size());
+      }
+      Assertions.assertEquals(
+          "3000", admin.topics(Set.of(topicName)).get(topicName).value("segment.ms").get());
+    }
+  }
+
+  @Test
+  void testRemainingConfigs() {
+    Assertions.assertEquals(
+        0,
+        TopicHandler.remainingConfigs(
+                PostRequest.of(
+                    Map.of(
+                        TopicHandler.TOPIC_NAME_KEY,
+                        "abc",
+                        TopicHandler.NUMBER_OF_PARTITIONS_KEY,
+                        "2",
+                        TopicHandler.NUMBER_OF_REPLICAS_KEY,
+                        "2")))
+            .size());
+
+    Assertions.assertEquals(
+        1,
+        TopicHandler.remainingConfigs(
+                PostRequest.of(Map.of(TopicHandler.TOPIC_NAME_KEY, "abc", "key", "value")))
+            .size());
+  }
 }