Add performance benchmarking script modes and README

* Add performance benchmarking script modes and README

* Fix createTopics return

* Add topic creation to benchmarks

* Remove needless batch size (msgs) 1

* Add performance example to semaphore

* Clean up the perf runner script

Author: milindl

.semaphore/semaphore.yml

@@ -108,6 +108,38 @@ blocks:
           commands:
             - npx eslint lib/kafkajs
 
+  - name: "Linux amd64: Performance"
+    dependencies: [ ]
+    task:
+      agent:
+        machine:
+          type: s1-prod-ubuntu20-04-amd64-4
+      env_vars:
+        - name: TARGET_PRODUCE_PERFORMANCE
+          value: "35"
+        - name: TARGET_CONSUME_PERFORMANCE
+          value: "18"
+        - name: TARGET_CTP_PERFORMANCE
+          value: "0.02"
+      prologue:
+        commands:
+          - sudo apt-get install -y librdkafka-dev bc
+          - export CKJS_LINKING=dynamic
+          - export BUILD_LIBRDKAFKA=0
+          - npm install
+          - npx node-pre-gyp --build-from-source clean
+          - npx node-pre-gyp --build-from-source configure
+          - npx node-pre-gyp --build-from-source build
+      jobs:
+        - name: "Performance Test"
+          commands:
+            - '[[ -z $DOCKERHUB_APIKEY ]] || docker login --username $DOCKERHUB_USER --password $DOCKERHUB_APIKEY'
+            - docker compose up -d && sleep 30
+            - export NODE_OPTIONS='--max-old-space-size=1536'
+            - cd examples/performance
+            - npm install
+            - ../../ci/tests/run_perf_tests.sh
+
   - name: "Linux amd64: Release"
     dependencies: [ ]
     run:

ci/tests/run_perf_test.sh

@@ -0,0 +1,63 @@
+#!/bin/bash
+
+testresultConfluentProducerConsumer=$(mktemp)
+testresultConfluentCtp=$(mktemp)
+testresultKjsProducerConsumer=$(mktemp)
+testresultKjsCtp=$(mktemp)
+
+MODE=confluent MESSAGE_COUNT=500000 node performance-consolidated.js --create-topics --consumer --producer 2>&1 | tee "$testresultConfluentProducerConsumer"
+MODE=kafkajs MESSAGE_COUNT=500000 node performance-consolidated.js --create-topics --consumer --producer 2>&1 | tee "$testresultKjsProducerConsumer"
+MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp 2>&1 | tee "$testresultConfluentCtp"
+MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp 2>&1 | tee "$testresultKjsCtp"
+
+producerConfluent=$(grep "=== Producer Rate:" "$testresultConfluentProducerConsumer" | cut -d':' -f2 | tr -d ' ')
+consumerConfluent=$(grep "=== Consumer Rate:" "$testresultConfluentProducerConsumer" | cut -d':' -f2 | tr -d ' ')
+ctpConfluent=$(grep "=== Consume-Transform-Produce Rate:" "$testresultConfluentCtp" | cut -d':' -f2 | tr -d ' ')
+producerKjs=$(grep "=== Producer Rate:" "$testresultKjsProducerConsumer" | cut -d':' -f2 | tr -d ' ')
+consumerKjs=$(grep "=== Consumer Rate:" "$testresultKjsProducerConsumer" | cut -d':' -f2 | tr -d ' ')
+ctpKjs=$(grep "=== Consume-Transform-Produce Rate:" "$testresultKjsCtp" | cut -d':' -f2 | tr -d ' ')
+
+echo "Producer rates: confluent $producerConfluent, kafkajs $producerKjs"
+echo "Consumer rates: confluent $consumerConfluent, kafkajs $consumerKjs"
+echo "CTP rates: confluent $ctpConfluent, kafkajs $ctpKjs"
+
+errcode=0
+
+# Compare against KJS
+if [[ $(echo "$producerConfluent < $producerKjs * 70 / 100" | bc -l) -eq 1 ]]; then
+  echo "Producer rates differ by more than 30%: confluent $producerConfluent, kafkajs $producerKjs"
+  errcode=1
+fi
+
+if [[ $(echo "$consumerConfluent < $consumerKjs * 70 / 100" | bc -l) -eq 1 ]]; then
+  echo "Consumer rates differ by more than 30%: confluent $consumerConfluent, kafkajs $consumerKjs"
+  errcode=1
+fi
+
+if [[ $(echo "$ctpConfluent < $ctpKjs * 70 / 100" | bc -l) -eq 1 ]]; then
+  echo "CTP rates differ by more than 30%: confluent $ctpConfluent, kafkajs $ctpKjs"
+  errcode=1
+fi
+
+# Compare against numbers set within semaphore config
+TARGET_PRODUCE="${TARGET_PRODUCE_PERFORMANCE:-35}"
+TARGET_CONSUME="${TARGET_CONSUME_PERFORMANCE:-18}"
+TARGET_CTP="${TARGET_CTP_PERFORMANCE:-0.02}"
+
+if [[ $(echo "$producerConfluent < $TARGET_PRODUCE" | bc -l) -eq 1 ]]; then
+  echo "Confluent producer rate is below target: $producerConfluent"
+  errcode=1
+fi
+
+if [[ $(echo "$consumerConfluent < $TARGET_CONSUME" | bc -l) -eq 1 ]]; then
+  echo "Confluent consumer rate is below target: $consumerConfluent"
+  errcode=1
+fi
+
+if [[ $(echo "$ctpConfluent < $TARGET_CTP" | bc -l) -eq 1 ]]; then
+  echo "Confluent CTP rate is below target: $ctpConfluent"
+  errcode=1
+fi
+
+exit $errcode
+

