Repository: skiptests/astraea
Pull request #136 (2021-12-02 04:35:44)

Enable transactional send
可對producer設定使用"transaction mode"。

可以設定有多少機率會出現"指定數量records"的transaction。
`--transaction --transaction.rate 0.3 --transaction.size 7`

作法：
```java
producer.beginTransaction();
// send 1 record or {transaction.size} records
producer.commitTransaction();
```


chia7712: > --transaction --transaction.rate 0.3 --transaction.size 7

第一個參數`--transaction`是需要的嗎？或許當使用者有指定任意一個transaction相關參數的時候就視為需要transaction?
chia7712: @chinghongfang 麻煩處理一下衝突
chinghongfang: 剛剛發現一個問題，我們紀錄publish latency時會做逐筆紀錄。那使用transactional produce時，也要逐筆紀錄嘛？或者要依transaction完成 做紀錄？
chia7712: > 剛剛發現一個問題，我們紀錄publish latency時會做逐筆紀錄。那使用transactional produce時，也要逐筆紀錄嘛？或者要依transaction完成 做紀錄？

這邊可能要看一下APi能不能得到transaction commit時的時間，或者我們就單純在commit transaction那個動作後去抓時間當作完成時間
chinghongfang: > 這邊可能要看一下APi能不能得到transaction commit時的時間，或者我們就單純在commit transaction那個動作後去抓時間當作完成時間

好的，那我再找找是否可以抓到transaction commit的時間，然後把這個(transaction commit time) 當做record的結束時間。
chinghongfang: 找不到API可以抓到transaction commit的時間。Kafka自己似乎也是直接抓commit完後的時間。

https://github.com/apache/kafka/blob/acd1f9c5631ed2aec2d6ab238e6b81c1a9eb47a2/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L739-L747

---

這邊就單純抓commit結束的時間。
```java
producer
    .transaction(senders)
    .forEach(
        future ->
            future.whenComplete(
                (m, e) ->
                    observer.accept(
                        System.currentTimeMillis() - m.timestamp(),
                        m.serializedValueSize())));
```
因為transactional commit是blocking，讓我們的`transaction(senders)`也變成blocking，所以這邊直接抓`transaction(senders)`結束的時間。
另外，考慮到transaction中每筆資料的start time不一樣，改使用metadata中的timestamp取得produce start time，以取得我們的publish latency。
chia7712: > 或許，我們的interface(TopicAdmin::topics())可以選擇是否抓取internal topic。(TopicAdmin::topics(boolean))

可否開另一隻PR先處理這個？
chia7712: @chinghongfang please fix conflicts
chia7712: @chinghongfang 麻煩解決衝突，並且將已經處理好的conversation點掉

Patch:
diff --git a/.gitignore b/.gitignore
index 3835bffb..810e79ed 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,6 +15,5 @@ zookeeper.dockerfile
 broker.dockerfile
 kafka_tool.dockerfile
 spark.dockerfile
-
 # we don't put binary file to git repo
-gradle-wrapper.jar
\ No newline at end of file
+gradle-wrapper.jar
diff --git a/README.md b/README.md
index 4c26b33c..7371e018 100644
--- a/README.md
+++ b/README.md
@@ -51,6 +51,7 @@ You can define `VERSION` to change the binary version.
 ### Run Kafka Broker
 
 After the zk env is running, you can copy the command (see above example) from zk script output to set up kafka. For example:
+
 ```shell
 ./docker/start_broker.sh zookeeper.connect=192.168.50.178:17228
 ```
@@ -59,10 +60,10 @@ The console will show the broker connection information and JMX address. For exa
 
 ```shell
 =================================================
+broker id: 677
 broker address: 192.168.50.178:12747
 jmx address: 192.168.50.178:10216
 exporter address: 192.168.50.178:10558
-broker id: 677
 =================================================
 ```
 
@@ -70,7 +71,24 @@ broker id: 677
 2. `jmx address` exports the java metrics by JMX
 3. `exporter address` is the address of prometheus exporter.
 
+You can set `CONFLUENT_BROKER` to true, if you want start the confluent version of the kafka cluster. For example:
+
+```shell
+env CONFLUENT_BROKER=true ./docker/start_broker.sh zookeeper.connect=192.168.50.178:17228
+```
+
+The console will show the broker connection information and exporter address. For example:
+
+```shell
+=================================================
+broker id: 1001
+broker address: 192.168.103.39:15230
+exporter address: 192.168.103.39:18928
+=================================================
+```
+
 There are 4 useful ENVs which can change JVM/container configuration.
