Configurable partitions consumed concurrently

Author: emasab

examples/performance/performance-consolidated.js

@@ -21,6 +21,7 @@ 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 || 'None';
 const numPartitions = process.env.PARTITIONS ? +process.env.PARTITIONS : 3;
+const partitionsConsumedConcurrently = process.env.PARTITIONS_CONSUMED_CONCURRENTLY ? +process.env.PARTITIONS_CONSUMED_CONCURRENTLY : 1;
 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;
@@ -121,7 +122,7 @@ function logParameters(parameters) {
         console.log(`  Topic: ${topic}`);
         console.log(`  Message Count: ${messageCount}`);
         startTrackingMemory();
-        const consumerRate = await runConsumer(parameters, topic, warmupMessages, messageCount, false, stats);
+        const consumerRate = await runConsumer(parameters, topic, warmupMessages, messageCount, false, partitionsConsumedConcurrently, stats);
         endTrackingMemory(`consumer-memory-message-${mode}.json`);
         console.log("=== Consumer Rate (eachMessage): ", consumerRate);
         console.log("=== Consumption time (eachMessage): ", stats.durationSeconds);
@@ -134,7 +135,7 @@ function logParameters(parameters) {
         console.log(`  Topic: ${topic}`);
         console.log(`  Message Count: ${messageCount}`);
         startTrackingMemory();
-        const consumerRate = await runConsumer(parameters, topic, warmupMessages, messageCount, true, stats);
+        const consumerRate = await runConsumer(parameters, topic, warmupMessages, messageCount, true, partitionsConsumedConcurrently, stats);
         endTrackingMemory(`consumer-memory-batch-${mode}.json`);
         console.log("=== Consumer Rate (eachBatch): ", consumerRate);
         console.log("=== Average eachBatch lag: ", stats.averageOffsetLag);

examples/performance/performance-primitives-common.js

@@ -1,6 +1,6 @@
 const { hrtime } = require('process');
 
-async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eachBatch, stats) {
+async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats) {
     await consumer.connect();
     await consumer.subscribe({ topic });
 
@@ -17,6 +17,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
 
     console.log("Starting consumer.");
     let consumeMethod = {
+        partitionsConsumedConcurrently,
         eachMessage: async ({ topic, partition, message }) => {
             messagesReceived++;
 
@@ -38,6 +39,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
     }
     if (eachBatch) {
         consumeMethod = {
+            partitionsConsumedConcurrently,
             eachBatch: async ({ batch }) => {
                 const messagesBeforeBatch = messagesReceived;
                 const topic = batch.topic;

examples/performance/performance-primitives-kafkajs.js

@@ -168,8 +168,8 @@ function newCompatibleConsumer(parameters) {
     return new CompatibleConsumer(consumer);
 }
 
-async function runConsumer(parameters, topic, warmupMessages, totalMessageCnt, eachBatch, stats) {
-    return runConsumerCommon(newCompatibleConsumer(parameters), topic, warmupMessages, totalMessageCnt, eachBatch, stats);
+async function runConsumer(parameters, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats) {
+    return runConsumerCommon(newCompatibleConsumer(parameters), topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats);
 }
 
 async function runConsumeTransformProduce(parameters, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {

examples/performance/performance-primitives.js

@@ -172,8 +172,8 @@ function newCompatibleConsumer(parameters) {
     return new CompatibleConsumer(consumer);
 }
 
-async function runConsumer(parameters, topic, warmupMessages, totalMessageCnt, eachBatch, stats) {
-    return runConsumerCommon(newCompatibleConsumer(parameters), topic, warmupMessages, totalMessageCnt, eachBatch, stats);
+async function runConsumer(parameters, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats) {
+    return runConsumerCommon(newCompatibleConsumer(parameters), topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats);
 }
 
 async function runConsumeTransformProduce(parameters, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {