Repository: skiptests/astraea
Pull request #444 (2022-07-01 12:41:04)

Set no partitioner as default
在測試Kafka的[新功能](https://github.com/skiptests/astraea/issues/390#issuecomment-1171978186)時，發現performance 需要 "不設定partitioner"，也需要 "傳送沒有key的record"的功能，所以更改預設值，以及新增了一個參數，用來指定record不要使用key。

Kafka 的預設改成 "不使用partitioner"，因此，我們的預設值也要改成不設定partitioner，才可以測試預設的方法。

1. 預設partitioner 改成不設定
2. 刪除參數："--record.size"
3. 新增參數："--value.distribution" 決定value的大小上限以及分佈種類
4. 修改參數："--key.distribution" 改成 "\<distributionType name\>\<dataSize\>"的輸入，並修改製作key的邏輯：key的大小將會跟隨分佈而不同。

chia7712: > 也需要 "傳送沒有key的record"的功能

這是一個方式，另一個方式是設定`partitioner.ignore.keys`=true，因此我會建議使用後者
chia7712: > 是的，不過現在的kafka版本中，沒有這項設定，這個設定是在KIP-794之後才有的。所以這裡才在我們的producer中決定key的有無。


ok，那就以沒有key來處理好了，這也算是一個情境

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/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/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));
   }
 }