Repository: skiptests/astraea
Pull request #474 (2022-07-11 07:12:16)

Simplify ClusterLogAllocation
這隻PR示範如何簡化`ClusterLogAllocation`

一個關鍵的地方是`LayeredClusterLogAllocation`很在意產生規劃過程中的物件建構成本，這導致`LayeredClusterLogAllocation`在設計上需要耗費很多程式碼去避免不斷產生規劃時額外製造的物件，但或許這個問題並不需要在意，原因有二：
1. 使用者不可能一次處理過多的partitions，這個搬移的成本會很恐怖
2. 假如使用者打算一次搬移很多partitions，那麼產生規劃所需的時間相比之下根本九牛一毛，更不用去在意如何少產生幾個物件

因此`LayeredClusterLogAllocation`這個物件其實可以省略，我們可以就immutable object的思維來設計`ClusterLogAllocation`

chia7712: @qoo332001 @garyparrot 這個是禮拜三會議的起頭，現在的balancer裡面還有很多地方有點過度設計可以簡化，當然這隻PR不一定正確，或許你們有看到一些我沒想到的地方，都可以在禮拜三之前先在這隻PR上討論
garyparrot: > 一個關鍵的地方是LayeredClusterLogAllocation很在意產生規劃過程中的物件建構成本，這導致LayeredClusterLogAllocation在設計上需要耗費很多程式碼去避免不斷產生規劃時額外製造的物件，但或許這個問題並不需要在意，原因有二：
> 使用者不可能一次處理過多的partitions，這個搬移的成本會很恐怖

當初的效能優化並非在優化處理大量 Partition 的搬移情境，那個情境和你說的一樣搬移成本很高，通常不太會有人這樣做，當時優化的目標是，**在大叢集環境中，增進搜尋計劃的速度**，這個計劃生成是整體流程的第一站，如果他跑太慢會連帶使後面的 Cost Function 計算也變慢，想象一個叢集有 7500 多個 Log，我們對他做 3 個搬移，可能有 `7500*7499*7498` 種可能(隨便算的)，然後當時因為資料複製的問題，導致這個計劃生成的第一站每秒只能產生 250 多個計劃，造成整體計劃搜尋效率的下降(浪費了很多 CPU 時間在複製幾乎相同的東西)。提升整體計劃搜尋速度才是主要要修正的東西

https://github.com/skiptests/astraea/pull/474/files#diff-f06d96b5278eae4f511ab84dff9de7dda606622975d4597be3330529b3dcb64bR177

這個實作內依然有複製整個 `Map`，他在 log 數量大的時候也會撞到類似的問題，`immutable object` 很好，但當我們複製的 object 變重的時候又是另外一回事。
chia7712: > 我們對他做 3 個搬移，可能有 7500*7499*7498 種可能(隨便算的)，

我確認一下，這邊說的log是指replica嗎？如果是的話，ClusterLogAllocation在使用上會保存「所有」的replica allocation嗎？我的想象中那應該是保存「目標」topic的partitions而已，換言之，在ClusterLogAllocation這一層保存的物件不會涵蓋整個叢集的範圍。如果會的話，那可能要重新審視架構面的東西，這代表有太多「類別」用來表達「同樣」的東西

> 然後當時因為資料複製的問題，導致這個計劃生成的第一站每秒只能產生 250 多個計劃，整體搜尋的效果會很差勁，造成整體計劃搜尋效率的下降(浪費了很多 CPU 時間在複製幾乎相同的東西)。提升整體計劃搜尋速度才是主要要修正的東西

這是一個好問題，一秒250個是足夠還是不足夠？感覺你是覺得不夠多，那麼可否分享一下原因是什麼？例如你做了實驗發現比較好的評估通常落在後面20000個計畫？如果是的話，或許是要考慮說為何好的計畫落在後面，而不是我們能多快找到好的計畫

> 或是說我們去哪裡找個 lazy 版本的 map 來使

承接上面的討論，我希望先確定這個「問題」是值得處理的，否則我們只是花了很多時間寫很複雜的程式碼去處理一個不值得的問題

garyparrot: > 我確認一下，這邊說的log是指replica嗎？如果是的話，ClusterLogAllocation在使用上會保存「所有」的replica allocation嗎？

看傳入的 `ClusterInfo` 而定，如果傳入的 `ClusterInfo` 只有部分訊息，那 generator 產生的結果也只有部分訊息。

> 我的想象中那應該是保存「目標」topic的partitions而已

目標是指?

> 這是一個好問題，一秒250個是足夠還是不足夠？感覺你是覺得不夠多，那麼可否分享一下原因是什麼？例如你做了實驗發現比較好的評估通常落在後面20000個計畫？如果是的話，或許是要考慮說為何好的計畫落在後面，而不是我們能多快找到好的計畫

嘗試次數本來就是多多易善，沒有足夠的問題。能不能快速找到好的計劃，除了要看 generator 吐出的計劃品質，能不能快速產生結果也有一些因素在，你的問題是前者(generator 的品質)的問題，而原先的 `layered` 版本在處理後者。
chia7712: > 目標是指?

我想像中的用法是基於topic來作負載平衡，所以這裡說的目標是指某個topic的replicas

> 嘗試次數本來就是多多易善，沒有足夠的問題。能不能快速找到好的計劃，除了要看 generator 吐出的計劃品質，能不能快速產生結果也有一些因素在，你的問題是前者(generator 的品質)的問題，而原先的 layered 版本在處理後者。

你這段話講的是正確的，但沒有回答說「為何你覺得後者這個問題值得處理」，例如你說一秒250個規劃的效果很差勁，這會有兩個議題（就如我前面說的，你這邊複述的）品質和速度，為何你選擇處理後者。另外你這邊提到的差勁是指什麼意思？是已經有一個方式來評估plan了嗎？如果是的話，可否分享一下方法？
chia7712: > 你是指 cluster log allocation 或說 generator 的輸出只包含他要做的變更嗎

是的

> 當時我還有餘力，所以就順帶處理後者，前者的部分我覺得現階段還不需要處理，隨機在小規模的情境下夠用。
> 差勁只是形容詞，指他花了不少力氣在複製資料而非評估計劃。評估計劃是 Cost Function 在做的事情

這個就很像之前 @wycccccc  一直被我抓著問，也是工程領域很需要的一個技能，叫做“這個解法、這個問題，值不值得"

簡單來說，你當時選擇了一個較為“複雜”的方式來處理“這個問題“，這是因為你發現了什麼關鍵的瓶頸還是只是”感覺“要處理所以就處理？如果是前者的話，就會想知道那個關鍵的瓶頸是什麼？（例如你前面給了一個叫做250/秒，當然這可能是你隨意給的或是憑感覺給的)，如果是後者的話，那我們可以試著去釐清你的“感覺”正不正確，如果正確，那我們接下來就討論該如何去處理這個問題，要維持原本較為複雜的方式？還是如我前面提到要去處理產生規劃的品質。反之你的感覺不正確，那故事就很簡單，我們一起來簡化balancer的架構
chia7712: @garyparrot @qoo332001 麻煩你們看一下~

Patch:
diff --git a/README.md b/README.md
index a5e6193b..11044bac 100644
--- a/README.md
+++ b/README.md
@@ -192,19 +192,20 @@ Run the benchmark from source
   The duration formats accepted are (a number) + (a time unit). 
   The time units can be "days", "day", "h", "m", "s", "ms", "us", "ns".
   e.g. "--run.until 1m" or "--run.until 89242records" Default: 1000records
-9. --record.size: the (bound of) record size in byte. Default: 1 KiB
-10. --prop.file: the path to property file.
-11. --partitioner: the partitioner to use in producers.
-12. --configs: the configurations pass to partitioner. 
+9. --prop.file: the path to property file.
+10. --partitioner: the partitioner to use in producers.
+11. --configs: the configurations pass to partitioner. 
   The configuration format is "\<key1\>=\<value1\>[,\<key2\>=\<value2\>]*". 
   eg. "--configs broker.1001.jmx.port=14338,org.astraea.cost.ThroughputCost=1"