+
 1. VERSION -> define the kafka version
 2. REVISION -> define the revision of kafka source code. If this is defined, it will run distribution based on the source code
 3. HEAP_OPTS -> define JVM memory options
diff --git a/app/src/main/java/org/astraea/cost/CostFunction.java b/app/src/main/java/org/astraea/cost/CostFunction.java
index 58a57bdc..b5318f00 100644
--- a/app/src/main/java/org/astraea/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/cost/CostFunction.java
@@ -12,7 +12,6 @@ public interface CostFunction {
   /**
    * score all nodes according to passed beans and cluster information.
    *
-   * @param beans beans of each node (key is node id)
    * @param clusterInfo cluster information
    * @return the score of each node. the score range is [0 - 1]
    */
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 08d3ed16..02e89d22 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -6,12 +6,14 @@ import java.time.Duration;
 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.ExecutionException;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.TopicPartition;
@@ -29,6 +31,7 @@ import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.producer.Producer;
+import org.astraea.producer.TransactionalProducer;
 import org.astraea.topic.TopicAdmin;
 import org.astraea.utils.DataSize;
 import org.astraea.utils.DataUnit;
@@ -104,7 +107,7 @@ public class Performance {
                                     .brokers(param.brokers)
                                     .topics(Set.of(param.topic))
                                     .groupId(groupId)
-                                    .configs(param.props())
+                                    .configs(param.consumerProps())
                                     .consumerRebalanceListener(
                                         ignore -> manager.countDownGetAssignment())
                                     .build(),
@@ -120,6 +123,10 @@ public class Performance {
                                     .configs(param.producerProps())
                                     .partitionClassName(param.partitioner)
                                     .build(),
+                                Producer.builder()
+                                    .configs(param.producerProps())
+                                    .partitionClassName(param.partitioner)
+                                    .buildTransactional(),
                                 param,
                                 producerMetrics.get(i),
                                 partitions,
@@ -171,6 +178,7 @@ public class Performance {
 
   static Executor producerExecutor(
       Producer<byte[], byte[]> producer,
+      TransactionalProducer<byte[], byte[]> transactionalProducer,
       Argument param,
       BiConsumer<Long, Long> observer,
       List<Integer> partitions,
@@ -181,23 +189,51 @@ public class Performance {
         // Wait for all consumers get assignment.
         manager.awaitPartitionAssignment();
         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) -> {
-                  if (e == null)
-                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize());
-                });
+        // Do transactional send.
+        if (param.transaction()) {
+          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())));
+        } else {
+          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;
       }
 
@@ -279,13 +315,6 @@ public class Performance {
         converter = DataSize.Field.class)
     DataSize recordSize = DataUnit.KiB.of(1);
 
