Repository: skiptests/astraea
Pull request #322 (2022-05-09 16:31:20)

Refactor ProducerExecutor
改善了以下幾個點：
1. 避免`Manager`涵蓋太多功能導致難以測試，原本的`Manager`掌管了producer/consuemr的資料量、也掌管了如何產生資料、更掌管了什麼時候要停止各個consumer/producer
2. throttle的使用方式改善，原本的方式是直接卡著等到可以回傳資料，但這樣會導致呼叫者只能乾等，這隻PR增加了幾個狀態列讓呼叫者可以自行決定大概何時要再來拿、或是過程要不要做一些事情
3. 避免建立多餘的producer，原本會同時建立兩個producer，一個負責transaction、一個負責一般的資料。然而實際上我們同時只會用一個...
4. 增加更多的測試

chia7712: @chinghongfang 我留了一個後續工作給你，就是把consumer executor那邊重構，讓`Manager`這個角色可以更專注在“某項工作”上，而非包山包海的管裡
chia7712: > 不過實際測試後發現，throughput會卡在100多KiB/sec，無法上去。

可以貼一下完整的數據嗎？我這邊沒有看到這麼慘烈的數據XDD
chinghongfang: 現在沒問題了。

似乎是因為使用transaction produce，因為上一個commit是使用`transaction size > 0`來判斷是否要使用transactional producer，而`transaction size` 預設是1。
chia7712: > 似乎是因為使用transaction produce，因為上一個commit是使用transaction size > 0來判斷是否要使用transactional producer，而transaction size 預設是1。

感謝驗證，不過這也暗示了用transaction後的效能很慘烈？可否幫我開個議題，然後把你看到的數據貼上去？

Patch:
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
index 298ec86b..733c31c2 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -141,5 +141,29 @@ public final class Utils {
     return (lastTime + Duration.ofSeconds(second).toMillis()) < System.currentTimeMillis();
   }
 
+  public static void sleep(Duration duration) {
+    try {
+      TimeUnit.MILLISECONDS.sleep(duration.toMillis());
+    } catch (InterruptedException ignored) {
+    }
+  }
+
+  public static String randomString(int len) {
+    StringBuilder string = new StringBuilder(randomString());
+    while (string.length() < len) {
+      string.append(string).append(randomString());
+    }
+    return string.substring(0, len);
+  }
+
+  /**
+   * a random string based on uuid without "-"
+   *
+   * @return random string
+   */
+  public static String randomString() {
+    return java.util.UUID.randomUUID().toString().replaceAll("-", "");
+  }
+
   private Utils() {}
 }
diff --git a/app/src/main/java/org/astraea/concurrent/ThreadPool.java b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
index ec32bfcb..4db9ef26 100644
--- a/app/src/main/java/org/astraea/concurrent/ThreadPool.java
+++ b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
@@ -14,6 +14,26 @@ import java.util.concurrent.atomic.AtomicBoolean;
  */
 public interface ThreadPool extends AutoCloseable {
 
+  // nothing to run.
+  ThreadPool EMPTY =
+      new ThreadPool() {
+        @Override
+        public void close() {}
+
+        @Override
+        public void waitAll() {}
+
+        @Override
+        public boolean isClosed() {
+          return true;
+        }
+
+        @Override
+        public int size() {
+          return 0;
+        }
+      };
+
   /** close all running threads. */
   @Override
   void close();
@@ -39,12 +59,13 @@ public interface ThreadPool extends AutoCloseable {
       return executors(List.of(executor));
     }
 
-    public Builder executors(Collection<Executor> executors) {
+    public Builder executors(Collection<? extends Executor> executors) {
       this.executors.addAll(Objects.requireNonNull(executors));
       return this;
     }
 
     public ThreadPool build() {
+      if (executors.isEmpty()) return EMPTY;
       var closed = new AtomicBoolean(false);
       var latch = new CountDownLatch(executors.size());
       var service = Executors.newFixedThreadPool(executors.size());
diff --git a/app/src/main/java/org/astraea/performance/DataSupplier.java b/app/src/main/java/org/astraea/performance/DataSupplier.java
new file mode 100644
index 00000000..657949a1
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/DataSupplier.java
@@ -0,0 +1,152 @@
+package org.astraea.performance;
+
+import java.util.Arrays;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+import org.astraea.utils.DataSize;
+import org.astraea.utils.DataUnit;
+
+@FunctionalInterface
+interface DataSupplier extends Supplier<DataSupplier.Data> {
+
+  static Data data(byte[] key, byte[] value) {
+    return new Data() {
+      @Override
+      public boolean done() {
+        return false;
+      }
+
+      @Override
+      public boolean throttled() {
+        return false;
+      }
+
+      @Override
+      public byte[] key() {
+        return key;
+      }
+
+      @Override
+      public byte[] value() {
+        return value;
+      }
+    };
+  }
+
+  Data NO_MORE_DATA =
+      new Data() {
+        @Override
+        public boolean done() {
+          return true;
+        }
+
+        @Override
+        public boolean throttled() {
+          return false;
+        }
+
+        @Override
+        public byte[] key() {
+          throw new IllegalStateException("there is no data");
+        }
+
+        @Override
+        public byte[] value() {
+          throw new IllegalStateException("there is no data");
+        }
+      };
+
+  Data THROTTLED_DATA =
+      new Data() {
+        @Override
+        public boolean done() {
+          return false;
+        }
+
+        @Override
+        public boolean throttled() {
+          return true;
+        }
+
+        @Override
+        public byte[] key() {
+          throw new IllegalStateException("it is throttled");
+        }
+
+        @Override
+        public byte[] value() {
+          throw new IllegalStateException("it is throttled");
+        }
+      };
+
+  interface Data {
+
+    /** @return true if there is no data. */
+    boolean done();
+
+    /** @return true if there are some data, but it is throttled now. */
+    boolean throttled();
+
+    /** @return true if there is accessible data */
+    default boolean hasData() {
+      return !done() && !throttled();
+    }
+
+    /** @return key or throw exception if there is no data, or it is throttled now */
+    byte[] key();
+
+    /** @return value or throw exception if there is no data, or it is throttled now */
+    byte[] value();
+  }
+
+  static DataSupplier of(
+      ExeTime exeTime,
+      Supplier<Long> keyDistribution,
+      DataSize valueSize,
+      Supplier<Long> valueDistribution,
+      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;
+
+      synchronized boolean checkAndAdd(int payloadLength) {
+        if (System.currentTimeMillis() - intervalStart > 1000) {
+          intervalStart = System.currentTimeMillis();
+          payloadBytes = payloadLength;
+          return true;
+        } else if (payloadBytes < throughput.measurement(DataUnit.Byte).longValue()) {
+          payloadBytes += payloadLength;
+          return true;
+        } else {
+          return false;
+        }
+      }
+
+      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);
+      }
+
+      public byte[] key() {
+        return (String.valueOf(keyDistribution.get())).getBytes();
+      }
+
+      @Override
+      public Data get() {
+        if (exeTime.percentage(dataCount.getAndIncrement(), System.currentTimeMillis() - start)
+            >= 100D) return NO_MORE_DATA;
+        var key = key();
+        var value = value();
+        if (checkAndAdd(value.length)) return data(key, value);
+        return THROTTLED_DATA;
+      }
+    };
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index c5648f79..25743440 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -1,15 +1,7 @@
 package org.astraea.performance;
 
-import java.util.Arrays;
 import java.util.List;
-import java.util.Optional;
-import java.util.Random;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
-import org.astraea.Utils;
-import org.astraea.utils.DataSize;
 import org.astraea.utils.DataUnit;
 
 /**
@@ -19,17 +11,8 @@ import org.astraea.utils.DataUnit;
  */
 public class Manager {
   private final ExeTime exeTime;
-  private final CountDownLatch getAssignment;
-  private final AtomicInteger producerClosed;
-
   private final List<Metrics> producerMetrics, consumerMetrics;
-  private final long start = System.currentTimeMillis();
-  private final AtomicLong payloadNum = new AtomicLong(0);
   private final Supplier<Long> keyDistribution;
-  private final RandomContent randomContent;
-  private long intervalStart = System.currentTimeMillis();
-  private long payloadBytes = 0L;
-  private final DataSize throughput;
 
   /**
    * Used to manage producing/consuming.
@@ -51,49 +34,10 @@ public class Manager {
     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.getAssignment = new CountDownLatch(argument.consumers);
-    this.producerClosed = new AtomicInteger(argument.producers);
     this.producerMetrics = producerMetrics;
     this.consumerMetrics = consumerMetrics;
     this.exeTime = argument.exeTime;
-    this.throughput = argument.throughput;
     this.keyDistribution = argument.keyDistributionType.create(100000);
-    this.randomContent =
-        new RandomContent(
-            argument.recordSize,
-            argument.sizeDistributionType.create(
-                argument.recordSize.measurement(DataUnit.Byte).intValue()));
-  }
-
-  /**
-   * Generate random byte array in random/fixed length. Warning: This method will block when the
-   * throughput (content generating rate) is higher than the given number (argument.throughput).
-   *
-   * @return random byte array. When no payload should be generated, Optional.empty() is generated.
-   *     Whether the payload should be generated is determined in construct time. "Number of
-   *     records" and "execution time" are considered.
-   */
-  public Optional<byte[]> payload() {
-    if (exeTime.percentage(payloadNum.getAndIncrement(), System.currentTimeMillis() - start)
-        >= 100D) return Optional.empty();
-
-    var payload = randomContent.getContent();
-
-    Utils.waitFor(() -> checkAndAdd(payload.length));
-    return Optional.of(payload);
-  }
-
-  synchronized boolean checkAndAdd(int payloadLength) {
-    if (System.currentTimeMillis() - intervalStart > 1000) {
-      intervalStart = System.currentTimeMillis();
-      payloadBytes = payloadLength;
-      return true;
-    } else if (payloadBytes < throughput.measurement(DataUnit.Byte).longValue()) {
-      payloadBytes += payloadLength;
-      return true;
-    } else {
-      return false;
-    }
   }
 
   public long producedRecords() {
@@ -104,64 +48,17 @@ public class Manager {
     return consumerMetrics.stream().mapToLong(Metrics::num).sum();
   }
 
-  /** Used in "consumerRebalanceListener" callback. To */
-  public void countDownGetAssignment() {
-    this.getAssignment.countDown();
-  }
-
-  /** Called after producer is closed. For informing consumers there will be no new records */
-  public void producerClosed() {
-    this.producerClosed.decrementAndGet();
-  }
-
-  public void awaitPartitionAssignment() throws InterruptedException {
-    getAssignment.await();
-  }
-
   public ExeTime exeTime() {
     return exeTime;
   }
 
-  /** Check if all producer are closed. */
-  public boolean producedDone() {
-    return producerClosed.get() == 0;
-  }
-
   /** Check if we should keep consuming record. */
   public boolean consumedDone() {
-    return producedDone()
-        && (consumerMetrics.size() == 0 || consumedRecords() >= producedRecords());
+    return consumerMetrics.size() == 0 || consumedRecords() >= producedRecords();
   }
 
   /** Randomly choose a key according to the distribution. */
   public byte[] getKey() {
     return (String.valueOf(keyDistribution.get())).getBytes();
   }
-
-  /** Randomly generate content before {@link #getContent()} is called. */
-  private static class RandomContent {
-    private final Random rand = new Random();
-    private final DataSize dataSize;
-    private final Supplier<Long> distribution;
-    private final byte[] content;
-
-    /**
-     * @param dataSize The size of each random generated content in bytes.
-     * @param distribution Determine whether to fix the size of random generated content or random
-     *     size with specified distribution
-     */
-    public RandomContent(DataSize dataSize, Supplier<Long> distribution) {
-      this.dataSize = dataSize;
-      this.distribution = distribution;
-      content = new byte[dataSize.measurement(DataUnit.Byte).intValue()];
-    }
-
-    public byte[] getContent() {
-      // Randomly change one position of the content;
-      content[rand.nextInt(dataSize.measurement(DataUnit.Byte).intValue())] =
-          (byte) rand.nextInt(256);
-      return Arrays.copyOfRange(
-          content, (int) (distribution.get() % content.length), content.length);
-    }
-  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 95df6ec3..014e8b60 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -5,16 +5,18 @@ import java.io.IOException;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
-import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 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.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
@@ -32,7 +34,6 @@ import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
-import org.astraea.producer.TransactionalProducer;
 import org.astraea.topic.Compression;
 import org.astraea.topic.TopicAdmin;
 import org.astraea.utils.DataSize;
@@ -71,6 +72,51 @@ public class Performance {
     execute(org.astraea.argument.Argument.parse(new Argument(), args));
   }
 
+  private static DataSupplier dataSupplier(Performance.Argument argument) {
+    return DataSupplier.of(
+        argument.exeTime,
+        argument.keyDistributionType.create(10000),
+        argument.recordSize,
+        argument.sizeDistributionType.create(
+            argument.recordSize.measurement(DataUnit.Byte).intValue()),
+        argument.throughput);
+  }
+
+  static List<ProducerExecutor> producerExecutors(
+      Performance.Argument argument,
+      List<? extends BiConsumer<Long, Long>> observers,
+      DataSupplier dataSupplier,
+      Supplier<Integer> partitionSupplier) {
+    return IntStream.range(0, argument.producers)
+        .mapToObj(
+            index ->
+                argument.isolation() == Isolation.READ_COMMITTED
+                    ? ProducerExecutor.of(
+                        argument.topic,
+                        argument.transactionSize,
+                        Producer.builder()
+                            .configs(argument.allConfigs())
+                            .brokers(argument.brokers)
+                            .compression(argument.compression)
+                            .partitionClassName(argument.partitioner)
+                            .buildTransactional(),
+                        observers.get(index),
+                        partitionSupplier,
+                        dataSupplier)
+                    : ProducerExecutor.of(
+                        argument.topic,
+                        Producer.builder()
+                            .configs(argument.allConfigs())
+                            .brokers(argument.brokers)
+                            .compression(argument.compression)
+                            .partitionClassName(argument.partitioner)
+                            .build(),
+                        observers.get(index),
+                        partitionSupplier,
+                        dataSupplier))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
   public static Result execute(final Argument param)
       throws InterruptedException, IOException, ExecutionException {
     List<Integer> partitions;
@@ -96,14 +142,28 @@ public class Performance {
             .collect(Collectors.toUnmodifiableList());
 
     var manager = new Manager(param, producerMetrics, consumerMetrics);
-    var tracker = new Tracker(producerMetrics, consumerMetrics, manager);
+    var groupId = "groupId-" + System.currentTimeMillis();
+    var consumerBalancerLatch = new CountDownLatch(param.consumers);
+    var dataSupplier = dataSupplier(param);
+    Supplier<Integer> partitionSupplier =
+        () -> partitions.isEmpty() ? -1 : partitions.get((int) (Math.random() * partitions.size()));
+
+    var producerExecutors =
+        producerExecutors(param, producerMetrics, dataSupplier, partitionSupplier);
+
+    Supplier<Boolean> producerDone =
+        () -> producerExecutors.stream().allMatch(ProducerExecutor::closed);
+
+    var tracker = new Tracker(producerMetrics, consumerMetrics, manager, producerDone);
+
     Collection<Executor> fileWriter =
         (param.CSVPath != null)
             ? List.of(
-                ReportFormat.createFileWriter(param.reportFormat, param.CSVPath, manager, tracker))
+                ReportFormat.createFileWriter(
+                    param.reportFormat, param.CSVPath, manager, producerDone, tracker))
             : List.of();
-    var groupId = "groupId-" + System.currentTimeMillis();
-    try (var threadPool =
+
+    try (var consumersPool =
         ThreadPool.builder()
             .executors(
                 IntStream.range(0, param.consumers)
@@ -114,44 +174,37 @@ public class Performance {
                                     .brokers(param.brokers)
                                     .topics(Set.of(param.topic))
                                     .groupId(groupId)
-                                    .configs(param.configs)
+                                    .configs(param.allConfigs())
                                     .isolation(param.isolation())
                                     .consumerRebalanceListener(
-                                        ignore -> manager.countDownGetAssignment())
+                                        ignore -> consumerBalancerLatch.countDown())
                                     .build(),
                                 consumerMetrics.get(i),
-                                manager))
-                    .collect(Collectors.toUnmodifiableList()))
-            .executors(
-                IntStream.range(0, param.producers)
-                    .mapToObj(
-                        i ->
-                            producerExecutor(
-                                Producer.builder()
-                                    .configs(param.configs)
-                                    .compression(param.compression)
-                                    .partitionClassName(param.partitioner)
-                                    .build(),
-                                Producer.builder()
-                                    .configs(param.configs)
-                                    .compression(param.compression)
-                                    .partitionClassName(param.partitioner)
-                                    .buildTransactional(),
-                                param,
-                                producerMetrics.get(i),
-                                partitions,
-                                manager))
+                                manager,
+                                producerDone))
                     .collect(Collectors.toUnmodifiableList()))
-            .executor(tracker)
-            .executors(fileWriter)
             .build()) {
-      threadPool.waitAll();
-      return new Result(param.topic);
+      // make sure all consumers get their partition assignment
+      consumerBalancerLatch.await();
+
+      try (var threadPool =
+          ThreadPool.builder()
+              .executors(producerExecutors)
+              .executor(tracker)
+              .executors(fileWriter)
+              .build()) {
+        threadPool.waitAll();
+        consumersPool.waitAll();
+        return new Result(param.topic);
+      }
     }
   }
 
   static Executor consumerExecutor(
-      Consumer<byte[], byte[]> consumer, BiConsumer<Long, Long> observer, Manager manager) {
+      Consumer<byte[], byte[]> consumer,
+      BiConsumer<Long, Long> observer,
+      Manager manager,
+      Supplier<Boolean> producerDone) {
     return new Executor() {
       @Override
       public State execute() {
@@ -167,7 +220,7 @@ public class Performance {
                         (long) record.serializedKeySize() + record.serializedValueSize());
                   });
           // Consumer reached the record upperbound or consumed all the record producer produced.
-          return manager.consumedDone() ? State.DONE : State.RUNNING;
+          return producerDone.get() && manager.consumedDone() ? State.DONE : State.RUNNING;
         } catch (WakeupException ignore) {
           // Stop polling and being ready to clean up
           return State.DONE;
@@ -186,92 +239,6 @@ public class Performance {
     };
   }
 
-  static Executor producerExecutor(
-      Producer<byte[], byte[]> producer,
-      TransactionalProducer<byte[], byte[]> transactionalProducer,
-      Argument param,
-      BiConsumer<Long, Long> observer,
-      List<Integer> partitions,
-      Manager manager) {
-    return new Executor() {
-
-      private State doTransaction() {
-        var rand = new Random();
-        var senders =
-            IntStream.range(0, param.transactionSize)
-                .mapToObj(i -> manager.payload())
-                .filter(Optional::isPresent)
-                .map(
-                    p ->
-                        producer
-                            .sender()
-                            .topic(param.topic)
-                            .partition(partitions.get(rand.nextInt(partitions.size())))
-                            .key(manager.getKey())
-                            .value(p.get())
-                            .timestamp(System.currentTimeMillis()))
-                .collect(Collectors.toList());
-
-        // No records to send
-        if (senders.isEmpty()) return State.DONE;
-        transactionalProducer
-            .transaction(senders)
-            .forEach(
-                future ->
-                    future.whenComplete(
-                        (m, e) ->
-                            observer.accept(
-                                System.currentTimeMillis() - m.timestamp(),
-                                m.serializedValueSize())));
-        return State.RUNNING;
-      }
-
-      private State doSend() {
-        var rand = new Random();
-        var payload = manager.payload();
-        if (payload.isEmpty()) return State.DONE;
-
-        long start = System.currentTimeMillis();
-        producer
-            .sender()
-            .topic(param.topic)
-            .partition(partitions.get(rand.nextInt(partitions.size())))
-            .key(manager.getKey())
-            .value(payload.get())
-            .timestamp(start)
-            .run()
-            .whenComplete(
-                (m, e) ->
-                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
-        return State.RUNNING;
-      }
-
-      @Override
-      public State execute() throws InterruptedException {
-        // Wait for all consumers get assignment.
-        manager.awaitPartitionAssignment();
-        // Do transactional send.
-        switch (param.isolation()) {
-          case READ_COMMITTED:
-            return doTransaction();
-          case READ_UNCOMMITTED:
-            return doSend();
-          default:
-            return State.RUNNING;
-        }
-      }
-
-      @Override
-      public void close() {
-        try {
-          producer.close();
-        } finally {
-          manager.producerClosed();
-        }
-      }
-    };
-  }
-
   // visible for test
   static List<Integer> partition(Argument param, TopicAdmin topicAdmin) {
     if (positiveSpecifyBroker(param)) {
@@ -289,6 +256,13 @@ public class Performance {
 
   public static class Argument extends org.astraea.argument.Argument {
 
+    Map<String, String> allConfigs() {
+      var all = new HashMap<>(configs);
+      props().forEach((k, v) -> all.put(k, v.toString()));
+      all.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.nameOfKafka());
+      return all;
+    }
+
     @Parameter(
         names = {"--topic"},
         description = "String: topic name",
@@ -361,7 +335,8 @@ public class Performance {
 
     @Parameter(
         names = {"--transaction.size"},
-        description = "integer: number of records in each transaction",
+        description =
+            "integer: number of records in each transaction. the value larger than 1 means the producer works for transaction",
         validateWith = PositiveLongField.class)
     int transactionSize = 1;
 
diff --git a/app/src/main/java/org/astraea/performance/ProducerExecutor.java b/app/src/main/java/org/astraea/performance/ProducerExecutor.java
new file mode 100644
index 00000000..b584b9f5
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/ProducerExecutor.java
@@ -0,0 +1,179 @@
+package org.astraea.performance;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.Utils;
+import org.astraea.concurrent.Executor;
+import org.astraea.concurrent.State;
+import org.astraea.producer.Producer;
+import org.astraea.producer.Sender;
+import org.astraea.producer.TransactionalProducer;
+
+abstract class ProducerExecutor implements Executor {
+
+  static ProducerExecutor of(
+      String topic,
+      int transactionSize,
+      TransactionalProducer<byte[], byte[]> producer,
+      BiConsumer<Long, Long> observer,
+      Supplier<Integer> partitionSupplier,
+      DataSupplier dataSupplier) {
+    return new ProducerExecutor(topic, producer, partitionSupplier, observer, dataSupplier) {
+
+      @Override
+      public State execute() {
+        var data =
+            IntStream.range(0, transactionSize)
+                .mapToObj(i -> dataSupplier.get())
+                .collect(Collectors.toUnmodifiableList());
+
+        // no more data
+        if (data.stream().allMatch(DataSupplier.Data::done)) return State.DONE;
+
+        // no data due to throttle
+        // TODO: we should return a precise sleep time
+        if (data.stream().allMatch(DataSupplier.Data::throttled)) {
+          Utils.sleep(Duration.ofSeconds(1));
+          return State.RUNNING;
+        }
+        return doSend(
+            senders(
+                data.stream()
+                    .filter(DataSupplier.Data::hasData)
+                    .collect(Collectors.toUnmodifiableList())));
+      }
+
+      List<Sender<byte[], byte[]>> senders(List<DataSupplier.Data> data) {
+        return data.stream()
+            .map(
+                d ->
+                    producer
+                        .sender()
+                        .topic(topic)
+                        .partition(partitionSupplier.get())
+                        .key(d.key())
+                        .value(d.value())
+                        .timestamp(System.currentTimeMillis()))
+            .collect(Collectors.toList());
+      }
+
+      State doSend(List<Sender<byte[], byte[]>> senders) {
+        producer
+            .transaction(senders)
+            .forEach(
+                future ->
+                    future.whenComplete(
+                        (m, e) ->
+                            observer.accept(
+                                System.currentTimeMillis() - m.timestamp(),
+                                m.serializedValueSize())));
+        return State.RUNNING;
+      }
+    };
+  }
+
+  static ProducerExecutor of(
+      String topic,
+      Producer<byte[], byte[]> producer,
+      BiConsumer<Long, Long> observer,
+      Supplier<Integer> partitionSupplier,
+      DataSupplier dataSupplier) {
+    return new ProducerExecutor(topic, producer, partitionSupplier, observer, dataSupplier) {
+
+      @Override
+      public State execute() {
+        var data = dataSupplier.get();
+        if (data.done()) return State.DONE;
+
+        // no data due to throttle
+        // TODO: we should return a precise sleep time
+        if (data.throttled()) {
+          Utils.sleep(Duration.ofSeconds(1));
+          return State.RUNNING;
+        }
+        return doSend(data.key(), data.value());
+      }
+
+      Sender<byte[], byte[]> sender(byte[] key, byte[] value) {
+        return producer
+            .sender()
+            .topic(topic)
+            .partition(partitionSupplier.get())
+            .key(key)
+            .value(value)
+            .timestamp(System.currentTimeMillis());
+      }
+
+      State doSend(byte[] key, byte[] value) {
+        sender(key, value)
+            .run()
+            .whenComplete(
+                (m, e) ->
+                    observer.accept(
+                        System.currentTimeMillis() - m.timestamp(), m.serializedValueSize()));
+        return State.RUNNING;
+      }
+    };
+  }
+
+  private final String topic;
+  private final Producer<byte[], byte[]> producer;
+  private final Supplier<Integer> partitionSupplier;
+  private final BiConsumer<Long, Long> observer;
+  private final DataSupplier dataSupplier;
+
+  ProducerExecutor(
+      String topic,
+      Producer<byte[], byte[]> producer,
+      Supplier<Integer> partitionSupplier,
+      BiConsumer<Long, Long> observer,
+      DataSupplier dataSupplier) {
+    this.topic = topic;
+    this.producer = producer;
+    this.partitionSupplier = partitionSupplier;
+    this.observer = observer;
+    this.dataSupplier = dataSupplier;
+  }
+
+  private final AtomicBoolean closed = new AtomicBoolean(false);
+
+  String topic() {
+    return topic;
+  }
+
+  Supplier<Integer> partitionSupplier() {
+    return partitionSupplier;
+  }
+
+  BiConsumer<Long, Long> observer() {
+    return observer;
+  }
+
+  DataSupplier dataSupplier() {
+    return dataSupplier;
+  }
+
+  /** @return true if the producer in this executor is transactional. */
+  boolean transactional() {
+    return producer instanceof TransactionalProducer;
+  }
+
+  /** @return true if this executor is closed. otherwise, false */
+  public boolean closed() {
+    return closed.get();
+  }
+
+  @Override
+  public void close() {
+    try {
+      producer.close();
+    } finally {
+      closed.set(true);
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/ReportFormat.java b/app/src/main/java/org/astraea/performance/ReportFormat.java
index f03782b1..98c452b4 100644
--- a/app/src/main/java/org/astraea/performance/ReportFormat.java
+++ b/app/src/main/java/org/astraea/performance/ReportFormat.java
@@ -10,6 +10,7 @@ import java.nio.file.Path;
 import java.text.SimpleDateFormat;
 import java.time.Duration;
 import java.util.Date;
+import java.util.function.Supplier;
 import java.util.stream.IntStream;
 import org.astraea.Utils;
 import org.astraea.concurrent.Executor;
@@ -41,7 +42,12 @@ public enum ReportFormat {
   }
 
   public static Executor createFileWriter(
-      ReportFormat reportFormat, Path path, Manager manager, Tracker tracker) throws IOException {
+      ReportFormat reportFormat,
+      Path path,
+      Manager manager,
+      Supplier<Boolean> producerDone,
+      Tracker tracker)
+      throws IOException {
     var filePath =
         FileSystems.getDefault()
             .getPath(
@@ -57,7 +63,7 @@ public enum ReportFormat {
         return new Executor() {
           @Override
           public State execute() throws InterruptedException {
-            if (logToCSV(writer, manager, tracker)) return State.DONE;
+            if (logToCSV(writer, manager, producerDone, tracker)) return State.DONE;
             Thread.sleep(1000);
             return State.RUNNING;
           }
@@ -72,7 +78,7 @@ public enum ReportFormat {
         return new Executor() {
           @Override
           public State execute() throws InterruptedException {
-            if (logToJSON(writer, manager, tracker)) return State.DONE;
+            if (logToJSON(writer, manager, producerDone, tracker)) return State.DONE;
             Thread.sleep(1000);
             return State.RUNNING;
           }
@@ -126,7 +132,8 @@ public enum ReportFormat {
     writer.newLine();
   }
 
-  private static boolean logToCSV(BufferedWriter writer, Manager manager, Tracker tracker) {
+  private static boolean logToCSV(
+      BufferedWriter writer, Manager manager, Supplier<Boolean> producerDone, Tracker tracker) {
     var result = processResult(manager, tracker);
     if (result.producerResult.completedRecords == 0) return false;
     try {
@@ -154,11 +161,12 @@ public enum ReportFormat {
       writer.newLine();
     } catch (IOException ignore) {
     }
-    return manager.producedDone() && manager.consumedDone();
+    return producerDone.get() && manager.consumedDone();
   }
 
   /** Write to writer. Output: "(timestamp)": { (many metrics ...) } */
-  private static boolean logToJSON(BufferedWriter writer, Manager manager, Tracker tracker) {
+  private static boolean logToJSON(
+      BufferedWriter writer, Manager manager, Supplier<Boolean> producerDone, Tracker tracker) {
     var result = processResult(manager, tracker);
     if (result.producerResult.completedRecords == 0) return false;
     try {
@@ -200,7 +208,7 @@ public enum ReportFormat {
       writer.newLine();
     } catch (IOException ignore) {
     }
-    return manager.producedDone() && manager.consumedDone();
+    return producerDone.get() && manager.consumedDone();
   }
 
   private static ProcessedResult processResult(Manager manager, Tracker tracker) {
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index b197dbf8..229df870 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -4,6 +4,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Supplier;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.utils.DataUnit;
@@ -13,17 +14,23 @@ public class Tracker implements Executor {
   private final List<Metrics> producerData;
   private final List<Metrics> consumerData;
   private final Manager manager;
+  private final Supplier<Boolean> producerDone;
   // Snapshot of metrics
   private Result producerResult;
   private Result consumerResult;
   long start = 0L;
 
-  public Tracker(List<Metrics> producerData, List<Metrics> consumerData, Manager manager) {
+  public Tracker(
+      List<Metrics> producerData,
+      List<Metrics> consumerData,
+      Manager manager,
+      Supplier<Boolean> producerDone) {
     this.producerData = producerData;
     this.consumerData = consumerData;
     this.producerResult = result(producerData);
     this.consumerResult = result(consumerData);
     this.manager = manager;
+    this.producerDone = producerDone;
   }
 
   @Override
@@ -76,7 +83,7 @@ public class Tracker implements Executor {
           "  producer[%d] average publish latency: %.3f ms%n", i, result.averageLatencies.get(i));
     }
     System.out.println("\n");
-    return manager.producedDone();
+    return producerDone.get();
   }
 
   private boolean logConsumers(Result result) {
@@ -102,7 +109,7 @@ public class Tracker implements Executor {
     }
     System.out.println("\n");
     // Target number of records consumed OR consumed all that produced
-    return manager.producedDone() && percentage >= 100D;
+    return producerDone.get() && percentage >= 100D;
   }
 
   public Result producerResult() {
diff --git a/app/src/main/java/org/astraea/producer/Sender.java b/app/src/main/java/org/astraea/producer/Sender.java
index be027d2c..9c48cd59 100644
--- a/app/src/main/java/org/astraea/producer/Sender.java
+++ b/app/src/main/java/org/astraea/producer/Sender.java
@@ -14,7 +14,7 @@ public interface Sender<Key, Value> {
   /**
    * define the data route if you don't want to partitioner to decide the target.
    *
-   * @param partition target partition
+   * @param partition target partition. negative value is ignored
    * @return this sender
    */
   Sender<Key, Value> partition(int partition);


Test patch:
diff --git a/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java b/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
index ab3a15ad..6548137d 100644
--- a/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
+++ b/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
@@ -63,4 +63,13 @@ public class ThreadPoolTest {
     pool.close();
     Assertions.assertTrue(pool.isClosed());
   }
+
+  @Test
+  void testEmpty() {
+    try (var empty = ThreadPool.builder().build()) {
+      Assertions.assertEquals(ThreadPool.EMPTY, empty);
+      Assertions.assertEquals(0, empty.size());
+      Assertions.assertTrue(empty.isClosed());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/performance/DataSupplierTest.java b/app/src/test/java/org/astraea/performance/DataSupplierTest.java
new file mode 100644
index 00000000..4d96f2b3
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/DataSupplierTest.java
@@ -0,0 +1,100 @@
+package org.astraea.performance;
+
+import java.util.concurrent.TimeUnit;
+import org.astraea.utils.DataUnit;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DataSupplierTest {
+
+  @Test
+  void testDuration() throws InterruptedException {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("2s"),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100));
+    Assertions.assertTrue(dataSupplier.get().hasData());
+    TimeUnit.SECONDS.sleep(3);
+    Assertions.assertFalse(dataSupplier.get().hasData());
+  }
+
+  @Test
+  void testRecordLimit() {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("2records"),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100));
+    Assertions.assertTrue(dataSupplier.get().hasData());
+    Assertions.assertTrue(dataSupplier.get().hasData());
+    Assertions.assertFalse(dataSupplier.get().hasData());
+  }
+
+  @Test
+  void testKeySize() {
+    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);
+  }
+
+  @Test
+  void testFixedValueSize() {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100),
+            DistributionType.FIXED.create(0),
+            DataUnit.KiB.of(100));
+    var data = dataSupplier.get();
+    Assertions.assertTrue(data.hasData());
+    // initial value size is 100KB and the distributed is fixed to zero, so the final size is 102400
+    Assertions.assertEquals(102400, data.value().length);
+  }
+
+  @Test
+  void testDistributedValueSize() {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100),
+            DistributionType.FIXED.create(10),
+            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);
+  }
+
+  @Test
+  void testThrottle() {
+    var dataSupplier =
+        DataSupplier.of(
+            ExeTime.of("10s"),
+            DistributionType.FIXED.create(10),
+            DataUnit.KiB.of(100),
+            DistributionType.FIXED.create(0),
+            DataUnit.KiB.of(150));
+    // total: 100KB, limit: 150KB -> no throttle
+    Assertions.assertTrue(dataSupplier.get().hasData());
+    // total: 200KB, limit: 150KB -> will throttle next data
+    Assertions.assertTrue(dataSupplier.get().hasData());
+    // throttled
+    Assertions.assertFalse(dataSupplier.get().hasData());
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 7cc1eb7e..8ebca7f4 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -1,58 +1,10 @@
 package org.astraea.performance;
 
-import java.util.Arrays;
 import java.util.List;
-import org.astraea.utils.DataUnit;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class ManagerTest {
-  @Test
-  void testReturnNull() throws InterruptedException {
-    var argument = new Performance.Argument();
-    var producerMetric = new Metrics();
-    argument.exeTime = ExeTime.of("1records");
-    var manager = new Manager(argument, List.of(producerMetric), List.of());
-
-    Assertions.assertTrue(manager.payload().isPresent());
-    producerMetric.accept(1L, 1L);
-    Assertions.assertTrue(manager.payload().isEmpty());
-
-    producerMetric = new Metrics();
-    argument.exeTime = ExeTime.of("50ms");
-    manager = new Manager(argument, List.of(producerMetric), List.of());
-
-    Assertions.assertTrue(manager.payload().isPresent());
-    producerMetric.accept(1L, 1L);
-    Thread.sleep(50);
-    Assertions.assertTrue(manager.payload().isEmpty());
-  }
-
-  @Test
-  void testRandomSize() {
-    var argument = new Performance.Argument();
-    argument.exeTime = ExeTime.of("3records");
-    argument.recordSize = DataUnit.KiB.of(100);
-    var dataManager = new Manager(argument, List.of(), List.of());
-    boolean sameSize = dataManager.payload().get().length == dataManager.payload().get().length;
-
-    // Assertion failed with probability 1/102400 ~ 0.001%
-    Assertions.assertFalse(sameSize);
-
-    Assertions.assertTrue(dataManager.payload().get().length <= 102400);
-  }
-
-  @Test
-  void testRandomContent() {
-    var argument = new Performance.Argument();
-    argument.exeTime = ExeTime.of("2records");
-    argument.recordSize = DataUnit.KiB.of(100);
-    var manager = new Manager(argument, List.of(), List.of());
-    boolean same = Arrays.equals(manager.payload().get(), manager.payload().get());
-
-    // Assertion failed with probability < 1/102400
-    Assertions.assertFalse(same);
-  }
 
   @Test
   void testGetKey() {
@@ -78,69 +30,22 @@ public class ManagerTest {
   @Test
   void testConsumerDone() {
     var argument = new Performance.Argument();
-    var producerMetrics = List.of(new Metrics());
-    var consumerMetrics = List.of(new Metrics());
+    var producerMetrics = new Metrics();
+    var consumerMetrics = new Metrics();
 
     argument.exeTime = ExeTime.of("1records");
-    var manager = new Manager(argument, producerMetrics, consumerMetrics);
-    Assertions.assertFalse(manager.consumedDone());
+    var manager = new Manager(argument, List.of(producerMetrics), List.of(consumerMetrics));
 
     // Produce one record
-    producerMetrics.get(0).accept(0L, 0L);
+    producerMetrics.accept(0L, 0L);
     Assertions.assertFalse(manager.consumedDone());
 
     // Consume one record
-    consumerMetrics.get(0).accept(0L, 0L);
-    Assertions.assertFalse(manager.consumedDone());
-
-    manager.producerClosed();
+    consumerMetrics.accept(0L, 0L);
     Assertions.assertTrue(manager.consumedDone());
 
     // Test zero consumer. (run for one record)
-    producerMetrics = List.of(new Metrics());
-    consumerMetrics = List.of();
-    manager = new Manager(argument, producerMetrics, consumerMetrics);
-    Assertions.assertFalse(manager.consumedDone());
-
-    // Produce one record
-    producerMetrics.get(0).accept(0L, 0L);
-    Assertions.assertFalse(manager.consumedDone());
-
-    manager.producerClosed();
+    manager = new Manager(argument, List.of(producerMetrics), List.of());
     Assertions.assertTrue(manager.consumedDone());
   }
-
-  @Test
-  void testProducerDone() {
-    var argument = new Performance.Argument();
-    argument.exeTime = ExeTime.of("1records");
-    var producerMetrics = List.of(new Metrics());
-    var consumerMetrics = List.of(new Metrics());
-    var manager = new Manager(new Performance.Argument(), producerMetrics, consumerMetrics);
-
-    Assertions.assertFalse(manager.producedDone());
-
-    producerMetrics.get(0).accept(0L, 0L);
-    Assertions.assertFalse(manager.producedDone());
-
-    manager.producerClosed();
-    Assertions.assertTrue(manager.producedDone());
-  }
-
-  @Test
-  void testCheckAndAdd() throws InterruptedException {
-    var argument = new Performance.Argument();
-    argument.throughput = DataUnit.KiB.of(3);
-    var recordSize = DataUnit.KiB.of(1).measurement(DataUnit.Byte).intValue();
-    var manager = new Manager(argument, List.of(), List.of());
-
-    Assertions.assertTrue(manager.checkAndAdd(recordSize));
-    Assertions.assertTrue(manager.checkAndAdd(recordSize));
-    Assertions.assertTrue(manager.checkAndAdd(recordSize));
-    Assertions.assertFalse(manager.checkAndAdd(recordSize));
-
-    Thread.sleep(1001);
-
-    Assertions.assertTrue(manager.checkAndAdd(recordSize));
-  }
 }
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 2aee96c2..3d6db039 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -1,16 +1,16 @@
 package org.astraea.performance;
 
-import static org.astraea.performance.Performance.partition;
-
 import com.beust.jcommander.ParameterException;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 import org.apache.kafka.common.TopicPartition;
-import org.astraea.Utils;
 import org.astraea.concurrent.Executor;
-import org.astraea.concurrent.ThreadPool;
+import org.astraea.concurrent.State;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
@@ -22,113 +22,58 @@ import org.junit.jupiter.api.Test;
 public class PerformanceTest extends RequireBrokerCluster {
 
   @Test
-  void testSpecifyBrokerProducerExecutor() {
-    var admin = TopicAdmin.of(bootstrapServers());
-    var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
-    admin.creator().topic(topicName).numberOfPartitions(10).create();
-    Utils.waitFor(() -> admin.publicTopicNames().contains(topicName));
-
-    var metrics = new Metrics();
-    var param = new Performance.Argument();
-    param.brokers = bootstrapServers();
-    param.topic = topicName;
-    param.sizeDistributionType = DistributionType.FIXED;
-    param.exeTime = ExeTime.of("100records");
-    param.specifyBroker = List.of(0);
-    param.consumers = 0;
-    param.partitions = 10;
-    try (var executor =
-        Performance.producerExecutor(
-            Producer.builder().brokers(bootstrapServers()).build(),
-            Producer.builder().brokers(bootstrapServers()).buildTransactional(),
-            param,
-            metrics,
-            partition(param, admin),
-            new Manager(param, List.of(), List.of()))) {
-      ThreadPool threadPool = ThreadPool.builder().executor(executor).build();
-      threadPool.waitAll();
-      threadPool.close();
-
-      Utils.waitFor(() -> metrics.num() == 100);
-      var offsets = admin.offsets(Set.of(topicName));
-      var partitions =
-          offsets.entrySet().stream()
-              .filter(entry -> entry.getValue().latest() > 0)
-              .map(entry -> entry.getKey().partition())
-              .collect(Collectors.toList());
-      var partitionsOfBrokers =
-          admin.partitionsOfBrokers(Set.of(topicName), Set.of(0)).stream()
-              .map(TopicPartition::partition)
-              .collect(Collectors.toSet());
-      partitions.forEach(
-          partition -> Assertions.assertTrue(partitionsOfBrokers.contains(partition)));
-    }
-  }
-
-  @Test
-  void testMultipleSpecifyBrokersProducerExecutor() {
-    var admin = TopicAdmin.of(bootstrapServers());
-    var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
-    admin.creator().topic(topicName).numberOfPartitions(10).create();
-    Utils.waitFor(() -> admin.publicTopicNames().contains(topicName));
-
-    var metrics = new Metrics();
-    var param = new Performance.Argument();
-    param.brokers = bootstrapServers();
-    param.topic = topicName;
-    param.sizeDistributionType = DistributionType.FIXED;
-    param.exeTime = ExeTime.of("100records");
-    param.specifyBroker = List.of(0, 1);
-    param.consumers = 0;
-    param.partitions = 10;
-    try (Executor executor =
-        Performance.producerExecutor(
-            Producer.builder().brokers(bootstrapServers()).build(),
-            Producer.builder().brokers(bootstrapServers()).buildTransactional(),
-            param,
-            metrics,
-            partition(param, admin),
-            new Manager(param, List.of(), List.of()))) {
-      ThreadPool threadPool = ThreadPool.builder().executor(executor).build();
-      threadPool.waitAll();
-      threadPool.close();
-
-      Utils.waitFor(() -> metrics.num() == 100);
-      var offsets = admin.offsets(Set.of(topicName));
-      var partitions =
-          offsets.entrySet().stream()
-              .filter(entry -> entry.getValue().latest() > 0)
-              .map(entry -> entry.getKey().partition())
-              .collect(Collectors.toList());
-      var partitionsOfBrokers =
-          admin.partitionsOfBrokers(Set.of(topicName), Set.of(0, 1)).stream()
-              .map(TopicPartition::partition)
-              .collect(Collectors.toSet());
-      partitions.forEach(
-          partition -> Assertions.assertTrue(partitionsOfBrokers.contains(partition)));
-    }
+  void testTransactionalProducer() {
+    var topic = "testTransactionalProducer";
+    String[] arguments1 = {
+      "--bootstrap.servers", bootstrapServers(), "--topic", topic, "--transaction.size", "2"
+    };
+    var latch = new CountDownLatch(1);
+    BiConsumer<Long, Long> observer = (x, y) -> latch.countDown();
+    var argument = org.astraea.argument.Argument.parse(new Performance.Argument(), arguments1);
+    var producerExecutors =
+        Performance.producerExecutors(
+            argument,
+            List.of(observer),
+            () ->
+                DataSupplier.data(
+                    "key".getBytes(StandardCharsets.UTF_8),
+                    "value".getBytes(StandardCharsets.UTF_8)),
+            () -> -1);
+    Assertions.assertEquals(1, producerExecutors.size());
+    Assertions.assertTrue(producerExecutors.get(0).transactional());
   }
 
   @Test
   void testProducerExecutor() throws InterruptedException {
-    var metrics = new Metrics();
-    var param = new Performance.Argument();
-    param.brokers = bootstrapServers();
-    param.topic = "testProducerExecutor-" + System.currentTimeMillis();
-    param.sizeDistributionType = DistributionType.FIXED;
-    param.consumers = 0;
-    try (Executor executor =
-        Performance.producerExecutor(
-            Producer.builder().brokers(bootstrapServers()).build(),
-            Producer.builder().brokers(bootstrapServers()).buildTransactional(),
-            param,
-            metrics,
-            List.of(-1),
-            new Manager(param, List.of(), List.of()))) {
-      executor.execute();
-
-      Utils.waitFor(() -> metrics.num() == 1);
-      Assertions.assertEquals(1024, metrics.bytes());
+    var topic = "testProducerExecutor";
+    String[] arguments1 = {
+      "--bootstrap.servers", bootstrapServers(), "--topic", topic, "--compression", "gzip"
+    };
+    var latch = new CountDownLatch(1);
+    BiConsumer<Long, Long> observer = (x, y) -> latch.countDown();
+    var argument = org.astraea.argument.Argument.parse(new Performance.Argument(), arguments1);
+    var producerExecutors =
+        Performance.producerExecutors(
+            argument,
+            List.of(observer),
+            () ->
+                DataSupplier.data(
+                    "key".getBytes(StandardCharsets.UTF_8),
+                    "value".getBytes(StandardCharsets.UTF_8)),
+            () -> -1);
+    Assertions.assertEquals(1, producerExecutors.size());
+    Assertions.assertFalse(producerExecutors.get(0).transactional());
+
+    try (var admin = TopicAdmin.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(1).create();
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(2);
+      admin.offsets(Set.of(topic)).values().forEach(o -> Assertions.assertEquals(0, o.latest()));
+
+      Assertions.assertEquals(State.RUNNING, producerExecutors.get(0).execute());
+      latch.await();
+      Assertions.assertEquals(
+          1, admin.offsets(Set.of(topic)).get(new TopicPartition(topic, 0)).latest());
     }
   }
 
@@ -142,7 +87,8 @@ public class PerformanceTest extends RequireBrokerCluster {
         Performance.consumerExecutor(
             Consumer.builder().topics(Set.of(topicName)).brokers(bootstrapServers()).build(),
             metrics,
-            new Manager(param, List.of(), List.of()))) {
+            new Manager(param, List.of(), List.of()),
+            () -> false)) {
       executor.execute();
 
       Assertions.assertEquals(0, metrics.num());
@@ -162,6 +108,8 @@ public class PerformanceTest extends RequireBrokerCluster {
   void testTransactionSet() {
     var argument = new Performance.Argument();
     Assertions.assertEquals(Isolation.READ_UNCOMMITTED, argument.isolation());
+    argument.transactionSize = 1;
+    Assertions.assertEquals(Isolation.READ_UNCOMMITTED, argument.isolation());
     argument.transactionSize = 3;
     Assertions.assertEquals(Isolation.READ_COMMITTED, argument.isolation());
   }
diff --git a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
new file mode 100644
index 00000000..604c5a58
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
@@ -0,0 +1,161 @@
+package org.astraea.performance;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+import org.apache.kafka.common.TopicPartition;
+import org.astraea.Utils;
+import org.astraea.concurrent.State;
+import org.astraea.producer.Producer;
+import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Named;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class ProducerExecutorTest extends RequireBrokerCluster {
+
+  private static ProducerExecutor ofProducer(
+      String topic,
+      Supplier<Integer> partitionSupplier,
+      BiConsumer<Long, Long> observer,
+      DataSupplier dataSupplier) {
+    return ProducerExecutor.of(
+        topic,
+        Producer.builder().brokers(bootstrapServers()).build(),
+        observer,
+        partitionSupplier,
+        dataSupplier);
+  }
+
+  private static ProducerExecutor ofTransactionalProducer(
+      String topic,
+      Supplier<Integer> partitionSupplier,
+      BiConsumer<Long, Long> observer,
+      DataSupplier dataSupplier) {
+    return ProducerExecutor.of(
+        topic,
+        10,
+        Producer.builder().brokers(bootstrapServers()).buildTransactional(),
+        observer,
+        partitionSupplier,
+        dataSupplier);
+  }
+
+  @ParameterizedTest
+  @MethodSource("offsetProducerExecutors")
+  void testSpecifiedPartition(ProducerExecutor executor) throws InterruptedException {
+    var specifiedPartition = 1;
+    ((MyPartitionSupplier) executor.partitionSupplier()).partition = specifiedPartition;
+    try (var admin = TopicAdmin.of(bootstrapServers())) {
+      admin.creator().topic(executor.topic()).numberOfPartitions(specifiedPartition + 1).create();
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(2);
+      admin
+          .offsets(Set.of(executor.topic()))
+          .values()
+          .forEach(o -> Assertions.assertEquals(0, o.latest()));
+      Assertions.assertEquals(State.RUNNING, executor.execute());
+      // wait for syncing data
+      TimeUnit.SECONDS.sleep(2);
+      // only specified partition gets value
+      // for normal producer, there is only one record
+      // for transactional producer, the size of transaction is 10 and there is one transaction
+      // control marker
+      Assertions.assertEquals(
+          executor.transactional() ? 11 : 1,
+          admin
+              .offsets(Set.of(executor.topic()))
+              .get(new TopicPartition(executor.topic(), specifiedPartition))
+              .latest());
+      // other partitions have no data
+      admin.offsets(Set.of(executor.topic())).entrySet().stream()
+          .filter(e -> !e.getKey().equals(new TopicPartition(executor.topic(), specifiedPartition)))
+          .forEach(e -> Assertions.assertEquals(0, e.getValue().latest()));
+    }
+  }
+
+  @ParameterizedTest
+  @MethodSource("offsetProducerExecutors")
+  void testDone(ProducerExecutor executor) throws InterruptedException {
+    ((MyDataSupplier) executor.dataSupplier()).data = DataSupplier.NO_MORE_DATA;
+    Assertions.assertEquals(State.DONE, executor.execute());
+  }
+
+  @ParameterizedTest
+  @MethodSource("offsetProducerExecutors")
+  void testClose(ProducerExecutor executor) {
+    executor.close();
+    Assertions.assertTrue(executor.closed());
+  }
+
+  @ParameterizedTest
+  @MethodSource("offsetProducerExecutors")
+  void testObserver(ProducerExecutor executor) throws InterruptedException {
+    Assertions.assertEquals(State.RUNNING, executor.execute());
+    // wait for async call
+    TimeUnit.SECONDS.sleep(2);
+    Assertions.assertEquals(
+        executor.transactional() ? 10 : 1, ((Observer) executor.observer()).recordsHook.size());
+    Assertions.assertEquals(
+        executor.transactional() ? 10 : 1, ((Observer) executor.observer()).elapsedHook.size());
+  }
+
+  private static class Observer implements BiConsumer<Long, Long> {
+    private final BlockingQueue<Long> recordsHook = new LinkedBlockingDeque<>();
+    private final BlockingQueue<Long> elapsedHook = new LinkedBlockingDeque<>();
+
+    @Override
+    public void accept(Long records, Long elapsed) {
+      Assertions.assertTrue(recordsHook.offer(records));
+      Assertions.assertTrue(elapsedHook.offer(elapsed));
+    }
+  }
+
+  private static class MyDataSupplier implements DataSupplier {
+
+    private Data data =
+        DataSupplier.data(
+            "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8));
+
+    @Override
+    public Data get() {
+      return data;
+    }
+  }
+
+  private static class MyPartitionSupplier implements Supplier<Integer> {
+    private int partition = -1;
+
+    @Override
+    public Integer get() {
+      return partition;
+    }
+  }
+
+  private static Stream<Arguments> offsetProducerExecutors() {
+    var normalTopic = Utils.randomString(10);
+    var transactionalTopic = Utils.randomString(10);
+    return Stream.of(
+        Arguments.of(
+            Named.of(
+                "normal producer for topic: " + normalTopic,
+                ofProducer(
+                    normalTopic, new MyPartitionSupplier(), new Observer(), new MyDataSupplier()))),
+        Arguments.of(
+            Named.of(
+                "transactional producer for topic: " + transactionalTopic,
+                ofTransactionalProducer(
+                    transactionalTopic,
+                    new MyPartitionSupplier(),
+                    new Observer(),
+                    new MyDataSupplier()))));
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/TrackerTest.java b/app/src/test/java/org/astraea/performance/TrackerTest.java
index 4e3e75dd..2ae05a67 100644
--- a/app/src/test/java/org/astraea/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/performance/TrackerTest.java
@@ -1,6 +1,7 @@
 package org.astraea.performance;
 
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.astraea.concurrent.State;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -15,21 +16,24 @@ public class TrackerTest {
     argument.exeTime = ExeTime.of("1records");
 
     var manager = new Manager(argument, producerData, consumerData);
-    try (Tracker tracker = new Tracker(producerData, consumerData, manager)) {
+    var producerDone = new AtomicBoolean(false);
+    try (Tracker tracker = new Tracker(producerData, consumerData, manager, producerDone::get)) {
+      producerDone.set(false);
       Assertions.assertEquals(State.RUNNING, tracker.execute());
       producerData.get(0).accept(1L, 1L);
       consumerData.get(0).accept(1L, 1L);
-      manager.producerClosed();
+      producerDone.set(true);
       Assertions.assertEquals(State.DONE, tracker.execute());
     }
 
     // Zero consumer
     producerData = List.of(new Metrics());
     manager = new Manager(argument, producerData, empty);
-    try (Tracker tracker = new Tracker(producerData, empty, manager)) {
+    try (Tracker tracker = new Tracker(producerData, empty, manager, producerDone::get)) {
+      producerDone.set(false);
       Assertions.assertEquals(State.RUNNING, tracker.execute());
       producerData.get(0).accept(1L, 1L);
-      manager.producerClosed();
+      producerDone.set(true);
       Assertions.assertEquals(State.DONE, tracker.execute());
     }
 
@@ -38,7 +42,8 @@ public class TrackerTest {
     producerData = List.of(new Metrics());
     consumerData = List.of(new Metrics());
     manager = new Manager(argument, producerData, consumerData);
-    try (Tracker tracker = new Tracker(producerData, consumerData, manager)) {
+    try (Tracker tracker = new Tracker(producerData, consumerData, manager, producerDone::get)) {
+      producerDone.set(false);
       tracker.start = System.currentTimeMillis();
       Assertions.assertEquals(State.RUNNING, tracker.execute());
 
@@ -46,8 +51,7 @@ public class TrackerTest {
       producerData.get(0).accept(1L, 1L);
       consumerData.get(0).accept(1L, 1L);
       Thread.sleep(2000);
-      manager.producerClosed();
-
+      producerDone.set(true);
       Assertions.assertEquals(State.DONE, tracker.execute());
     }
   }