Add benchmark for send operation using kafka-clients

Author: dmytro-lazebnyi

README.md

@@ -1 +1 @@
-# performance-storage
+# Benchmarks of third-party tools

kafka-benchmarks/pom.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <artifactId>scalecube-third-party-benchmarks-parent</artifactId>
+    <groupId>io.scalecube</groupId>
+    <version>2.0.1-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>kafka-benchmarks</artifactId>
+
+  <properties>
+    <kafka.version>2.1.0</kafka.version>
+    <reactor-kafka.version>1.1.0.RELEASE</reactor-kafka.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-streams</artifactId>
+      <version>${kafka.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.projectreactor.kafka</groupId>
+      <artifactId>reactor-kafka</artifactId>
+      <version>${reactor-kafka.version}</version>
+    </dependency>
+  </dependencies>
+
+</project>

kafka-benchmarks/src/main/java/io/scalecube/benchmarks/kafka/client/KafkaState.java

@@ -0,0 +1,35 @@
+package io.scalecube.benchmarks.kafka.client;
+
+import io.scalecube.benchmarks.BenchmarkSettings;
+import io.scalecube.benchmarks.BenchmarkState;
+import java.util.Properties;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+
+final class KafkaState extends BenchmarkState<KafkaState> {
+
+  static final String TOPIC = "benchmark-topic";
+
+  private static final String SERVERS = "127.0.0.1:9092";
+
+  private final KafkaProducer<byte[], byte[]> producer;
+
+  KafkaState(BenchmarkSettings settings) {
+    super(settings);
+
+    producer = new KafkaProducer<>(producerProps());
+  }
+
+  KafkaProducer<byte[], byte[]> producer() {
+    return producer;
+  }
+
+  private static Properties producerProps() {
+    final Properties props = new Properties();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    return props;
+  }
+}

kafka-benchmarks/src/main/java/io/scalecube/benchmarks/kafka/client/ProducerBenchmark.java

@@ -0,0 +1,8 @@
+package io.scalecube.benchmarks.kafka.client;
+
+final class ProducerBenchmark {
+
+  public static void main(String[] args) {
+    ProducerSendScenario.runWith(args, KafkaState::new);
+  }
+}

kafka-benchmarks/src/main/java/io/scalecube/benchmarks/kafka/client/ProducerSendScenario.java

@@ -0,0 +1,41 @@
+package io.scalecube.benchmarks.kafka.client;
+
+import io.scalecube.benchmarks.BenchmarkSettings;
+import io.scalecube.benchmarks.metrics.BenchmarkTimer;
+import io.scalecube.benchmarks.metrics.BenchmarkTimer.Context;
+import java.time.Duration;
+import java.util.function.Function;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import reactor.core.publisher.Mono;
+
+final class ProducerSendScenario {
+
+  static void runWith(String[] args, Function<BenchmarkSettings, KafkaState> stateSupplier) {
+    BenchmarkSettings settings =
+        BenchmarkSettings.from(args)
+            .numberThreads(1)
+            .warmUpDuration(Duration.ofSeconds(1))
+            .executionTaskDuration(Duration.ofMinutes(3))
+            .build();
+
+    stateSupplier
+        .apply(settings)
+        .runForAsync(
+            state -> {
+              BenchmarkTimer timer = state.timer("write");
+
+              return iteration -> {
+                byte[] value = "Hello".getBytes();
+
+                ProducerRecord<byte[], byte[]> record =
+                    new ProducerRecord<>(KafkaState.TOPIC, value);
+
+                Context time = timer.time();
+
+                state.producer().send(record, (metadata, exception) -> time.stop());
+
+                return Mono.empty();
+              };
+            });
+  }
+}

pom.xml

@@ -35,6 +35,7 @@
 
   <modules>
     <module>chronicle-map-benchmarks</module>
+    <module>kafka-benchmarks</module>
     <module>lmdb-benchmarks</module>
     <module>rocksdb-benchmarks</module>
     <module>storage-benchmarks-common</module>