examples/performance/README.md

@@ -0,0 +1,39 @@
+# Performance Benchmarking
+
+The library can be benchmarked by running the following command:
+
+```bash
+node performance-consolidated.js [--producer] [--consumer] [--ctp] [--all]
+```
+
+The `--producer` flag will run the producer benchmark, the `--consumer` flag
+will run the consumer benchmark, and the `--ctp` flag will run the
+consume-transform-produce benchmark.
+
+The `--create-topics` flag will create the topics before running the benchmarks
+(and delete any existing topics of the same name). It's recommended to use this
+unless the number of partitions or replication factor needs to be changed.
+
+If no flags are provided, no benchmarks will be run. If the `--all` flag is
+provided, all benchmarks will be run ignoring any other flags.
+
+The benchmarks assume topics are already created (unless usig `--create-topics`).
+The consumer benchmark assumes that the topic already has at least `MESSAGE_COUNT` messages within,
+which can generally be done by running the producer benchmark along with it.
+
+The following environment variables can be set to configure the benchmark, with
+default values given in parentheses.
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| KAFKA_BROKERS | Kafka brokers to connect to | localhost:9092 |
+| KAFKA_TOPIC | Kafka topic to produce to/consume from | test-topic |
+| KAFKA_TOPIC2 | Kafka topic to produce to after consumption in consume-transform-produce | test-topic2 |
+| MESSAGE_COUNT | Number of messages to produce/consume | 1000000 |
+| MESSAGE_SIZE | Size of each message in bytes | 256 |
+| BATCH_SIZE | Number of messages to produce in a single batch | 100 |
+| COMPRESSION | Compression codec to use (None, GZIP, Snappy, LZ4, ZSTD) | None |
+| WARMUP_MESSAGES | Number of messages to produce before starting the produce benchmark | BATCH_SIZE * 10 |
+| MESSAGE_PROCESS_TIME_MS | Time to sleep after consuming each message in the consume-transform-produce benchmark. Simulates "transform". May be 0. | 5 |
+| CONSUME_TRANSFORM_PRODUCE_CONCURRENCY | partitionsConsumedConcurrently for the consume-transform-produce benchmark | 1 |
+| MODE | Mode to run the benchmarks in (confluent, kafkajs). Can be used for comparison with KafkaJS | confluent |

examples/performance/performance-consolidated.js

@@ -1,21 +1,37 @@
-const { runProducer, runConsumer, runConsumeTransformProduce } = require('./performance-primitives');
+const mode = process.env.MODE ? process.env.MODE : 'confluent';
 
-const { CompressionTypes } = require('../../').KafkaJS;
+let runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics;
+if (mode === 'confluent') {
+    ({ runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics } = require('./performance-primitives'));
+} else {
+    ({ runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics } = require('./performance-primitives-kafkajs'));
+}
 
 const brokers = process.env.KAFKA_BROKERS || 'localhost:9092';
 const topic = process.env.KAFKA_TOPIC || 'test-topic';
 const topic2 = process.env.KAFKA_TOPIC2 || 'test-topic2';
 const messageCount = process.env.MESSAGE_COUNT ? +process.env.MESSAGE_COUNT : 1000000;
 const messageSize = process.env.MESSAGE_SIZE ? +process.env.MESSAGE_SIZE : 256;
 const batchSize = process.env.BATCH_SIZE ? +process.env.BATCH_SIZE : 100;
-const compression = process.env.COMPRESSION || CompressionTypes.NONE;
+const compression = process.env.COMPRESSION || 'None';
 const warmupMessages = process.env.WARMUP_MESSAGES ? +process.env.WARMUP_MESSAGES : (batchSize * 10);
+const messageProcessTimeMs = process.env.MESSAGE_PROCESS_TIME_MS ? +process.env.MESSAGE_PROCESS_TIME_MS : 5;
+const ctpConcurrency = process.env.CONSUME_TRANSFORM_PRODUCE_CONCURRENCY ? +process.env.CONSUME_TRANSFORM_PRODUCE_CONCURRENCY : 1;
 
 (async function () {
     const producer = process.argv.includes('--producer');
     const consumer = process.argv.includes('--consumer');
     const ctp = process.argv.includes('--ctp');
     const all = process.argv.includes('--all');
+    const createTopics = process.argv.includes('--create-topics');
+
+    if (createTopics || all) {
+        console.log("=== Creating Topics (deleting if they exist already):");
+        console.log(`  Brokers: ${brokers}`);
+        console.log(`  Topic: ${topic}`);
+        console.log(`  Topic2: ${topic2}`);
+        await runCreateTopics(brokers, topic, topic2);
+    }
 
     if (producer || all) {
         console.log("=== Running Basic Producer Performance Test:")
@@ -48,7 +64,7 @@ const warmupMessages = process.env.WARMUP_MESSAGES ? +process.env.WARMUP_MESSAGE
         console.log(`  Message Count: ${messageCount}`);
         // Seed the topic with messages
         await runProducer(brokers, topic, batchSize, warmupMessages, messageCount, messageSize, compression);
-        const ctpRate = await runConsumeTransformProduce(brokers, topic, topic2, messageCount);
+        const ctpRate = await runConsumeTransformProduce(brokers, topic, topic2, messageCount, messageProcessTimeMs, ctpConcurrency);
         console.log("=== Consume-Transform-Produce Rate: ", ctpRate);
     }