-    @Parameter(
-        names = {"--jmx.servers"},
-        description =
-            "String: server to get jmx metrics <broker_id>.<jmx_serverPort>[,<broker_id>.<jmx_serverPort>]*",
-        validateWith = NonEmptyStringField.class)
-    String jmxServers = "";
-
     @Parameter(
         names = {"--partitioner"},
         description = "String: the full class name of the desired partitioner",
@@ -302,11 +331,16 @@ public class Performance {
     public Map<String, Object> producerProps() {
       var props = props();
       props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.name);
-      if (!this.jmxServers.isEmpty()) props.put("jmx_servers", this.jmxServers);
       props.putAll(configs);
       return props;
     }
 
+    public Map<String, Object> consumerProps() {
+      var props = props();
+      if (transaction()) props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+      return props;
+    }
+
     @Parameter(
         names = {"--createCSV"},
         description = "put the metrics into a csv file if this flag is set")
@@ -319,6 +353,16 @@ public class Performance {
         converter = CompressionField.class)
     CompressionType compression = CompressionType.NONE;
 
+    @Parameter(
+        names = {"--transaction.size"},
+        description = "integer: number of records in each transaction",
+        validateWith = PositiveLongField.class)
+    int transactionSize = 1;
+
+    public boolean transaction() {
+      return transactionSize > 1;
+    }
+
     @Parameter(
         names = {"--key.distribution"},
         description =
diff --git a/app/src/main/java/org/astraea/producer/AbstractSender.java b/app/src/main/java/org/astraea/producer/AbstractSender.java
new file mode 100644
index 00000000..7ba1bc21
--- /dev/null
+++ b/app/src/main/java/org/astraea/producer/AbstractSender.java
@@ -0,0 +1,51 @@
+package org.astraea.producer;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import org.astraea.consumer.Header;
+
+abstract class AbstractSender<Key, Value> implements Sender<Key, Value> {
+  protected Key key;
+  protected Value value;
+  protected String topic;
+  protected Integer partition;
+  protected Long timestamp;
+  protected Collection<Header> headers = List.of();
+
+  @Override
+  public Sender<Key, Value> key(Key key) {
+    this.key = key;
+    return this;
+  }
+
+  @Override
+  public Sender<Key, Value> value(Value value) {
+    this.value = value;
+    return this;
+  }
+
+  @Override
+  public Sender<Key, Value> topic(String topic) {
+    this.topic = Objects.requireNonNull(topic);
+    return this;
+  }
+
+  @Override
+  public Sender<Key, Value> partition(int partition) {
+    if (partition >= 0) this.partition = partition;
+    return this;
+  }
+
+  @Override
+  public Sender<Key, Value> timestamp(long timestamp) {
+    this.timestamp = timestamp;
+    return this;
+  }
+
+  @Override
+  public Sender<Key, Value> headers(Collection<Header> headers) {
+    this.headers = headers;
+    return this;
+  }
+}
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index 44b9184c..944f41b1 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -1,15 +1,21 @@
 package org.astraea.producer;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.OutOfOrderSequenceException;
+import org.apache.kafka.common.errors.ProducerFencedException;
 import org.astraea.consumer.Header;
 
 public class Builder<Key, Value> {
@@ -58,54 +64,65 @@ public class Builder<Key, Value> {
 
       @Override
       public Sender<Key, Value> sender() {
-        return new Sender<>() {
-          private Key key;
-          private Value value;
-          private String topic;
-          private Integer partition;
-          private Long timestamp;
-          private Collection<Header> headers = List.of();
-
+        return new AbstractSender<>() {
           @Override
-          public Sender<Key, Value> key(Key key) {
-            this.key = key;
-            return this;
+          public CompletionStage<Metadata> run() {
+            var completableFuture = new CompletableFuture<Metadata>();
+            kafkaProducer.send(
+                new ProducerRecord<>(topic, partition, timestamp, key, value, Header.of(headers)),
+                (metadata, exception) -> {
+                  if (exception == null) completableFuture.complete(Metadata.of(metadata));
+                  else completableFuture.completeExceptionally(exception);
+                });
+            return completableFuture;
           }
+        };
+      }
 
-          @Override
-          public Sender<Key, Value> value(Value value) {
-            this.value = value;
-            return this;
-          }
+      @Override
+      public void flush() {
+        kafkaProducer.flush();
+      }
 
-          @Override
-          public Sender<Key, Value> topic(String topic) {
-            this.topic = Objects.requireNonNull(topic);
-            return this;
-          }
+      @Override
+      public KafkaProducer<Key, Value> kafkaProducer() {
+        return kafkaProducer;
+      }
 
-          @Override
-          public Sender<Key, Value> partition(int partition) {
-            if (partition >= 0) this.partition = partition;
-            return this;
-          }
+      @Override
+      public void close() {
+        kafkaProducer.close();
+      }
+    };
+  }
 
-          @Override
-          public Sender<Key, Value> timestamp(long timestamp) {
-            this.timestamp = timestamp;
-            return this;
-          }
+  @SuppressWarnings("unchecked")
+  public TransactionalProducer<Key, Value> buildTransactional() {
+    var transactionConfigs = new HashMap<>(configs);
+    transactionConfigs.putIfAbsent(
+        ProducerConfig.TRANSACTIONAL_ID_CONFIG, "id" + new Random().nextLong());
+    // For transactional send
+    var transactionProducer =
+        new KafkaProducer<>(
+            transactionConfigs,
+            Serializer.of((Serializer<Key>) keySerializer),
+            Serializer.of((Serializer<Value>) valueSerializer));
+    transactionProducer.initTransactions();
+    return new TransactionalProducer<>() {
 
+      @Override
+      public Sender<Key, Value> sender() {
+        return new TransactionalSender<>() {
+          /** Send one transactional record. */
           @Override
-          public Sender<Key, Value> headers(Collection<Header> headers) {
-            this.headers = headers;
-            return this;
+          public CompletionStage<Metadata> run() {
+            return transaction(List.of(this)).stream().findFirst().orElseThrow();
           }
 
           @Override
-          public CompletionStage<Metadata> run() {
+          CompletionStage<Metadata> send() {
             var completableFuture = new CompletableFuture<Metadata>();
-            kafkaProducer.send(
+            transactionProducer.send(
                 new ProducerRecord<>(topic, partition, timestamp, key, value, Header.of(headers)),
                 (metadata, exception) -> {
                   if (exception == null) completableFuture.complete(Metadata.of(metadata));
@@ -117,18 +134,51 @@ public class Builder<Key, Value> {
       }
 
       @Override
-      public void flush() {
-        kafkaProducer.flush();
+      public KafkaProducer<Key, Value> kafkaProducer() {
+        return transactionProducer;
       }
 
       @Override
-      public KafkaProducer<Key, Value> kafkaProducer() {
-        return kafkaProducer;
+      public void flush() {
+        transactionProducer.flush();
       }
 
       @Override
       public void close() {
-        kafkaProducer.close();
+        transactionProducer.close();
+      }
+
+      /**
+       * Send a collection of records as a transaction operation. For example,
+       *
+       * <pre>{@code
+       * try(var producer = Producer.builder().brokers("localhost:9092").buildTransactional()){
+       *     producer.transaction(
+       *             IntStream.range(0, 10)
+       *                     .mapToObj(i -> producer.sender().topic("topic1").value(new byte[10]))
+       *                     .collect(Collectors.toList()));
+       * }
+       * }</pre>
+       */
+      @Override
+      public Collection<CompletionStage<Metadata>> transaction(
+          Collection<Sender<Key, Value>> senders) {
+        var futures = new ArrayList<CompletionStage<Metadata>>(senders.size());
+        try {
+          synchronized (transactionProducer) {
+            transactionProducer.beginTransaction();
+            senders.forEach(
+                sender -> futures.add(((TransactionalSender<Key, Value>) sender).send()));
+            transactionProducer.commitTransaction();
+          }
+        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
+          transactionProducer.close();
+          // Error occur
+          throw e;
+        } catch (KafkaException ke) {
+          transactionProducer.abortTransaction();
+        }
+        return futures;
       }
     };
   }
diff --git a/app/src/main/java/org/astraea/producer/Producer.java b/app/src/main/java/org/astraea/producer/Producer.java
index a4051338..2d49f8c5 100644
--- a/app/src/main/java/org/astraea/producer/Producer.java
+++ b/app/src/main/java/org/astraea/producer/Producer.java
@@ -6,7 +6,6 @@ import org.apache.kafka.clients.producer.KafkaProducer;
 /** An interface for sending records. */
 public interface Producer<Key, Value> extends AutoCloseable {
   Sender<Key, Value> sender();
-
   /** this method is blocked until all data in buffer are sent. */
   void flush();
 
diff --git a/app/src/main/java/org/astraea/producer/TransactionalProducer.java b/app/src/main/java/org/astraea/producer/TransactionalProducer.java
new file mode 100644
index 00000000..f2ab8846
--- /dev/null
+++ b/app/src/main/java/org/astraea/producer/TransactionalProducer.java
@@ -0,0 +1,17 @@
+package org.astraea.producer;
+
+import java.util.Collection;
+import java.util.concurrent.CompletionStage;
+
+/**
+ * Difference between this and {@link org.astraea.producer.Producer} is that it is used for sending
+ * records as one transaction. TransactionalProducer can be generated from {@link
+ * org.astraea.producer.Builder} For example,
+ *
+ * <pre>{@code
+ * var producer = Producer.builder().brokers("localhost:9092").buildTransactional();
+ * }</pre>
+ */
+public interface TransactionalProducer<Key, Value> extends Producer<Key, Value> {
+  Collection<CompletionStage<Metadata>> transaction(Collection<Sender<Key, Value>> senders);
+}
diff --git a/app/src/main/java/org/astraea/producer/TransactionalSender.java b/app/src/main/java/org/astraea/producer/TransactionalSender.java
new file mode 100644
index 00000000..77ae34b8
--- /dev/null
+++ b/app/src/main/java/org/astraea/producer/TransactionalSender.java
@@ -0,0 +1,7 @@
+package org.astraea.producer;
+
+import java.util.concurrent.CompletionStage;
+
+public abstract class TransactionalSender<Key, Value> extends AbstractSender<Key, Value> {
+  abstract CompletionStage<Metadata> send();
+}
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index 187cd986..b853b9d3 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -5,9 +5,9 @@ source $DOCKER_FOLDER/docker_build_common.sh
 
 # ===============================[global variables]===============================
 declare -r VERSION=${REVISION:-${VERSION:-3.1.0}}
-declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/broker}
-declare -r IMAGE_NAME="$REPO:$VERSION"
 declare -r DOCKERFILE=$DOCKER_FOLDER/broker.dockerfile
+declare -r CONFLUENT_BROKER=${CONFLUENT_BROKER:-false}
+declare -r CONFLUENT_VERSION=${CONFLUENT_VERSION:-7.0.1}
 declare -r DATA_FOLDER_IN_CONTAINER_PREFIX="/tmp/log-folder"
 declare -r EXPORTER_VERSION="0.16.1"
 declare -r EXPORTER_PORT=${EXPORTER_PORT:-"$(getRandomPort)"}
@@ -26,13 +26,21 @@ declare -r JMX_OPTS="-Dcom.sun.management.jmxremote \
   -Dcom.sun.management.jmxremote.port=$BROKER_JMX_PORT \
   -Dcom.sun.management.jmxremote.rmi.port=$BROKER_JMX_PORT \
   -Djava.rmi.server.hostname=$ADDRESS"
+declare -r ZOOKEEPER_CONNECT=$(echo $1 | sed -n 's/^zookeeper.connect=\(.\+\)$/\1/p')
 declare -r HEAP_OPTS="${HEAP_OPTS:-"-Xmx2G -Xms2G"}"
 declare -r BROKER_PROPERTIES="/tmp/server-${BROKER_PORT}.properties"
-
+if [[ "$CONFLUENT_BROKER" = "true" ]]; then
+    declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/confluent.broker}
+    declare -r IMAGE_NAME="$REPO:$CONFLUENT_VERSION"
+    declare -r SCRIPT_LOCATION_IN_CONTAINER="./bin/kafka-server-start"
+else
+    declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/broker}
+    declare -r IMAGE_NAME="$REPO:$VERSION"
+    declare -r SCRIPT_LOCATION_IN_CONTAINER="./bin/kafka-server-start.sh"
+fi
 # cleanup the file if it is existent
 [[ -f "$BROKER_PROPERTIES" ]] && rm -f "$BROKER_PROPERTIES"
 
-
 # ===================================[functions]===================================
 
 function showHelp() {
@@ -49,7 +57,7 @@ function showHelp() {
   echo "    VERSION=3.1.0                            set version of kafka distribution"
   echo "    BUILD=false                              set true if you want to build image locally"
   echo "    RUN=false                                set false if you want to build/pull image only"
-  echo "    DATA_FOLDERS=/tmp/folder1,/tmp/folder2   set host folders used by broker"
+  echo "    DATA_FOLDERS=/tmp/folder1                set host folders used by broker"
 }
 
 function rejectProperty() {
@@ -68,6 +76,28 @@ function requireProperty() {
   fi
 }
 
+function generateConfluentDockerfile() {
+  echo "# this dockerfile is generated dynamically
+FROM confluentinc/cp-server:$CONFLUENT_VERSION
+USER root
+RUN usermod -l $USER appuser && groupadd $USER && usermod -a -G $USER $USER
+RUN yum -y update && yum -y install git unzip
+
+# download jmx exporter
+RUN mkdir /opt/jmx_exporter
+WORKDIR /opt/jmx_exporter
+RUN wget https://raw.githubusercontent.com/prometheus/jmx_exporter/master/example_configs/kafka-2_0_0.yml --output-document=$JMX_CONFIG_FILE_IN_CONTAINER_PATH
+RUN wget https://REPO1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/${EXPORTER_VERSION}/jmx_prometheus_javaagent-${EXPORTER_VERSION}.jar
+
+# change user
+RUN chown -R $USER:$USER /tmp
+RUN chown -R $USER:$USER /var
+USER $USER
+
+WORKDIR /
+" >"$DOCKERFILE"
+}
+
 function generateDockerfileBySource() {
   echo "# this dockerfile is generated dynamically
 FROM ubuntu:20.04 AS build
@@ -157,10 +187,14 @@ WORKDIR /opt/kafka
 }
 
 function generateDockerfile() {
-  if [[ -n "$REVISION" ]]; then
-    generateDockerfileBySource
+  if [[ "$CONFLUENT_BROKER" = "true" ]]; then
+    generateConfluentDockerfile
   else
-    generateDockerfileByVersion
+    if [[ -n "$REVISION" ]]; then
+      generateDockerfileBySource
+    else
+      generateDockerfileByVersion
+    fi
   fi
 }
 
@@ -182,6 +216,9 @@ function setListener() {
   else
     echo "listeners=PLAINTEXT://:9092" >>"$BROKER_PROPERTIES"
     echo "advertised.listeners=PLAINTEXT://${ADDRESS}:$BROKER_PORT" >>"$BROKER_PROPERTIES"
+    if [[ "$CONFLUENT_BROKER" = "true" ]]; then
+      echo "confluent.metadata.server.listeners=http://0.0.0.0:$BROKER_PORT" >>"$BROKER_PROPERTIES"
+    fi
   fi
 }
 
@@ -252,15 +289,12 @@ function fetchBrokerId() {
   else
     echo "$id"
   fi
-
 }
 
 # ===================================[main]===================================
 
 checkDocker
-generateDockerfile
 buildImageIfNeed "$IMAGE_NAME"
-
 if [[ "$RUN" != "true" ]]; then
   echo "docker image: $IMAGE_NAME is created"
   exit 0
@@ -289,6 +323,20 @@ setPropertyIfEmpty "num.partitions" "8"
 setPropertyIfEmpty "transaction.state.log.replication.factor" "1"
 setPropertyIfEmpty "offsets.topic.replication.factor" "1"
 setPropertyIfEmpty "transaction.state.log.min.isr" "1"
+if [[ "$CONFLUENT_BROKER" = "true" ]]; then
+    rejectProperty "metric.reporters"
+    setPropertyIfEmpty "confluent.metrics.reporter.zookeeper.connect" "$ZOOKEEPER_CONNECT"
+    setPropertyIfEmpty "metric.reporters" "io.confluent.metrics.reporter.ConfluentMetricsReporter"
+    setPropertyIfEmpty "confluent.metrics.reporter.bootstrap.servers" "${ADDRESS}:$BROKER_PORT"
+    setPropertyIfEmpty "confluent.metrics.reporter.topic.replicas" "1"
+    setPropertyIfEmpty "confluent.topic.replication.factor" "1"
+    setPropertyIfEmpty "confluent.license.topic.replication.factor" "1"
+    setPropertyIfEmpty "confluent.metadata.topic.replication.factor" "1"
+    setPropertyIfEmpty "confluent.security.event.logger.exporter.kafka.topic.replicas" "1"
+    setPropertyIfEmpty "confluent.balancer.topic.replication.factor" "1"
+    setPropertyIfEmpty "confluent.balancer.enable" "true"
+    setPropertyIfEmpty "confluent.topic.bootstrap.servers" "${ADDRESS}:$BROKER_PORT"
+fi
 setLogDirs
 
 docker run -d --init \
@@ -302,13 +350,17 @@ docker run -d --init \
   -p $BROKER_PORT:9092 \
   -p $BROKER_JMX_PORT:$BROKER_JMX_PORT \
   -p $EXPORTER_PORT:$EXPORTER_PORT \
-  "$IMAGE_NAME" ./bin/kafka-server-start.sh /tmp/broker.properties
+  "$IMAGE_NAME" "$SCRIPT_LOCATION_IN_CONTAINER" /tmp/broker.properties
 
 echo "================================================="
 [[ -n "$DATA_FOLDERS" ]] && echo "mount $DATA_FOLDERS to container: $CONTAINER_NAME"
 echo "broker id: $(fetchBrokerId)"
 echo "broker address: ${ADDRESS}:$BROKER_PORT"
-echo "jmx address: ${ADDRESS}:$BROKER_JMX_PORT"
+
+#confluent broker may not support JMX server now,so remove it temporarily
+if [[ "$CONFLUENT_BROKER" != "true" ]]; then
+    echo "jmx address: ${ADDRESS}:$BROKER_JMX_PORT"
+fi
 echo "exporter address: ${ADDRESS}:$EXPORTER_PORT"
 if [[ "$SASL" == "true" ]]; then
   user_jaas_file=/tmp/user-jaas-${BROKER_PORT}.conf
diff --git a/docker/start_zookeeper.sh b/docker/start_zookeeper.sh
index 59612197..75e5d638 100755
--- a/docker/start_zookeeper.sh
+++ b/docker/start_zookeeper.sh
@@ -99,4 +99,5 @@ fi
 
 echo "================================================="
 echo "run $DOCKER_FOLDER/start_broker.sh zookeeper.connect=$ADDRESS:$ZOOKEEPER_PORT to join kafka broker"
+echo "run env CONFLUENT_BROKER=true $DOCKER_FOLDER/start_broker.sh zookeeper.connect=$ADDRESS:$ZOOKEEPER_PORT to join confluent kafka broker"
 echo "================================================="


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 686d1e17..e8e2c7a7 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -39,6 +39,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     try (var executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
+            Producer.builder().brokers(bootstrapServers()).buildTransactional(),
             param,
             metrics,
             partition(param, admin),
@@ -82,6 +83,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     try (Executor executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
+            Producer.builder().brokers(bootstrapServers()).buildTransactional(),
             param,
             metrics,
             partition(param, admin),
@@ -117,6 +119,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     try (Executor executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
+            Producer.builder().brokers(bootstrapServers()).buildTransactional(),
             param,
             metrics,
             List.of(-1),
@@ -154,6 +157,14 @@ public class PerformanceTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testTransactionSet() {
+    var argument = new Performance.Argument();
+    Assertions.assertFalse(argument.transaction());
+    argument.transactionSize = 3;
+    Assertions.assertTrue(argument.transaction());
+  }
+
   @Test
   void testArgument() {
     String[] arguments1 = {
@@ -173,8 +184,6 @@ public class PerformanceTest extends RequireBrokerCluster {
       "1000records",
       "--record.size",
       "10KiB",
-      "--jmx.servers",
-      "localhost:9000@1",
       "--partitioner",
       "org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner",
       "--compression",
@@ -224,11 +233,6 @@ public class PerformanceTest extends RequireBrokerCluster {
         IllegalArgumentException.class,
         () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments8));
 
-    String[] arguments9 = {"--bootstrap.servers", "localhost:9092", "--jmx.servers", ""};
-    Assertions.assertThrows(
-        ParameterException.class,
-        () -> org.astraea.argument.Argument.parse(new Performance.Argument(), arguments9));
-
     String[] arguments10 = {"--bootstrap.servers", "localhost:9092", "--partitioner", ""};
     Assertions.assertThrows(
         ParameterException.class,
diff --git a/app/src/test/java/org/astraea/producer/ProducerTest.java b/app/src/test/java/org/astraea/producer/ProducerTest.java
index e8e123af..f5cd9bf3 100644
--- a/app/src/test/java/org/astraea/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/producer/ProducerTest.java
@@ -1,9 +1,12 @@
 package org.astraea.producer;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
@@ -53,4 +56,42 @@ public class ProducerTest extends RequireBrokerCluster {
       Assertions.assertArrayEquals(header.value(), actualHeader.value());
     }
   }
+
+  @Test
+  void testTransaction() {
+    var topicName = "testTransaction-" + System.currentTimeMillis();
+    var key = "key";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+    try (var producer =
+        Producer.builder()
+            .brokers(bootstrapServers())
+            .keySerializer(Serializer.STRING)
+            .buildTransactional()) {
+      var senders = new ArrayList<Sender<String, byte[]>>(3);
+      while (senders.size() < 3) {
+        senders.add(
+            producer
+                .sender()
+                .topic(topicName)
+                .key(key)
+                .timestamp(timestamp)
+                .headers(List.of(header)));
+      }
+      producer.transaction(senders);
+    }
+
+    Map<String, Object> prop = Map.of(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .fromBeginning()
+            .topics(Set.of(topicName))
+            .keyDeserializer(Deserializer.STRING)
+            .configs(prop)
+            .build()) {
+      var records = consumer.poll(Duration.ofSeconds(10));
+      Assertions.assertEquals(3, records.size());
+    }
+  }
 }