CONSUMER_MAX_BATCH_SIZE in performance test

Author: emasab

ci/tests/run_perf_test.js

@@ -47,6 +47,9 @@ async function main() {
   if (concurrentRun) {
     skipCtpTest = true;
   }
+  if (!process.env.CONSUMER_MAX_BATCH_SIZE) {
+    process.env.CONSUMER_MAX_BATCH_SIZE = '-1';
+  }
   if (!process.env.PARTITIONS_CONSUMED_CONCURRENTLY) {
     process.env.PARTITIONS_CONSUMED_CONCURRENTLY = '2';
   }

examples/performance/performance-consolidated.js

@@ -20,8 +20,8 @@ const saslPassword = process.env.SASL_PASSWORD;
 const topic = process.env.KAFKA_TOPIC || `test-topic-${mode}`;
 const topic2 = process.env.KAFKA_TOPIC2 || `test-topic2-${mode}`;
 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 messageSize = process.env.MESSAGE_SIZE ? +process.env.MESSAGE_SIZE : 4096;
+const batchSize = process.env.PRODUCER_BATCH_SIZE ? +process.env.PRODUCER_BATCH_SIZE : 100;
 const compression = process.env.COMPRESSION || 'None';
 // Between 0 and 1, percentage of random bytes in each message
 const randomness = process.env.RANDOMNESS ? +process.env.RANDOMNESS : 0.5;

examples/performance/performance-primitives.js

@@ -20,6 +20,7 @@ module.exports = {
 };
 
 
+const CONSUMER_MAX_BATCH_SIZE = process.env.CONSUMER_MAX_BATCH_SIZE ? +process.env.CONSUMER_MAX_BATCH_SIZE : null;
 const IS_HIGHER_LATENCY_CLUSTER = process.env.IS_HIGHER_LATENCY_CLUSTER === 'true';
 
 function baseConfiguration(parameters) {
@@ -156,6 +157,9 @@ function newCompatibleConsumer(parameters, eachBatch) {
     const autoCommitOpts = autoCommit > 0 ? 
         { 'enable.auto.commit': true, 'auto.commit.interval.ms': autoCommit } :
         { 'enable.auto.commit': false };
+    const jsOpts = (eachBatch && CONSUMER_MAX_BATCH_SIZE !== null) ? {
+        'js.consumer.max.batch.size': CONSUMER_MAX_BATCH_SIZE
+    } : {};
     const higherLatencyClusterOpts = IS_HIGHER_LATENCY_CLUSTER ? {
         'max.partition.fetch.bytes': '8388608'
     } : {};
@@ -170,6 +174,7 @@ function newCompatibleConsumer(parameters, eachBatch) {
         'auto.offset.reset': 'earliest',
         'fetch.queue.backoff.ms': '100',
         ...autoCommitOpts,
+        ...jsOpts,
         ...higherLatencyClusterOpts,
     });
     return new CompatibleConsumer(consumer);