A

benchmarks/src/it/docker-compose.yml → benchmarks/docker-compose.yml

@@ -37,5 +37,3 @@ services:
       CONFLUENT_METRICS_ENABLE: 'true'
       CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
 
-
-

benchmarks/src/main/resources/commands

@@ -5,4 +5,7 @@ docker exec -it broker kafka-console-consumer --bootstrap-server broker:9092 --t
 
 docker exec -it broker /bin/bash
 
-kafka-producer-perf-test --topic topic2 --num-records 100 --throughput -1 --producer-props bootstrap.servers=localhost:9092 batch.size=1000 acks=1 linger.ms=1000 buffer.memory=4294967296 request.timeout.ms=300000 --record-size 10
+ docker stop $(docker ps -a -q)
+
+#from root directory
+sbt 'benchmarks/jmh:run -o results.txt monix.kafka.benchmarks.KafkaProducerBenchmark'

benchmarks/src/main/scala/monix/kafka/benchmarks/Ap.scala

@@ -0,0 +1,5 @@
+package monix.kafka.benchmarks
+
+object Ap {
+
+}

benchmarks/src/main/scala/monix/kafka/benchmarks/KafkaFixture.scala

@@ -8,7 +8,10 @@ import org.scalatest.Suite
 import scala.util.Random
 trait KafkaFixture {
 
-  val topic_1P_2RF = "topic_1P_2RF"
+  val topic_1P_1RF = "topic_1P_1RF"
+  val topic_2P_1RF = "topic_2P_4RF"
+  val topic_2P_2RF = "topic_2P_2RF"
+  val topic_2P_4RF = "topic_2P_4RF"
 
   val producerConf = KafkaProducerConfig.default.copy(
     bootstrapServers = List("127.0.0.1:9092"),
@@ -21,9 +24,5 @@ trait KafkaFixture {
     groupId = "kafka-tests"
   )
 
-  val strToProducedRecord: String => Seq[ProducerRecord[String, String]] = {
-    v: String =>
-      Seq(new ProducerRecord[String, String](topic_1P_2RF, Random.nextInt().toString, v))
-  }
 
 }

benchmarks/src/main/scala/monix/kafka/benchmarks/KafkaProducerBenchmark.scala

@@ -22,12 +22,31 @@ class KafkaProducerBenchmark extends KafkaFixture {
 
   val producer = KafkaProducer[String, String](producerConf, io)
 
-  @Benchmark // kafka producer benchmark with 1 partition and 2 repartition factor
-  def produce_1P_2RF(): Seq[Option[RecordMetadata]] = {
-    val tasks = (0 until size).map(i => producer.send(topic = topic_1P_2RF, i.toString))
+  //syntax (P, P, RF) === (Parallelism factor, Partitions, Replication Factor)
+
+  @Benchmark
+  def produce_1P_1RF(): Seq[Option[RecordMetadata]] = {
+    val tasks = (0 until size).map(i => producer.send(topic = topic_1P_1RF, i.toString))
+    val results = Task.sequence(tasks).runSyncUnsafe()
+    require(results.filter(_.isDefined).isEmpty)
+    results
+  }
+
+  @Benchmark
+  def produce_2P_1RF(): Seq[Option[RecordMetadata]] = {
+    val tasks = (0 until size).map(i => producer.send(topic = topic_2P_1RF, i.toString))
+    val results = Task.sequence(tasks).runSyncUnsafe()
+    require(results.filter(_.isDefined).isEmpty)
+    results
+  }
+
+  @Benchmark
+  def produce_2P_2RF(): Seq[Option[RecordMetadata]] = {
+    val tasks = (0 until size).map(i => producer.send(topic = topic_2P_2RF, i.toString))
     val results = Task.sequence(tasks).runSyncUnsafe()
     require(results.filter(_.isDefined).isEmpty)
     results
   }
+
 
 }

benchmarks/src/it/start-kafka-cluster.sh → benchmarks/start-kafka-cluster.sh

@@ -19,5 +19,4 @@ docker ps
 sleep 15
 
 create_topic topic_1P_1RF 1 1
-create_topic topic_2P_1RF 1 1
-create_topic topic_2P_2RF 1 1
+create_topic topic_2P_1RF 2 1