-13. --throughput: the produce rate for all producers. e.g. "--throughput 2MiB". Default: 500 GiB (per second)
-14. --key.distribution: name of the distribution on key. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: (No key)
-15. --size.distribution: name of the distribution on value size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: "uniform"
-16. --specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
-17. --report.path: A path to place the report file. Default: (no report)
-18. --report.format: Select output file format. Available format: "csv", "json". Default: "csv"
-19. --transaction.size: number of records in each transaction. Default: 1
+12. --throughput: the produce rate for all producers. e.g. "--throughput 2MiB". Default: 500 GiB (per second)
+13. --key.size: the bound of DataSize of the key. Default: 4Byte
+14. --value.size: the bound of DataSize of the value. Default: 1KiB
+15. --key.distribution: distribution name for key and key size. Available distribution names: "fixed" "uniform", "zipfian", "latest". Default: fixed
+16. --value.distribution: distribution name for value and record size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: uniform
+17. --specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
+18. --report.path: A path to place the report file. Default: (no report)
+19. --report.format: Select output file format. Available format: "csv", "json". Default: "csv"
+20. --transaction.size: number of records in each transaction. Default: 1
 
 ---
 
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java b/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
index e11ff360..ebb04c4c 100644
--- a/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
+++ b/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
@@ -23,7 +23,6 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
-import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
 
 /** Execute every possible migration immediately. */
 public class StraightPlanExecutor implements RebalancePlanExecutor {
@@ -33,7 +32,7 @@ public class StraightPlanExecutor implements RebalancePlanExecutor {
   @Override
   public void run(RebalanceAdmin rebalanceAdmin, ClusterLogAllocation logAllocation) {
     final var clusterInfo = rebalanceAdmin.clusterInfo();
-    final var currentLogAllocation = LayeredClusterLogAllocation.of(clusterInfo);
+    final var currentLogAllocation = ClusterLogAllocation.of(clusterInfo);
     final var migrationTargets =
         ClusterLogAllocation.findNonFulfilledAllocation(currentLogAllocation, logAllocation);
 
diff --git a/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java b/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java
index abb7c0a5..26c46ca2 100644
--- a/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java
+++ b/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java
@@ -20,7 +20,6 @@ import java.util.stream.Stream;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.balancer.RebalancePlanProposal;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
-import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
 
 /** */
 public interface RebalancePlanGenerator {
@@ -35,7 +34,7 @@ public interface RebalancePlanGenerator {
    * @return a {@link Stream} generating rebalance plan regarding the given {@link ClusterInfo}
    */
   default Stream<RebalancePlanProposal> generate(ClusterInfo clusterInfo) {
-    return generate(clusterInfo, LayeredClusterLogAllocation.of(clusterInfo));
+    return generate(clusterInfo, ClusterLogAllocation.of(clusterInfo));
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java b/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
index 2ce0b269..0de7f2d7 100644
--- a/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
+++ b/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
@@ -20,16 +20,16 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.RebalancePlanProposal;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
-import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
 import org.astraea.app.balancer.log.LogPlacement;
 
 /**
@@ -60,21 +60,6 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
     this.numberOfShuffle = numberOfShuffle;
   }
 
-  private int sourceTopicPartitionSelector(List<TopicPartition> migrationCandidates) {
-    return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
-  }
-
-  private int sourceLogPlacementSelector(List<LogPlacement> migrationCandidates) {
-    return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
-  }
-
-  private <T> T randomElement(Collection<T> collection) {
-    return collection.stream()
-        .skip(ThreadLocalRandom.current().nextInt(0, collection.size()))
-        .findFirst()
-        .orElseThrow();
-  }
-
   @Override
   public Stream<RebalancePlanProposal> generate(
       ClusterInfo clusterInfo, ClusterLogAllocation baseAllocation) {
@@ -105,92 +90,104 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
                 .build();
 
           final var shuffleCount = numberOfShuffle.get();
-          final var newAllocation = LayeredClusterLogAllocation.of(baseAllocation);
-          final var pickingList =
-              newAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableList());
 
           rebalancePlanBuilder.addInfo(
               "Make " + shuffleCount + (shuffleCount > 0 ? " shuffles." : " shuffle."));
-          for (int i = 0; i < shuffleCount; i++) {
-            final var sourceTopicPartitionIndex = sourceTopicPartitionSelector(pickingList);
-            final var sourceTopicPartition = pickingList.get(sourceTopicPartitionIndex);
-            final var sourceLogPlacements = newAllocation.logPlacements(sourceTopicPartition);
-            final var sourceLogPlacementIndex = sourceLogPlacementSelector(sourceLogPlacements);
-            final var sourceLogPlacement = sourceLogPlacements.get(sourceLogPlacementIndex);
-            final var sourceIsLeader = sourceLogPlacementIndex == 0;
-            final var sourceBroker = sourceLogPlacement.broker();
-
-            Consumer<Integer> replicaSetMigration =
-                (targetBroker) -> {
-                  var destDir = randomElement(clusterInfo.dataDirectories(targetBroker));
-                  newAllocation.migrateReplica(
-                      sourceTopicPartition, sourceBroker, targetBroker, destDir);
-                  rebalancePlanBuilder.addInfo(
-                      String.format(
-                          "Change replica set of topic %s partition %d, from %d to %d at %s.",
-                          sourceTopicPartition.topic(),
-                          sourceTopicPartition.partition(),
-                          sourceLogPlacement.broker(),
-                          targetBroker,
-                          destDir));
-                };
-            Consumer<LogPlacement> leaderFollowerMigration =
-                (newLeaderReplicaCandidate) -> {
-                  newAllocation.letReplicaBecomeLeader(
-                      sourceTopicPartition, newLeaderReplicaCandidate.broker());
-                  rebalancePlanBuilder.addInfo(
-                      String.format(
-                          "Change the log identity of topic %s partition %d replica at broker %d, from %s to %s",
-                          sourceTopicPartition.topic(),
-                          sourceTopicPartition.partition(),
-                          sourceLogPlacement.broker(),
-                          sourceIsLeader ? "leader" : "follower",
-                          sourceIsLeader ? "follower" : "leader"));
-                };
-
-            // generate a set of valid migration broker for given placement.
-            final var validMigrationCandidates = new ArrayList<Movement>();
-            // [Valid movement 1] add all brokers and remove all broker in current replica set
-            brokerIds.stream()
-                .filter(
-                    broker -> sourceLogPlacements.stream().noneMatch(log -> log.broker() == broker))
-                .map(targetBroker -> (Runnable) () -> replicaSetMigration.accept(targetBroker))
-                .map(Movement::replicaSetMovement)
-                .forEach(validMigrationCandidates::add);
-            // [Valid movement 2] add all leader/follower change candidate
-            if (sourceLogPlacements.size() > 1) {
-              if (sourceIsLeader) {
-                // leader can migrate its identity to follower.
-                sourceLogPlacements.stream()
-                    .skip(1)
-                    .map(
-                        followerReplica ->
-                            (Runnable) () -> leaderFollowerMigration.accept(followerReplica))
-                    .map(Movement::replicaSetMovement)
-                    .forEach(validMigrationCandidates::add);
-              } else {
-                // follower can migrate its identity to leader.
-                validMigrationCandidates.add(
-                    Movement.replicaSetMovement(
-                        () -> leaderFollowerMigration.accept(sourceLogPlacement)));
-              }
-            }
-
-            // pick a migration and execute
-            final var selectedMigrationIndex = randomElement(validMigrationCandidates);
-            selectedMigrationIndex.run();
-          }
-
-          return rebalancePlanBuilder.withRebalancePlan(newAllocation).build();
+
+          var candidates =
+              IntStream.range(0, shuffleCount)
+                  .mapToObj(i -> allocationGenerator(clusterInfo, rebalancePlanBuilder))
+                  .collect(Collectors.toUnmodifiableList());
+
+          var currentAllocation = baseAllocation;
+          for (var candidate : candidates) currentAllocation = candidate.apply(currentAllocation);
+
+          return rebalancePlanBuilder.withRebalancePlan(currentAllocation).build();
         });
   }
 
-  /** Action to trigger upon this migration */
-  interface Movement extends Runnable {
-    static ReplicaSetMovement replicaSetMovement(Runnable runnable) {
-      return runnable::run;
-    }
+  private static Function<ClusterLogAllocation, ClusterLogAllocation> allocationGenerator(
+      ClusterInfo clusterInfo, RebalancePlanProposal.Build rebalancePlanBuilder) {
+    return currentAllocation -> {
+      var brokerIds =
+          clusterInfo.nodes().stream().map(NodeInfo::id).collect(Collectors.toUnmodifiableSet());
+      var pickingList = new ArrayList<>(currentAllocation.topicPartitions());
+      final var sourceTopicPartitionIndex = sourceTopicPartitionSelector(pickingList);
+      final var sourceTopicPartition = pickingList.get(sourceTopicPartitionIndex);
+      final var sourceLogPlacements = currentAllocation.logPlacements(sourceTopicPartition);
+      final var sourceLogPlacementIndex = sourceLogPlacementSelector(sourceLogPlacements);
+      final var sourceLogPlacement = sourceLogPlacements.get(sourceLogPlacementIndex);
+      final var sourceIsLeader = sourceLogPlacementIndex == 0;
+      final var sourceBroker = sourceLogPlacement.broker();
+      var targetPlacements =
+          sourceLogPlacements.size() > 1
+              ? sourceIsLeader
+                  ? sourceLogPlacements.stream().skip(1).collect(Collectors.toUnmodifiableList())
+                  : List.of(sourceLogPlacement)
+              : List.<LogPlacement>of();
+      // generate a set of valid migration broker for given placement.
+      final var validMigrationCandidates =
+          Stream.concat(
+
+                  // [Valid movement 1] add all brokers and remove all
+                  // broker in current replica set
+                  brokerIds.stream()
+                      .filter(
+                          broker ->
+                              sourceLogPlacements.stream().noneMatch(log -> log.broker() == broker))
+                      .map(
+                          targetBroker ->
+                              (Supplier<ClusterLogAllocation>)
+                                  () -> {
+                                    var destDir =
+                                        randomElement(clusterInfo.dataDirectories(targetBroker));
+                                    rebalancePlanBuilder.addInfo(
+                                        String.format(
+                                            "Change replica set of topic %s partition %d, from %d to %d at %s.",
+                                            sourceTopicPartition.topic(),
+                                            sourceTopicPartition.partition(),
+                                            sourceLogPlacement.broker(),
+                                            targetBroker,
+                                            destDir));
+                                    return currentAllocation.migrateReplica(
+                                        sourceTopicPartition, sourceBroker, targetBroker, destDir);
+                                  }),
+                  // [Valid movement 2] add all leader/follower change
+                  // candidate
+                  targetPlacements.stream()
+                      .map(
+                          followerReplica ->
+                              () -> {
+                                rebalancePlanBuilder.addInfo(
+                                    String.format(
+                                        "Change the log identity of topic %s partition %d replica at broker %d, from %s to %s",
+                                        sourceTopicPartition.topic(),
+                                        sourceTopicPartition.partition(),
+                                        sourceLogPlacement.broker(),
+                                        sourceIsLeader ? "leader" : "follower",
+                                        sourceIsLeader ? "follower" : "leader"));
+                                return currentAllocation.letReplicaBecomeLeader(
+                                    sourceTopicPartition, followerReplica.broker());
+                              }))
+              .collect(Collectors.toUnmodifiableList());
+      // pick a migration and execute
+      final var selectedMigrationIndex = randomElement(validMigrationCandidates);
+      return selectedMigrationIndex.get();
+    };
   }
 
-  interface ReplicaSetMovement extends Movement {}
+  private static int sourceTopicPartitionSelector(List<TopicPartition> migrationCandidates) {
+    return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
+  }
+
+  private static int sourceLogPlacementSelector(List<LogPlacement> migrationCandidates) {
+    return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
+  }
+
+  private static <T> T randomElement(Collection<T> collection) {
+    return collection.stream()
+        .skip(ThreadLocalRandom.current().nextInt(0, collection.size()))
+        .findFirst()
+        .orElseThrow();
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index 850bc703..482418df 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -16,10 +16,18 @@
  */
 package org.astraea.app.balancer.log;
 
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
 import java.util.Set;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
+import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 
 /**
@@ -29,6 +37,42 @@ import org.astraea.app.admin.TopicPartition;
  */
 public interface ClusterLogAllocation {
 
+  static ClusterLogAllocation of(ClusterInfo clusterInfo) {
+    return of(
+        clusterInfo.topics().stream()
+            .map(clusterInfo::replicas)
+            .flatMap(Collection::stream)
+            .collect(
+                Collectors.groupingBy(
+                    replica ->
+                        TopicPartition.of(replica.topic(), Integer.toString(replica.partition()))))
+            .entrySet()
+            .stream()
+            .map(
+                (entry) -> {
+                  // validate if the given log placements are valid
+                  if (entry.getValue().stream().filter(ReplicaInfo::isLeader).count() != 1)
+                    throw new IllegalArgumentException(
+                        "The " + entry.getKey() + " leader count mismatch 1.");
+
+                  final var topicPartition = entry.getKey();
+                  final var logPlacements =
+                      entry.getValue().stream()
+                          .sorted(Comparator.comparingInt(replica -> replica.isLeader() ? 0 : 1))
+                          .map(
+                              replica ->
+                                  LogPlacement.of(
+                                      replica.nodeInfo().id(), replica.dataFolder().orElse(null)))
+                          .collect(Collectors.toList());
+
+                  return Map.entry(topicPartition, logPlacements);
+                })
+            .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue)));
+  }
+
+  static ClusterLogAllocation of(Map<TopicPartition, List<LogPlacement>> allocation) {
+    return new ClusterLogAllocationImpl(allocation);
+  }
   /**
    * let specific broker leave the replica set and let another broker join the replica set. Which
    * data directory the migrated replica will be is up to the Kafka broker implementation to decide.
@@ -37,8 +81,9 @@ public interface ClusterLogAllocation {
    * @param atBroker the id of the broker about to remove
    * @param toBroker the id of the broker about to replace the removed broker
    */
-  default void migrateReplica(TopicPartition topicPartition, int atBroker, int toBroker) {
-    migrateReplica(topicPartition, atBroker, toBroker, null);
+  default ClusterLogAllocation migrateReplica(
+      TopicPartition topicPartition, int atBroker, int toBroker) {
+    return migrateReplica(topicPartition, atBroker, toBroker, null);
   }
   // TODO: Revise the log argument by TopicPartitionReplica, once #411 is merged
 
@@ -52,17 +97,23 @@ public interface ClusterLogAllocation {
    *     the destination broker, if {@code null} is specified then the data directory choice is left
    *     up to the Kafka broker implementation.
    */
-  void migrateReplica(TopicPartition topicPartition, int atBroker, int toBroker, String toDir);
+  ClusterLogAllocation migrateReplica(
+      TopicPartition topicPartition, int atBroker, int toBroker, String toDir);
   // TODO: Revise the log argument by TopicPartitionReplica, once #411 is merged
 
   /** let specific follower log become the leader log of this topic/partition. */
-  void letReplicaBecomeLeader(TopicPartition topicPartition, int followerReplica);
+  ClusterLogAllocation letReplicaBecomeLeader(TopicPartition topicPartition, int followerReplica);
 
-  /** Retrieve the log placements of specific {@link TopicPartition}. */
+  /**
+   * Retrieve the log placements of specific {@link TopicPartition}.
+   *
+   * @param topicPartition to query
+   * @return log placements or empty collection if there is no log placements
+   */
   List<LogPlacement> logPlacements(TopicPartition topicPartition);
 
   /** Retrieve the stream of all topic/partition pairs in allocation. */
-  Stream<TopicPartition> topicPartitionStream();
+  Set<TopicPartition> topicPartitions();
 
   /**
    * Find a subset of topic/partitions in the source allocation, that has any non-fulfilled log
@@ -73,10 +124,8 @@ public interface ClusterLogAllocation {
   static Set<TopicPartition> findNonFulfilledAllocation(
       ClusterLogAllocation source, ClusterLogAllocation target) {
 
-    final var sourceTopicPartition =
-        source.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
-    final var targetTopicPartition =
-        target.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+    final var sourceTopicPartition = source.topicPartitions();
+    final var targetTopicPartition = target.topicPartitions();
 
     if (!sourceTopicPartition.equals(targetTopicPartition))
       throw new IllegalArgumentException(
@@ -90,8 +139,7 @@ public interface ClusterLogAllocation {
   static String toString(ClusterLogAllocation allocation) {
     StringBuilder stringBuilder = new StringBuilder();
 
-    allocation
-        .topicPartitionStream()
+    allocation.topicPartitions().stream()
         .sorted()
         .forEach(
             tp -> {
@@ -109,4 +157,112 @@ public interface ClusterLogAllocation {
 
     return stringBuilder.toString();
   }
+
+  class ClusterLogAllocationImpl implements ClusterLogAllocation {
+
+    private final Map<TopicPartition, List<LogPlacement>> allocation;
+
+    private ClusterLogAllocationImpl(Map<TopicPartition, List<LogPlacement>> allocation) {
+      this.allocation = Collections.unmodifiableMap(allocation);
+
+      this.allocation.keySet().stream()
+          .collect(Collectors.groupingBy(TopicPartition::topic))
+          .forEach(
+              (topic, tp) -> {
+                int maxPartitionId =
+                    tp.stream().mapToInt(TopicPartition::partition).max().orElseThrow();
+                if ((maxPartitionId + 1) != tp.size())
+                  throw new IllegalArgumentException(
+                      "The partition size of " + topic + " is illegal");
+              });
+
+      this.allocation.forEach(
+          (tp, logs) -> {
+            long uniqueBrokers = logs.stream().map(LogPlacement::broker).distinct().count();
+            if (uniqueBrokers != logs.size() || logs.size() == 0)
+              throw new IllegalArgumentException(
+                  "The topic "
+                      + tp.topic()
+                      + " partition "
+                      + tp.partition()
+                      + " has illegal replica set "
+                      + logs);
+          });
+    }
+
+    @Override
+    public ClusterLogAllocation migrateReplica(
+        TopicPartition topicPartition, int atBroker, int toBroker, String toDir) {
+      var sourceLogPlacements = this.logPlacements(topicPartition);
+      if (sourceLogPlacements.isEmpty())
+        throw new IllegalMigrationException(
+            topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
+
+      int sourceLogIndex = indexOfBroker(sourceLogPlacements, atBroker).orElse(-1);
+      if (sourceLogIndex == -1)
+        throw new IllegalMigrationException(
+            atBroker + " is not part of the replica set for " + topicPartition);
+
+      var newAllocations = new HashMap<>(allocation);
+      newAllocations.put(
+          topicPartition,
+          IntStream.range(0, sourceLogPlacements.size())
+              .mapToObj(
+                  index ->
+                      index == sourceLogIndex
+                          ? LogPlacement.of(toBroker, toDir)
+                          : sourceLogPlacements.get(index))
+              .collect(Collectors.toUnmodifiableList()));
+      return new ClusterLogAllocationImpl(newAllocations);
+    }
+
+    @Override
+    public ClusterLogAllocation letReplicaBecomeLeader(
+        TopicPartition topicPartition, int followerReplica) {
+      final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
+      if (sourceLogPlacements.isEmpty())
+        throw new IllegalMigrationException(
+            topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
+
+      int leaderLogIndex = 0;
+      int followerLogIndex = indexOfBroker(sourceLogPlacements, followerReplica).orElse(-1);
+      if (followerLogIndex == -1)
+        throw new IllegalArgumentException(
+            followerReplica + " is not part of the replica set for " + topicPartition);
+
+      if (leaderLogIndex == followerLogIndex) return this; // nothing to do
+
+      final var leaderLog = this.logPlacements(topicPartition).get(leaderLogIndex);
+      final var followerLog = this.logPlacements(topicPartition).get(followerLogIndex);
+
+      var newAllocations = new HashMap<>(allocation);
+      newAllocations.put(
+          topicPartition,
+          IntStream.range(0, sourceLogPlacements.size())
+              .mapToObj(
+                  index ->
+                      index == leaderLogIndex
+                          ? followerLog
+                          : index == followerLogIndex ? leaderLog : sourceLogPlacements.get(index))
+              .collect(Collectors.toUnmodifiableList()));
+
+      return new ClusterLogAllocationImpl(newAllocations);
+    }
+
+    @Override
+    public List<LogPlacement> logPlacements(TopicPartition topicPartition) {
+      return allocation.getOrDefault(topicPartition, List.of());
+    }
+
+    @Override
+    public Set<TopicPartition> topicPartitions() {
+      return allocation.keySet();
+    }
+
+    private static OptionalInt indexOfBroker(List<LogPlacement> logPlacements, int targetBroker) {
+      return IntStream.range(0, logPlacements.size())
+          .filter(index -> logPlacements.get(index).broker() == targetBroker)
+          .findFirst();
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java
deleted file mode 100644
index 3ae58fca..00000000
--- a/app/src/main/java/org/astraea/app/balancer/log/LayeredClusterLogAllocation.java
+++ /dev/null
@@ -1,250 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.astraea.app.balancer.log;
-
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.OptionalInt;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import java.util.stream.Stream;
-import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.admin.ReplicaInfo;
-import org.astraea.app.admin.TopicPartition;
-
-public class LayeredClusterLogAllocation implements ClusterLogAllocation {
-
-  // set operation is guard by this
-  private final AtomicBoolean isLocked = new AtomicBoolean(false);
-  private final LayeredClusterLogAllocation upperLayer;
-
-  // guard by this
-  private final Map<TopicPartition, List<LogPlacement>> allocation;
-
-  private LayeredClusterLogAllocation(
-      LayeredClusterLogAllocation upperLayer, Map<TopicPartition, List<LogPlacement>> allocation) {
-    allocation.keySet().stream()
-        .collect(Collectors.groupingBy(TopicPartition::topic))
-        .forEach(
-            (topic, tp) -> {
-              int maxPartitionId =
-                  tp.stream().mapToInt(TopicPartition::partition).max().orElseThrow();
-              if ((maxPartitionId + 1) != tp.size())
-                throw new IllegalArgumentException(
-                    "The partition size of " + topic + " is illegal");
-            });
-    allocation.forEach(
-        (tp, logs) -> {
-          long uniqueBrokers = logs.stream().map(LogPlacement::broker).distinct().count();
-          if (uniqueBrokers != logs.size() || logs.size() == 0)
-            throw new IllegalArgumentException(
-                "The topic "
-                    + tp.topic()
-                    + " partition "
-                    + tp.partition()
-                    + " has illegal replica set "
-                    + logs);
-        });
-    this.upperLayer = upperLayer;
-    this.allocation = allocation;
-  }
-
-  public static LayeredClusterLogAllocation of(ClusterInfo clusterInfo) {
-    final Map<TopicPartition, List<LogPlacement>> allocation =
-        clusterInfo.topics().stream()
-            .map(clusterInfo::replicas)
-            .flatMap(Collection::stream)
-            .collect(
-                Collectors.groupingBy(
-                    replica ->
-                        TopicPartition.of(replica.topic(), Integer.toString(replica.partition()))))
-            .entrySet()
-            .stream()
-            .map(
-                (entry) -> {
-                  // validate if the given log placements are valid
-                  if (entry.getValue().stream().filter(ReplicaInfo::isLeader).count() != 1)
-                    throw new IllegalArgumentException(
-                        "The " + entry.getKey() + " leader count mismatch 1.");
-
-                  final var topicPartition = entry.getKey();
-                  final var logPlacements =
-                      entry.getValue().stream()
-                          .sorted(Comparator.comparingInt(replica -> replica.isLeader() ? 0 : 1))
-                          .map(
-                              replica ->
-                                  LogPlacement.of(
-                                      replica.nodeInfo().id(), replica.dataFolder().orElse(null)))
-                          .collect(Collectors.toList());
-
-                  return Map.entry(topicPartition, logPlacements);
-                })
-            .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue));
-    return new LayeredClusterLogAllocation(null, allocation);
-  }
-
-  public static LayeredClusterLogAllocation of(
-      Map<TopicPartition, List<LogPlacement>> allocationMap) {
-    return new LayeredClusterLogAllocation(null, allocationMap);
-  }
-
-  /**
-   * Create a new {@link LayeredClusterLogAllocation}. If the given {@link ClusterLogAllocation} is
-   * an instance of {@link LayeredClusterLogAllocation}. Then the given {@link
-   * LayeredClusterLogAllocation} will be locked and become unmodifiable. The new log allocation
-   * will continue from the unmodifiable allocation. Doing so can improve the performance of new log
-   * allocation creation as we avoid the heavy data copy work.
-   */
-  public static LayeredClusterLogAllocation of(ClusterLogAllocation clusterLogAllocation) {
-    if (clusterLogAllocation instanceof LayeredClusterLogAllocation) {
-      return createNewLayer((LayeredClusterLogAllocation) clusterLogAllocation);
-    } else {
-      // no layer support, do the expensive copy work.
-      final var map = new ConcurrentHashMap<TopicPartition, List<LogPlacement>>();
-      clusterLogAllocation
-          .topicPartitionStream()
-          .forEach(
-              topicPartition ->
-                  map.put(topicPartition, clusterLogAllocation.logPlacements(topicPartition)));
-      return of(map);
-    }
-  }
-
-  /**
-   * Create a new {@link LayeredClusterLogAllocation} based on the given one. The given layered
-   * cluster log will become unmodifiable afterward.
-   */
-  private static LayeredClusterLogAllocation createNewLayer(
-      LayeredClusterLogAllocation baseLogAllocation) {
-    if (!baseLogAllocation.isLocked()) {
-      baseLogAllocation.lockLayer();
-    }
-    return new LayeredClusterLogAllocation(baseLogAllocation, new ConcurrentHashMap<>());
-  }
-
-  /** Make this {@link LayeredClusterLogAllocation} unmodifiable. */
-  private synchronized void lockLayer() {
-    isLocked.set(true);
-  }
-
-  /** Is this {@link LayeredClusterLogAllocation} unmodifiable. */
-  private boolean isLocked() {
-    return isLocked.get();
-  }
-
-  private synchronized void ensureNotLocked() {
-    if (isLocked())
-      throw new IllegalStateException("This layered cluster log allocation has been locked");
-  }
-
-  private static OptionalInt indexOfBroker(List<LogPlacement> logPlacements, int targetBroker) {
-    return IntStream.range(0, logPlacements.size())
-        .filter(index -> logPlacements.get(index).broker() == targetBroker)
-        .findFirst();
-  }
-
-  @Override
-  public synchronized void migrateReplica(
-      TopicPartition topicPartition, int broker, int destinationBroker, String toDir) {
-    ensureNotLocked();
-
-    final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
-    if (sourceLogPlacements == null)
-      throw new IllegalMigrationException(
-          topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
-
-    int sourceLogIndex = indexOfBroker(sourceLogPlacements, broker).orElse(-1);
-    if (sourceLogIndex == -1)
-      throw new IllegalMigrationException(
-          broker + " is not part of the replica set for " + topicPartition);
-
-    this.allocation.put(
-        topicPartition,
-        IntStream.range(0, sourceLogPlacements.size())
-            .mapToObj(
-                index ->
-                    index == sourceLogIndex
-                        ? LogPlacement.of(destinationBroker, toDir)
-                        : sourceLogPlacements.get(index))
-            .collect(Collectors.toUnmodifiableList()));
-  }
-
-  @Override
-  public synchronized void letReplicaBecomeLeader(
-      TopicPartition topicPartition, int followerReplica) {
-    ensureNotLocked();
-
-    final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
-    if (sourceLogPlacements == null)
-      throw new IllegalMigrationException(
-          topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
-
-    int leaderLogIndex = 0;
-    if (sourceLogPlacements.size() == 0)
-      throw new IllegalStateException("This partition has no log");
-
-    int followerLogIndex = indexOfBroker(sourceLogPlacements, followerReplica).orElse(-1);
-    if (followerLogIndex == -1)
-      throw new IllegalArgumentException(
-          followerReplica + " is not part of the replica set for " + topicPartition);
-
-    if (leaderLogIndex == followerLogIndex) return; // nothing to do
-
-    final var leaderLog = this.logPlacements(topicPartition).get(leaderLogIndex);
-    final var followerLog = this.logPlacements(topicPartition).get(followerLogIndex);
-
-    this.allocation.put(
-        topicPartition,
-        IntStream.range(0, sourceLogPlacements.size())
-            .mapToObj(
-                index ->
-                    index == leaderLogIndex
-                        ? followerLog
-                        : index == followerLogIndex ? leaderLog : sourceLogPlacements.get(index))
-            .collect(Collectors.toUnmodifiableList()));
-  }
-
-  @Override
-  public List<LogPlacement> logPlacements(TopicPartition topicPartition) {
-    if (allocation.containsKey(topicPartition)) return allocation.get(topicPartition);
-    else if (upperLayer != null) return upperLayer.logPlacements(topicPartition);
-    else return null;
-  }
-
-  @Override
-  public Stream<TopicPartition> topicPartitionStream() {
-    if (upperLayer == null) return this.allocation.keySet().stream();
-    else {
-      LayeredClusterLogAllocation baseAllocationLookUp = this;
-      while (baseAllocationLookUp.upperLayer != null)
-        baseAllocationLookUp = baseAllocationLookUp.upperLayer;
-      // Ignore all the intermediate layers, just return the keys set of the base layer.
-      // THIS WILL WORK, AS LONG AS THE UPPER LAYERS DO NOT DO THE FOLLOWING.
-      // 1. delete topic (should balancer do this?)
-      // 2. create topic (should balancer do this?)
-      // 3. shrink partition size (impossible for the current Kafka)
-      // 4. expand partition size (should balancer do this?)
-      // Of course, we can consider every element in all the layers, but doing that hurt
-      // performance.
-      return baseAllocationLookUp.allocation.keySet().stream();
-    }
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/performance/DataSupplier.java b/app/src/main/java/org/astraea/app/performance/DataSupplier.java
index 6a2ffaf1..b20caf48 100644
--- a/app/src/main/java/org/astraea/app/performance/DataSupplier.java
+++ b/app/src/main/java/org/astraea/app/performance/DataSupplier.java
@@ -16,8 +16,9 @@
  */
 package org.astraea.app.performance;
 
-import java.util.Arrays;
+import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 import org.astraea.app.common.DataSize;
@@ -116,19 +117,48 @@ interface DataSupplier extends Supplier<DataSupplier.Data> {
     byte[] value();
   }
 
+  /**
+   * Generate Data according to the given arguments. The returned supplier map the 64-bit number
+   * supplied by key(/value) distribution to a byte array. That is, if we want the DataSupplier to
+   * produce the same content, the key(/value) distribution should always produce the same number.
+   * For example,
+   *
+   * <pre>{@code
+   * DataSupplier.of(exeTime,
+   *                 ()->1,
+   *                 keySizeDistribution,
+   *                 ()->1,
+   *                 valueSizeDistribution,
+   *                 throughput)
+   * }</pre>
+   *
+   * It is not recommend to supply too many unique number. This DataSupplier store every unique
+   * number and its content in a map structure.
+   *
+   * @param exeTime the time for stop supplying data
+   * @param keyDistribution supply abstract keys which is represented by a 64-bit integer
+   * @param keySizeDistribution supply the size of newly created key
+   * @param valueDistribution supply abstract value which is represented by a 64-bit integer
+   * @param valueSizeDistribution supply the size of newly created value
+   * @param throughput the limit on data produced per second
+   * @return supply data with given distribution. It will map the 64-bit number supplied by
+   *     key(/value) distribution to a byte array.
+   */
   static DataSupplier of(
       ExeTime exeTime,
       Supplier<Long> keyDistribution,
-      DataSize valueSize,
+      Supplier<Long> keySizeDistribution,
       Supplier<Long> valueDistribution,
+      Supplier<Long> valueSizeDistribution,
       DataSize throughput) {
     return new DataSupplier() {
       private final long start = System.currentTimeMillis();
       private final Random rand = new Random();
-      private final byte[] content = new byte[valueSize.measurement(DataUnit.Byte).intValue()];
       private final AtomicLong dataCount = new AtomicLong(0);
       private long intervalStart = 0;
       private long payloadBytes;
+      private final Map<Long, byte[]> recordKeyTable = new ConcurrentHashMap<>();
+      private final Map<Long, byte[]> recordValueTable = new ConcurrentHashMap<>();
 
       synchronized boolean checkAndAdd(int payloadLength) {
         if (System.currentTimeMillis() - intervalStart > 1000) {
@@ -144,14 +174,13 @@ interface DataSupplier extends Supplier<DataSupplier.Data> {
       }
 
       byte[] value() {
-        // Randomly change one position of the content;
-        content[rand.nextInt(content.length)] = (byte) rand.nextInt(256);
-        return Arrays.copyOfRange(
-            content, (int) (valueDistribution.get() % content.length), content.length);
+        return getOrNew(
+            recordValueTable, valueDistribution, valueSizeDistribution.get().intValue(), rand);
       }
 
       public byte[] key() {
-        return (String.valueOf(keyDistribution.get())).getBytes();
+        return getOrNew(
+            recordKeyTable, keyDistribution, keySizeDistribution.get().intValue(), rand);
       }
 
       @Override
@@ -160,9 +189,25 @@ interface DataSupplier extends Supplier<DataSupplier.Data> {
             >= 100D) return NO_MORE_DATA;
         var key = key();
         var value = value();
-        if (checkAndAdd(value.length)) return data(key, value);
+        if (checkAndAdd((value != null ? value.length : 0) + (key != null ? key.length : 0)))
+          return data(key, value);
         return THROTTLED_DATA;
       }
     };
   }
+
+  // Find the key from the table, if the record has been produced before. Randomly generate a
+  // byte array if
+  // the record has not been produced.
+  private static byte[] getOrNew(
+      Map<Long, byte[]> table, Supplier<Long> distribution, int size, Random rand) {
+    return table.computeIfAbsent(
+        distribution.get(),
+        ignore -> {
+          if (size == 0) return null;
+          var value = new byte[size];
+          rand.nextBytes(value);
+          return value;
+        });
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/performance/Manager.java b/app/src/main/java/org/astraea/app/performance/Manager.java
index d73e3127..249df9a3 100644
--- a/app/src/main/java/org/astraea/app/performance/Manager.java
+++ b/app/src/main/java/org/astraea/app/performance/Manager.java
@@ -17,9 +17,9 @@
 package org.astraea.app.performance;
 
 import java.util.List;
-import java.util.function.Supplier;
-import org.astraea.app.common.DataUnit;
 
+// TODO: Remove this class by reconstructing consumerExecutor. See issue#476
+// https://github.com/skiptests/astraea/issues/476
 /**
  * Thread safe This class is used for managing the start/end of the producer/consumer threads.
  * Producer can start producing until all consumers get assignment. Consumers can stop after all
@@ -28,7 +28,6 @@ import org.astraea.app.common.DataUnit;
 public class Manager {
   private final ExeTime exeTime;
   private final List<Metrics> producerMetrics, consumerMetrics;
-  private final Supplier<Long> keyDistribution;
 
   /**
    * Used to manage producing/consuming.
@@ -47,13 +46,9 @@ public class Manager {
    */
   public Manager(
       Performance.Argument argument, List<Metrics> producerMetrics, List<Metrics> consumerMetrics) {
-    if (argument.recordSize.greaterThan(DataUnit.Byte.of(Integer.MAX_VALUE)))
-      throw new IllegalArgumentException(
-          "Record size should be smaller than or equal to 2147483648 (Integer.MAX_VALUE) bytes");
     this.producerMetrics = producerMetrics;
     this.consumerMetrics = consumerMetrics;
     this.exeTime = argument.exeTime;
-    this.keyDistribution = argument.keyDistributionType.create(100000);
   }
 
   public long producedRecords() {
@@ -72,9 +67,4 @@ public class Manager {
   public boolean consumedDone() {
     return consumerMetrics.size() == 0 || consumedRecords() >= producedRecords();
   }
-
-  /** Randomly choose a key according to the distribution. */
-  public byte[] getKey() {
-    return (String.valueOf(keyDistribution.get())).getBytes();
-  }
 }
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index 7bdad597..6f23073e 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -31,7 +31,6 @@ import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.errors.WakeupException;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.Compression;
@@ -89,9 +88,10 @@ public class Performance {
     return DataSupplier.of(
         argument.exeTime,
         argument.keyDistributionType.create(10000),
-        argument.recordSize,
-        argument.sizeDistributionType.create(
-            argument.recordSize.measurement(DataUnit.Byte).intValue()),
+        argument.keyDistributionType.create(argument.keySize.measurement(DataUnit.Byte).intValue()),
+        argument.valueDistributionType.create(10000),
+        argument.valueDistributionType.create(
+            argument.valueSize.measurement(DataUnit.Byte).intValue()),
         argument.throughput);
   }
 
@@ -310,17 +310,11 @@ public class Performance {
         converter = ExeTime.Field.class)
     ExeTime exeTime = ExeTime.of("1000records");
 
-    @Parameter(
-        names = {"--record.size"},
-        description = "DataSize: size of each record. e.g. \"500KiB\"",
-        converter = DataSize.Field.class)
-    DataSize recordSize = DataUnit.KiB.of(1);
-
     @Parameter(
         names = {"--partitioner"},
         description = "String: the full class name of the desired partitioner",
         validateWith = NonEmptyStringField.class)
-    String partitioner = DefaultPartitioner.class.getName();
+    String partitioner = null;
 
     @Parameter(
         names = {"--compression"},
@@ -340,19 +334,31 @@ public class Performance {
       return transactionSize > 1 ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED;
     }
 
+    @Parameter(
+        names = {"--key.size"},
+        description = "DataSize of the key. Default: 4Byte",
+        converter = DataSize.Field.class)
+    DataSize keySize = DataUnit.Byte.of(4);
+
+    @Parameter(
+        names = {"--value.size"},
+        description = "DataSize of the value. Default: 1KiB",
+        converter = DataSize.Field.class)
+    DataSize valueSize = DataUnit.KiB.of(1);
+
     @Parameter(
         names = {"--key.distribution"},
         description =
-            "String: Distribution name. Available distribution names: \"fixed\" \"uniform\", \"zipfian\", \"latest\". Default: uniform",
+            "Distribution name for key and key size. Available distribution names: \"fixed\" \"uniform\", \"zipfian\", \"latest\". Default: fixed",
         converter = DistributionType.DistributionTypeField.class)
-    DistributionType keyDistributionType = DistributionType.UNIFORM;
+    DistributionType keyDistributionType = DistributionType.FIXED;
 
     @Parameter(
-        names = {"--size.distribution"},
+        names = {"--value.distribution"},
         description =
-            "String: Distribution name. Available distribution names: \"uniform\", \"zipfian\", \"latest\", \"fixed\". Default: \"uniform\"",
+            "Distribution name for value and value size. Available distribution names: \"uniform\", \"zipfian\", \"latest\", \"fixed\". Default: uniform",
         converter = DistributionType.DistributionTypeField.class)
-    DistributionType sizeDistributionType = DistributionType.UNIFORM;
+    DistributionType valueDistributionType = DistributionType.UNIFORM;
 
     @Parameter(
         names = {"--specify.broker"},
diff --git a/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java b/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java
index 371d4852..0bb32281 100644
--- a/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java
+++ b/app/src/main/java/org/astraea/app/performance/ProducerExecutor.java
@@ -86,7 +86,7 @@ abstract class ProducerExecutor implements Executor {
                         (m, e) ->
                             observer.accept(
                                 System.currentTimeMillis() - m.timestamp(),
-                                m.serializedValueSize())));
+                                m.serializedValueSize() + m.serializedKeySize())));
         return State.RUNNING;
       }
     };
diff --git a/app/src/main/java/org/astraea/app/producer/Builder.java b/app/src/main/java/org/astraea/app/producer/Builder.java
index 4a8c7cef..da59a9d9 100644
--- a/app/src/main/java/org/astraea/app/producer/Builder.java
+++ b/app/src/main/java/org/astraea/app/producer/Builder.java
@@ -70,8 +70,10 @@ public class Builder<Key, Value> {
   }
 
   public Builder<Key, Value> partitionClassName(String partitionClassName) {
-    return config(
-        ProducerConfig.PARTITIONER_CLASS_CONFIG, Objects.requireNonNull(partitionClassName));
+    // Don't set partitioner to make sure DefaultPartitioner (Kafka version 3.2 and before) or
+    // BuiltInPartitioner (Kafka version 3.3) both works.
+    if (partitionClassName == null || partitionClassName.isEmpty()) return this;
+    return config(ProducerConfig.PARTITIONER_CLASS_CONFIG, partitionClassName);
   }
 
   public Builder<Key, Value> compression(Compression compression) {
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 50f66897..5cf41702 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 # please update the gradle version in gradlew too
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.5-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
index 24bdac75..801309b2 100755
--- a/gradlew
+++ b/gradlew
@@ -116,7 +116,7 @@ esac
 
 # gradle-wrapper.jar is not in git repo anymore, so we have to download it now.
 if [ ! -e "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" ]; then
-  if ! curl -s -S --retry 3 -L -o "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" "https://raw.githubusercontent.com/gradle/gradle/v7.4.2/gradle/wrapper/gradle-wrapper.jar"; then
+  if ! curl -s -S --retry 3 -L -o "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" "https://raw.githubusercontent.com/gradle/gradle/v7.5.0/gradle/wrapper/gradle-wrapper.jar"; then
     rm -f "$APP_HOME/gradle/wrapper/gradle-wrapper.jar"
     echo "failed to download gradle-wrapper.jar"
     exit 1


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java b/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java
index 65d22c42..e8e82352 100644
--- a/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java
@@ -16,8 +16,7 @@
  */
 package org.astraea.app.balancer;
 
-import java.util.stream.Collectors;
-import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.app.cost.ClusterInfoProvider;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -27,7 +26,7 @@ class RebalancePlanProposalTest {
   @Test
   void testBuild() {
     final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
-    final var thisAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var thisAllocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var build =
         RebalancePlanProposal.builder()
             .withRebalancePlan(thisAllocation)
@@ -41,10 +40,8 @@ class RebalancePlanProposalTest {
 
     Assertions.assertTrue(build.rebalancePlan().isPresent());
     final var thatAllocation = build.rebalancePlan().orElseThrow();
-    final var thisTps =
-        thisAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
-    final var thatTps =
-        thatAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+    final var thisTps = thisAllocation.topicPartitions();
+    final var thatTps = thatAllocation.topicPartitions();
     Assertions.assertEquals(thisTps, thatTps);
     thisTps.forEach(
         tp ->
@@ -63,7 +60,7 @@ class RebalancePlanProposalTest {
     // A builder should only build once. If a builder can build multiple times then it will have to
     // do much copy work once a new build is requested. This will harm performance.
     final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
-    final var logAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var logAllocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var build = RebalancePlanProposal.builder().withRebalancePlan(logAllocation);
 
     Assertions.assertDoesNotThrow(build::build);
@@ -73,7 +70,7 @@ class RebalancePlanProposalTest {
   @Test
   void testNoModifyAfterBuild() {
     final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
-    final var logAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var logAllocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var build = RebalancePlanProposal.builder().withRebalancePlan(logAllocation);
 
     RebalancePlanProposal proposal = build.build();
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
index fbb7280c..b1cffa4f 100644
--- a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
@@ -24,7 +24,6 @@ import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
-import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
 import org.astraea.app.balancer.log.LogPlacement;
 import org.astraea.app.common.Utils;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -40,8 +39,7 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
       final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
       admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 2).create();
       Utils.sleep(Duration.ofSeconds(2));
-      final var originalAllocation =
-          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var originalAllocation = ClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
       Utils.sleep(Duration.ofSeconds(3));
       final var broker0 = 0;
       final var broker1 = 1;
@@ -53,19 +51,16 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
           IntStream.range(0, 10)
               .mapToObj(i -> new TopicPartition(topicName, i))
               .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
-      final var expectedAllocation = LayeredClusterLogAllocation.of(allocationMap);
-      final var expectedTopicPartition =
-          expectedAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var expectedAllocation = ClusterLogAllocation.of(allocationMap);
+      final var expectedTopicPartition = expectedAllocation.topicPartitions();
       final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
 
       // act
       new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
 
       // assert
-      final var currentAllocation =
-          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
-      final var currentTopicPartition =
-          currentAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var currentAllocation = ClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var currentTopicPartition = currentAllocation.topicPartitions();
       Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
       expectedTopicPartition.forEach(
           topicPartition ->
@@ -90,8 +85,7 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
       final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
       admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 3).create();
       Utils.sleep(Duration.ofSeconds(2));
-      final var originalAllocation =
-          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var originalAllocation = ClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
       Utils.sleep(Duration.ofSeconds(3));
       final var broker0 = 0;
       final var broker1 = 1;
@@ -102,19 +96,16 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
           IntStream.range(0, 10)
               .mapToObj(i -> new TopicPartition(topicName, i))
               .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
-      final var expectedAllocation = LayeredClusterLogAllocation.of(allocationMap);
-      final var expectedTopicPartition =
-          expectedAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var expectedAllocation = ClusterLogAllocation.of(allocationMap);
+      final var expectedTopicPartition = expectedAllocation.topicPartitions();
       final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
 
       // act
       new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
 
       // assert
-      final var currentAllocation =
-          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
-      final var currentTopicPartition =
-          currentAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var currentAllocation = ClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var currentTopicPartition = currentAllocation.topicPartitions();
       Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
       expectedTopicPartition.forEach(
           topicPartition ->
diff --git a/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java b/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
index 16a28918..a2b1d583 100644
--- a/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
@@ -20,7 +20,7 @@ import java.time.Duration;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Collectors;
-import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.app.common.Utils;
 import org.astraea.app.cost.ClusterInfoProvider;
 import org.junit.jupiter.api.Assertions;
@@ -48,7 +48,7 @@ class ShufflePlanGeneratorTest {
   @ValueSource(ints = {3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 301})
   void testMovement(int shuffle) {
     final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(30, 30, 20, 5);
-    final var allocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var allocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var shufflePlanGenerator = new ShufflePlanGenerator(() -> shuffle);
 
     shufflePlanGenerator
@@ -57,10 +57,8 @@ class ShufflePlanGeneratorTest {
         .forEach(
             proposal -> {
               final var that = proposal.rebalancePlan().orElseThrow();
-              final var thisTps =
-                  allocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
-              final var thatTps =
-                  that.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+              final var thisTps = allocation.topicPartitions();
+              final var thatTps = that.topicPartitions();
               final var thisMap =
                   thisTps.stream()
                       .collect(Collectors.toUnmodifiableMap(x -> x, allocation::logPlacements));
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java b/app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java
similarity index 59%
rename from app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
rename to app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java
index 42556139..3055f9eb 100644
--- a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java
@@ -28,19 +28,19 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
-class LayeredClusterLogAllocationTest {
+class ClusterLogAllocationTest {
 
   @Test
   void creation() {
     // empty replica set
     var badAllocation0 = Map.of(TopicPartition.of("topic", "0"), List.<LogPlacement>of());
     Assertions.assertThrows(
-        IllegalArgumentException.class, () -> LayeredClusterLogAllocation.of(badAllocation0));
+        IllegalArgumentException.class, () -> ClusterLogAllocation.of(badAllocation0));
 
     // partial topic/partition
     var badAllocation1 = Map.of(TopicPartition.of("topic", "999"), List.of(LogPlacement.of(1001)));
     Assertions.assertThrows(
-        IllegalArgumentException.class, () -> LayeredClusterLogAllocation.of(badAllocation1));
+        IllegalArgumentException.class, () -> ClusterLogAllocation.of(badAllocation1));
 
     // duplicate replica
     var badAllocation2 =
@@ -48,17 +48,17 @@ class LayeredClusterLogAllocationTest {
             TopicPartition.of("topic", "0"),
             List.of(LogPlacement.of(1001), LogPlacement.of(1001), LogPlacement.of(1001)));
     Assertions.assertThrows(
-        IllegalArgumentException.class, () -> LayeredClusterLogAllocation.of(badAllocation2));
+        IllegalArgumentException.class, () -> ClusterLogAllocation.of(badAllocation2));
   }
 
   @Test
   void migrateReplica() {
     final var fakeClusterInfo =
         ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 1, (i) -> Set.of("topic"));
-    final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    var clusterLogAllocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var sourceTopicPartition = TopicPartition.of("topic", "0");
 
-    clusterLogAllocation.migrateReplica(sourceTopicPartition, 0, 1);
+    clusterLogAllocation = clusterLogAllocation.migrateReplica(sourceTopicPartition, 0, 1);
 
     Assertions.assertEquals(
         1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
@@ -68,7 +68,6 @@ class LayeredClusterLogAllocationTest {
             .get(0)
             .logDirectory()
             .orElse(null));
-    Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
   }
 
   @ParameterizedTest
@@ -78,10 +77,11 @@ class LayeredClusterLogAllocationTest {
     dataDirectory = dataDirectory.equals("null") ? null : dataDirectory;
     final var fakeClusterInfo =
         ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 1, (i) -> Set.of("topic"));
-    final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    var clusterLogAllocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var sourceTopicPartition0 = TopicPartition.of("topic", "0");
 
-    clusterLogAllocation.migrateReplica(sourceTopicPartition0, 0, 1, dataDirectory);
+    clusterLogAllocation =
+        clusterLogAllocation.migrateReplica(sourceTopicPartition0, 0, 1, dataDirectory);
 
     Assertions.assertEquals(
         1, clusterLogAllocation.logPlacements(sourceTopicPartition0).get(0).broker());
@@ -94,7 +94,8 @@ class LayeredClusterLogAllocationTest {
             .orElse(null));
 
     final var sourceTopicPartition1 = TopicPartition.of("topic", "0");
-    clusterLogAllocation.migrateReplica(sourceTopicPartition1, 1, 1, dataDirectory);
+    clusterLogAllocation =
+        clusterLogAllocation.migrateReplica(sourceTopicPartition1, 1, 1, dataDirectory);
     Assertions.assertEquals(
         1, clusterLogAllocation.logPlacements(sourceTopicPartition1).get(0).broker());
     Assertions.assertEquals(
@@ -104,30 +105,27 @@ class LayeredClusterLogAllocationTest {
             .get(0)
             .logDirectory()
             .orElse(null));
-
-    Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
   }
 
   @Test
   void letReplicaBecomeLeader() {
     final var fakeClusterInfo =
         ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 2, (i) -> Set.of("topic"));
-    final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    var clusterLogAllocation = ClusterLogAllocation.of(fakeClusterInfo);
     final var sourceTopicPartition = TopicPartition.of("topic", "0");
 
-    clusterLogAllocation.letReplicaBecomeLeader(sourceTopicPartition, 1);
+    clusterLogAllocation = clusterLogAllocation.letReplicaBecomeLeader(sourceTopicPartition, 1);
 
     Assertions.assertEquals(
         1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
     Assertions.assertEquals(
         0, clusterLogAllocation.logPlacements(sourceTopicPartition).get(1).broker());
-    Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
   }
 
   @Test
   void logPlacements() {
     final var allocation =
-        LayeredClusterLogAllocation.of(
+        ClusterLogAllocation.of(
             Map.of(TopicPartition.of("topic", "0"), List.of(LogPlacement.of(0, "/nowhere"))));
 
     Assertions.assertEquals(1, allocation.logPlacements(TopicPartition.of("topic", "0")).size());
@@ -140,25 +138,24 @@ class LayeredClusterLogAllocationTest {
             .get(0)
             .logDirectory()
             .orElseThrow());
-    Assertions.assertNull(allocation.logPlacements(TopicPartition.of("no", "0")));
+    Assertions.assertEquals(0, allocation.logPlacements(TopicPartition.of("no", "0")).size());
     allocation.logPlacements(TopicPartition.of("no", "0"));
   }
 
   @Test
   void topicPartitionStream() {
     final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 3);
-    final var allocation0 = LayeredClusterLogAllocation.of(fakeClusterInfo);
-    final var allocation1 = LayeredClusterLogAllocation.of(allocation0);
-    fakeClusterInfo.topics().stream()
-        .flatMap(x -> fakeClusterInfo.replicas(x).stream())
-        .forEach(
-            replica ->
-                allocation1.letReplicaBecomeLeader(
-                    TopicPartition.of(replica.topic(), Integer.toString(replica.partition())), 1));
+    var allocation = ClusterLogAllocation.of(fakeClusterInfo);
+    for (var replica :
+        fakeClusterInfo.topics().stream()
+            .flatMap(x -> fakeClusterInfo.replicas(x).stream())
+            .collect(Collectors.toUnmodifiableList()))
+      allocation =
+          allocation.letReplicaBecomeLeader(
+              TopicPartition.of(replica.topic(), Integer.toString(replica.partition())), 1);
 
     final var allTopicPartitions =
-        allocation1
-            .topicPartitionStream()
+        allocation.topicPartitions().stream()
             .sorted(
                 Comparator.comparing(TopicPartition::topic)
                     .thenComparing(TopicPartition::partition))
@@ -176,73 +173,46 @@ class LayeredClusterLogAllocationTest {
     Assertions.assertEquals(expectedTopicPartitions, allTopicPartitions);
   }
 
-  @Test
-  void lockWorks() {
-    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 3);
-    final var allocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
-    final var toModify = allocation.topicPartitionStream().findFirst().orElseThrow();
-
-    // can modify before lock
-    Assertions.assertDoesNotThrow(() -> allocation.migrateReplica(toModify, 0, 9));
-    Assertions.assertDoesNotThrow(() -> allocation.migrateReplica(toModify, 1, 8, "somewhere"));
-    Assertions.assertDoesNotThrow(() -> allocation.letReplicaBecomeLeader(toModify, 2));
-
-    final var extended = LayeredClusterLogAllocation.of(allocation);
-
-    // cannot modify after some other layer rely on it
-    Assertions.assertThrows(
-        IllegalStateException.class, () -> allocation.migrateReplica(toModify, 9, 0));
-    Assertions.assertThrows(
-        IllegalStateException.class, () -> allocation.migrateReplica(toModify, 0, 9, "somewhere"));
-    Assertions.assertThrows(
-        IllegalStateException.class, () -> allocation.letReplicaBecomeLeader(toModify, 0));
-
-    // the extended one can modify
-    Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 9, 0));
-    Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 8, 1, "somewhere"));
-    Assertions.assertDoesNotThrow(() -> extended.letReplicaBecomeLeader(toModify, 0));
-  }
-
   @Test
   void findNonFulfilledAllocation() {
     final var clusterInfo = ClusterInfoProvider.fakeClusterInfo(3, 10, 10, 2);
-    final var a = LayeredClusterLogAllocation.of(clusterInfo);
-    final var b = LayeredClusterLogAllocation.of(clusterInfo);
-    Assertions.assertEquals(Set.of(), ClusterLogAllocation.findNonFulfilledAllocation(a, b));
+    Assertions.assertEquals(
+        Set.of(),
+        ClusterLogAllocation.findNonFulfilledAllocation(
+            ClusterLogAllocation.of(clusterInfo), ClusterLogAllocation.of(clusterInfo)));
 
-    final var source = LayeredClusterLogAllocation.of(clusterInfo);
-    final var oneTopicPartition = source.topicPartitionStream().findFirst().orElseThrow();
-    final var twoTopicPartition = source.topicPartitionStream().skip(1).findFirst().orElseThrow();
+    final var source = ClusterLogAllocation.of(clusterInfo);
+    final var oneTopicPartition = source.topicPartitions().iterator().next();
+    final var twoTopicPartition = source.topicPartitions().stream().skip(1).findFirst().get();
 
-    final var target0 = LayeredClusterLogAllocation.of(a);
-    target0.migrateReplica(oneTopicPartition, 0, 0, "/somewhere");
+    final var target0 = source.migrateReplica(oneTopicPartition, 0, 0, "/somewhere");
     Assertions.assertEquals(
         Set.of(oneTopicPartition),
         ClusterLogAllocation.findNonFulfilledAllocation(source, target0));
 
-    final var target1 = LayeredClusterLogAllocation.of(a);
-    target1.migrateReplica(oneTopicPartition, 0, 2);
+    final var target1 = source.migrateReplica(oneTopicPartition, 0, 2);
     Assertions.assertEquals(
         Set.of(oneTopicPartition),
         ClusterLogAllocation.findNonFulfilledAllocation(source, target1));
 
-    final var target2 = LayeredClusterLogAllocation.of(a);
-    target2.letReplicaBecomeLeader(oneTopicPartition, 1);
+    final var target2 = source.letReplicaBecomeLeader(oneTopicPartition, 1);
     Assertions.assertEquals(
         Set.of(oneTopicPartition),
         ClusterLogAllocation.findNonFulfilledAllocation(source, target2));
 
-    final var target3 = LayeredClusterLogAllocation.of(a);
-    target3.migrateReplica(oneTopicPartition, 0, 2);
-    target3.migrateReplica(oneTopicPartition, 2, 2, "/somewhere");
-    target3.letReplicaBecomeLeader(twoTopicPartition, 1);
+    final var target3 =
+        source
+            .migrateReplica(oneTopicPartition, 0, 2)
+            .migrateReplica(oneTopicPartition, 2, 2, "/somewhere")
+            .letReplicaBecomeLeader(twoTopicPartition, 1);
     Assertions.assertEquals(
         Set.of(oneTopicPartition, twoTopicPartition),
         ClusterLogAllocation.findNonFulfilledAllocation(source, target3));
 
-    final var map4 = a.topicPartitionStream().collect(Collectors.toMap(x -> x, a::logPlacements));
+    final var map4 =
+        source.topicPartitions().stream().collect(Collectors.toMap(x -> x, source::logPlacements));
     map4.put(new TopicPartition("NewTopic", 0), List.of(LogPlacement.of(0, "?")));
-    final var target4 = LayeredClusterLogAllocation.of(map4);
+    final var target4 = ClusterLogAllocation.of(map4);
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () -> ClusterLogAllocation.findNonFulfilledAllocation(source, target4));
diff --git a/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java b/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
index 3f5e5193..c197dfa9 100644
--- a/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
+++ b/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
@@ -17,6 +17,8 @@
 package org.astraea.app.performance;
 
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicLong;
 import org.astraea.app.common.DataUnit;
 import org.astraea.app.common.Utils;
 import org.junit.jupiter.api.Assertions;
@@ -28,11 +30,7 @@ public class DataSupplierTest {
   void testDuration() {
     var dataSupplier =
         DataSupplier.of(
-            ExeTime.of("2s"),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100));
+            ExeTime.of("2s"), () -> 1L, () -> 4L, () -> 1L, () -> 10L, DataUnit.KiB.of(100));
     Assertions.assertTrue(dataSupplier.get().hasData());
     Utils.sleep(Duration.ofSeconds(3));
     Assertions.assertFalse(dataSupplier.get().hasData());
@@ -43,10 +41,11 @@ public class DataSupplierTest {
     var dataSupplier =
         DataSupplier.of(
             ExeTime.of("2records"),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100));
+            () -> DataUnit.KiB.of(2).measurement(DataUnit.Byte).longValue(),
+            () -> 10L,
+            () -> DataUnit.KiB.of(100).measurement(DataUnit.Byte).longValue(),
+            () -> 10L,
+            DataUnit.KiB.of(102));
     Assertions.assertTrue(dataSupplier.get().hasData());
     Assertions.assertTrue(dataSupplier.get().hasData());
     Assertions.assertFalse(dataSupplier.get().hasData());
@@ -57,14 +56,17 @@ public class DataSupplierTest {
     var dataSupplier =
         DataSupplier.of(
             ExeTime.of("10s"),
-            DistributionType.FIXED.create(9),
-            DataUnit.KiB.of(100),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100));
-    var data = dataSupplier.get();
-    Assertions.assertTrue(data.hasData());
-    // key content is fixed to "9", so the size is 1 byte
-    Assertions.assertEquals(1, data.key().length);
+            () -> 1L,
+            () -> DataUnit.Byte.of(20).measurement(DataUnit.Byte).longValue(),
+            () -> 2L,
+            () -> DataUnit.KiB.of(100).measurement(DataUnit.Byte).longValue(),
+            DataUnit.KiB.of(200));
+    var data1 = dataSupplier.get();
+    Assertions.assertTrue(data1.hasData());
+    var data2 = dataSupplier.get();
+    Assertions.assertTrue(data2.hasData());
+    // key content is fixed, the keys are the same
+    Assertions.assertEquals(data1.key(), data2.key());
   }
 
   @Test
@@ -72,9 +74,10 @@ public class DataSupplierTest {
     var dataSupplier =
         DataSupplier.of(
             ExeTime.of("10s"),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100),
-            DistributionType.FIXED.create(0),
+            () -> 1L,
+            () -> DataUnit.Byte.of(20).measurement(DataUnit.Byte).longValue(),
+            () -> 2L,
+            () -> DataUnit.KiB.of(100).measurement(DataUnit.Byte).longValue(),
             DataUnit.KiB.of(100));
     var data = dataSupplier.get();
     Assertions.assertTrue(data.hasData());
@@ -82,20 +85,96 @@ public class DataSupplierTest {
     Assertions.assertEquals(102400, data.value().length);
   }
 
+  @Test
+  void testKeyDistribution() {
+    var counter = new AtomicLong(0L);
+    var counter2 = new AtomicLong(0L);
+
+    // Round-robin on 2 keys. Round-robin key size between 100Byte and 101Byte
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"),
+            () -> counter.getAndIncrement() % 2,
+            () -> 100L + counter2.getAndIncrement(),
+            () -> 10L,
+            () -> 10L,
+            DataUnit.KiB.of(200));
+    var data1 = dataSupplier.get();
+    Assertions.assertTrue(data1.hasData());
+    var data2 = dataSupplier.get();
+    Assertions.assertTrue(data2.hasData());
+    var data3 = dataSupplier.get();
+    Assertions.assertTrue(data3.hasData());
+
+    // The key of data1 and data2 should have size 100 bytes and 101 bytes respectively.
+    Assertions.assertEquals(100, data1.key().length);
+    Assertions.assertEquals(101, data2.key().length);
+    // Round-robin key distribution with 2 possible key.
+    Assertions.assertEquals(data1.key(), data3.key());
+
+    // Round-robin on 2 keys. Fixed key size to 100 bytes.
+    dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"),
+            () -> counter.getAndIncrement() % 2,
+            () -> 100L,
+            () -> 10L,
+            () -> 10L,
+            DataUnit.KiB.of(200));
+    data1 = dataSupplier.get();
+    Assertions.assertTrue(data1.hasData());
+    data2 = dataSupplier.get();
+    Assertions.assertTrue(data2.hasData());
+    // Same size but different content
+    Assertions.assertEquals(100, data1.key().length);
+    Assertions.assertEquals(100, data2.key().length);
+    Assertions.assertFalse(Arrays.equals(data1.key(), data2.key()));
+  }
+
   @Test
   void testDistributedValueSize() {
+    var counter = new AtomicLong(0);
+    var counter2 = new AtomicLong(0);
+
+    // Round-robin on 2 values. Round-robin value size between 100Byte and 101Byte
     var dataSupplier =
         DataSupplier.of(
             ExeTime.of("10s"),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100),
-            DistributionType.FIXED.create(10),
+            () -> 10L,
+            () -> 10L,
+            () -> counter.getAndIncrement() % 2,
+            () -> 100L + counter2.getAndIncrement(),
             DataUnit.KiB.of(100));
-    var data = dataSupplier.get();
-    Assertions.assertTrue(data.hasData());
-    // initial value size is 100KB and the distributed is fixed to 10, so the final size is between
-    // (102400 - 10, 102400 + 10)
-    Assertions.assertTrue(data.value().length >= 102400 - 10 && data.value().length <= 102400 + 10);
+    var data1 = dataSupplier.get();
+    Assertions.assertTrue(data1.hasData());
+    var data2 = dataSupplier.get();
+    Assertions.assertTrue(data2.hasData());
+    var data3 = dataSupplier.get();
+    Assertions.assertTrue(data3.hasData());
+
+    // The value of data1 and data2 should have size 100 bytes and 101 bytes respectively.
+    Assertions.assertEquals(100, data1.value().length);
+    Assertions.assertEquals(101, data2.value().length);
+    // Round-robin value distribution with 2 possible value.
+    Assertions.assertEquals(data1.value(), data3.value());
+
+    // Round-robin on 2 values. Fixed value size.
+    dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"),
+            () -> 10L,
+            () -> 10L,
+            () -> counter.getAndIncrement() % 2,
+            () -> 100L,
+            DataUnit.KiB.of(100));
+    data1 = dataSupplier.get();
+    Assertions.assertTrue(data1.hasData());
+    data2 = dataSupplier.get();
+    Assertions.assertTrue(data2.hasData());
+    // Same size but different content
+    Assertions.assertEquals(100, data1.value().length);
+    Assertions.assertEquals(100, data2.value().length);
+    Assertions.assertFalse(Arrays.equals(data1.value(), data2.value()));
   }
 
   @Test
@@ -103,9 +182,10 @@ public class DataSupplierTest {
     var dataSupplier =
         DataSupplier.of(
             ExeTime.of("10s"),
-            DistributionType.FIXED.create(10),
-            DataUnit.KiB.of(100),
-            DistributionType.FIXED.create(0),
+            () -> 10L,
+            () -> DataUnit.KiB.of(50).measurement(DataUnit.Byte).longValue(),
+            () -> 10L,
+            () -> DataUnit.KiB.of(50).measurement(DataUnit.Byte).longValue(),
             DataUnit.KiB.of(150));
     // total: 100KB, limit: 150KB -> no throttle
     Assertions.assertTrue(dataSupplier.get().hasData());
@@ -114,4 +194,25 @@ public class DataSupplierTest {
     // throttled
     Assertions.assertFalse(dataSupplier.get().hasData());
   }
+
+  @Test
+  void testNoKey() {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"), () -> 10L, () -> 0L, () -> 10L, () -> 10L, DataUnit.KiB.of(200));
+
+    var data = dataSupplier.get();
+    Assertions.assertTrue(data.hasData());
+    Assertions.assertNull(data.key());
+  }
+
+  @Test
+  void testNoValue() {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"), () -> 10L, () -> 10L, () -> 10L, () -> 0L, DataUnit.KiB.of(200));
+    var data = dataSupplier.get();
+    Assertions.assertTrue(data.hasData());
+    Assertions.assertNull(data.value());
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/performance/ManagerTest.java b/app/src/test/java/org/astraea/app/performance/ManagerTest.java
index 99e68545..caad22f0 100644
--- a/app/src/test/java/org/astraea/app/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/app/performance/ManagerTest.java
@@ -21,28 +21,6 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class ManagerTest {
-
-  @Test
-  void testGetKey() {
-    var argument = new Performance.Argument();
-
-    argument.keyDistributionType = DistributionType.UNIFORM;
-    var manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().length > 0);
-
-    argument.keyDistributionType = DistributionType.ZIPFIAN;
-    manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().length > 0);
-
-    argument.keyDistributionType = DistributionType.LATEST;
-    manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().length > 0);
-
-    argument.keyDistributionType = DistributionType.FIXED;
-    manager = new Manager(argument, List.of(), List.of());
-    Assertions.assertTrue(manager.getKey().length > 0);
-  }
-
   @Test
   void testConsumerDone() {
     var argument = new Performance.Argument();
diff --git a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index 6036e09d..c9872082 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -100,7 +100,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     Metrics metrics = new Metrics();
     var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
     var param = new Performance.Argument();
-    param.sizeDistributionType = DistributionType.FIXED;
+    param.valueDistributionType = DistributionType.FIXED;
     try (Executor executor =
         Performance.consumerExecutor(
             Consumer.forTopics(Set.of(topicName)).bootstrapServers(bootstrapServers()).build(),
@@ -149,12 +149,16 @@ public class PerformanceTest extends RequireBrokerCluster {
       "1",
       "--run.until",
       "1000records",
-      "--record.size",
+      "--value.size",
       "10KiB",
+      "--value.distribution",
+      "uniform",
       "--partitioner",
       "org.astraea.partitioner.smooth.SmoothWeightPartitioner",
       "--compression",
       "lz4",
+      "--key.size",
+      "4Byte",
       "--key.distribution",
       "zipfian",
       "--specify.broker",
@@ -189,25 +193,34 @@ public class PerformanceTest extends RequireBrokerCluster {
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments7));
 
-    String[] arguments8 = {"--bootstrap.servers", "localhost:9092", "--record.size", "1"};
+    String[] arguments8 = {"--bootstrap.servers", "localhost:9092", "--key.distribution", "f"};
+    Assertions.assertThrows(
+        ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments8));
+
+    String[] arguments9 = {"--bootstrap.servers", "localhost:9092", "--key.size", "1"};
     Assertions.assertThrows(
         IllegalArgumentException.class,
-        () -> Argument.parse(new Performance.Argument(), arguments8));
+        () -> Argument.parse(new Performance.Argument(), arguments9));
 
-    String[] arguments10 = {"--bootstrap.servers", "localhost:9092", "--partitioner", ""};
+    String[] arguments10 = {"--bootstrap.servers", "localhost:9092", "--value.distribution", "u"};
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments10));
 
-    String[] arguments11 = {"--bootstrap.servers", "localhost:9092", "--compression", ""};
+    String[] arguments11 = {"--bootstrap.servers", "localhost:9092", "--value.size", "2"};
     Assertions.assertThrows(
-        ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments11));
+        IllegalArgumentException.class,
+        () -> Argument.parse(new Performance.Argument(), arguments11));
 
-    String[] arguments12 = {"--bootstrap.servers", "localhost:9092", "--key.distribution", ""};
+    String[] arguments12 = {"--bootstrap.servers", "localhost:9092", "--partitioner", ""};
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments12));
 
-    String[] arguments13 = {"--bootstrap.servers", "localhost:9092", "--specify.broker", ""};
+    String[] arguments13 = {"--bootstrap.servers", "localhost:9092", "--compression", ""};
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments13));
+
+    String[] arguments14 = {"--bootstrap.servers", "localhost:9092", "--specify.broker", ""};
+    Assertions.assertThrows(
+        ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments14));
   }
 }