Performance testing tool

README.md

@@ -14,10 +14,11 @@ This project offers many kafka tools to simplify the life for kafka users.
 
 1. [Kafka quick start](#kafka-cluster-quick-start): set up a true kafka cluster in one minute
 2. [Kafka benchmark](#latency-benchmark): run producers/consumers to test the performance and consistency for kafka cluster
-3. [Kafka offset explorer](#topic-explorer): check the start/end offsets of kafka topics
-4. [Kafka official tool](#kafka-official-tool): run any one specific kafka official tool. All you have to prepare is the docker env.
-5. [Kafka metric client](#kafka-metric-client): utility for accessing kafka Mbean metrics via JMX.
-6. [Replica Collie](#replica-collie): move replicas from brokers to others. You can use this tool to obstruct specific brokers from hosting specific topics. 
+3. [Kafka performance](#Performance-Benchmark): check producing/consuming performance.
+4. [Kafka offset explorer](#topic-explorer): check the start/end offsets of kafka topics
+5. [Kafka official tool](#kafka-official-tool): run any one specific kafka official tool. All you have to prepare is the docker env.
+6. [Kafka metric client](#kafka-metric-client): utility for accessing kafka Mbean metrics via JMX.
+7. [Replica Collie](#replica-collie): move replicas from brokers to others. You can use this tool to obstruct specific brokers from hosting specific topics.
 
 [Release page](https://github.com/skiptests/astraea/releases) offers the uber jar including all tools.
 ```shell
@@ -94,6 +95,28 @@ java -jar app-0.0.1-SNAPSHOT-all.jar latency --bootstrap.servers 192.168.50.224:
 
 ---
 
+## Performance Benchmark
+This tool is used test to following metrics.
+1. publish latency: the time of completing producer data request
+2. E2E latency: the time for a record to travel through Kafka
+3. input rate: sum of consumer inputs in MByte per second
+4. output rate: sum of producer outputs in MByte per second
+
+Run the benchmark from source
+```shell
+./gradlew run --args="Performance --bootstrap.servers localhost:9092 --topic topic --topicConfig partitions:10,replicationFactor:3 --producers 5 --consumers 1 --records 100000 --recordSize 10000"
+```
+### Performance Benchmark Configurations
+1. --bootstrap.servers: the server to connect to
+2. --topic: the topic name
+3. --partitions: topic config when creating new topic. Default: 1 
+4. --replicas: topic config when creating new topic. Default: 1
+5. --consumers: the number of consumers (threads). Default: 1
+6. --records: the total number of records sent by the producers. Default: 1000
+7. --record.size: the record size in byte. Default: 1024 byte
+
+---
+
 ## Topic Explorer
 
 This tool can expose both earliest offset and latest offset for all (public and private) topics.

app/src/main/java/org/astraea/App.java

@@ -6,6 +6,7 @@
 import java.util.List;
 import java.util.Map;
 import org.astraea.metrics.kafka.KafkaMetricClientApp;
+import org.astraea.performance.Performance;
 import org.astraea.performance.latency.End2EndLatency;
 import org.astraea.topic.ReplicaCollie;
 import org.astraea.topic.TopicExplorer;
@@ -16,7 +17,8 @@ public class App {
           "latency", End2EndLatency.class,
           "offset", TopicExplorer.class,
           "metrics", KafkaMetricClientApp.class,
-          "replica", ReplicaCollie.class);
+          "replica", ReplicaCollie.class,
+          "performance", Performance.class);
 
   static void execute(Map<String, Class<?>> mains, List<String> args) throws Throwable {
 

app/src/main/java/org/astraea/argument/ArgumentUtil.java

@@ -96,6 +96,13 @@ public Boolean convert(String value) {
     }
   }
 
+  public static class ShortConverter implements IStringConverter<Short> {
+    @Override
+    public Short convert(String value) {
+      return Short.valueOf(value);
+    }
+  }
+
   private static <C extends Collection<T>, T> C requireNonEmpty(C collection) {
     if (collection.isEmpty()) throw new ParameterException("array type can't be empty");
     return collection;

app/src/main/java/org/astraea/performance/ComponentFactory.java

@@ -0,0 +1,78 @@
+package org.astraea.performance;
+
+import java.util.Collections;
+import java.util.Properties;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+
+/** An interface used for creating producer, consumer. */
+public interface ComponentFactory {
+  Producer createProducer();
+
+  Consumer createConsumer();
+
+  /**
+   * (Optional) Setting partitioner for producer. If not set, use default partitioner RoundRobin
+   *
+   * @param partitionerName The partitioner class to use
+   * @return This factory
+   */
+  default ComponentFactory partitioner(String partitionerName) {
+    return this;
+  }
+
+  /**
+   * (Optional) JMX server for user defined partitioner. Default partitioner will not use this
+   * attribute
+   *
+   * @param JMXAddress The jmx address to connect to
+   * @return This factory
+   */
+  default ComponentFactory JMXAddress(String JMXAddress) {
+    return this;
+  }
+
+  /**
+   * Used for creating Kafka producer, consumer of the same Kafka server and the same topic. The
+   * consumers generated by the same object from `fromKafka(brokers)` subscribe the same topic and
+   * have the same groupID.
+   */
+  static ComponentFactory fromKafka(String brokers, String topic) {
+    return new ComponentFactory() {
+      private final String groupId = "groupId:" + System.currentTimeMillis();
+      private String partitionerName = DefaultPartitioner.class.getName();
+      private String JMXAddress = "0.0.0.0@0";
+      /** Create Producer with KafkaProducer<byte[], byte[]> functions */
+      @Override
+      public Producer createProducer() {
+        Properties prop = new Properties();
+        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        prop.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, partitionerName);
+        prop.put("jmx_servers", JMXAddress);
+        return Producer.fromKafka(prop, topic);
+      }
+
+      /** Create Consumer with KafkaConsumer<byte[], byte[]> functions */
+      @Override
+      public Consumer createConsumer() {
+        Properties prop = new Properties();
+        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        prop.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+        return Consumer.fromKafka(prop, Collections.singleton(topic));
+      }
+
+      @Override
+      public ComponentFactory partitioner(String partitionerName) {
+        this.partitionerName = partitionerName;
+        return this;
+      }
+
+      @Override
+      public ComponentFactory JMXAddress(String JMXAddress) {
+        this.JMXAddress = JMXAddress;
+        return this;
+      }
+    };
+  }
+}

app/src/main/java/org/astraea/performance/Consumer.java

@@ -0,0 +1,43 @@
+package org.astraea.performance;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Properties;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+
+/** An interface for polling records. */
+public interface Consumer {
+  ConsumerRecords<byte[], byte[]> poll(Duration timeout);
+
+  void wakeup();
+
+  void cleanup();
+
+  /** Create a Consumer with KafkaConsumer<byte[], byte[]> functionality */
+  static Consumer fromKafka(Properties prop, Collection<String> topics) {
+
+    var kafkaConsumer =
+        new KafkaConsumer<byte[], byte[]>(
+            prop, new ByteArrayDeserializer(), new ByteArrayDeserializer());
+    kafkaConsumer.subscribe(topics);
+    return new Consumer() {
+
+      @Override
+      public ConsumerRecords<byte[], byte[]> poll(Duration timeout) {
+        return kafkaConsumer.poll(timeout);
+      }
+
+      @Override
+      public void wakeup() {
+        kafkaConsumer.wakeup();
+      }
+
+      @Override
+      public void cleanup() {
+        kafkaConsumer.close();
+      }
+    };
+  }
+}

app/src/main/java/org/astraea/performance/Metrics.java

@@ -0,0 +1,67 @@
+package org.astraea.performance;
+
+import java.util.concurrent.atomic.LongAdder;
+
+/** Used to record statistics. This is thread safe. */
+public class Metrics {
+  private double avgLatency;
+  private long num;
+  private long max;
+  private long min;
+  private final LongAdder bytes;
+
+  public Metrics() {
+    avgLatency = 0;
+    num = 0;
+    max = 0;
+    min = Long.MAX_VALUE;
+    bytes = new LongAdder();
+  }
+
+  /** Simultaneously add latency and bytes. */
+  public synchronized void put(long latency, long bytes) {
+    putLatency(latency);
+    addBytes(bytes);
+  }
+  /** Add a new value to latency metric. */
+  public synchronized void putLatency(long latency) {
+    if (min > latency) min = latency;
+    if (max < latency) max = latency;
+    ++num;
+    avgLatency += (((double) latency) - avgLatency) / (double) num;
+  }
+  /** Add a new value to bytes. */
+  public synchronized void addBytes(long bytes) {
+    this.bytes.add(bytes);
+  }
+
+  /** Get the number of latency put. */
+  public synchronized long num() {
+    return num;
+  }
+  /** Get the maximum of latency put. */
+  public synchronized long max() {
+    return max;
+  }
+  /** Get the minimum of latency put. */
+  public synchronized long min() {
+    return min;
+  }
+  /** Get the average latency. */
+  public synchronized double avgLatency() {
+    return avgLatency;
+  }
+  /** Reset to 0 and returns the old value of bytes */
+  public synchronized long bytesThenReset() {
+    return this.bytes.sumThenReset();
+  }
+  /** Set all attributes to default value */
+  public synchronized void reset() {
+    avgLatency = 0;
+    num = 0;
+    max = 0;
+    // 初始為最大的integer值
+    min = Long.MAX_VALUE;
+    bytes.reset();
+  }
+}