Fix CI performance run

Author: emasab

ci/tests/run_perf_test.js

@@ -42,8 +42,11 @@ function belowTarget(value, target) {
 async function main() {
   // Run performance tests and store outputs in memory
   const messageCount = process.env.MESSAGE_COUNT ? +process.env.MESSAGE_COUNT : 50000;
-  const skipCtpTest = process.env.SKIP_CTP_TEST ? process.env.SKIP_CTP_TEST === 'true' : false;
+  let skipCtpTest = process.env.SKIP_CTP_TEST ? process.env.SKIP_CTP_TEST === 'true' : false;
   const concurrentRun = process.env.CONCURRENT_RUN ? process.env.CONCURRENT_RUN === 'true' : false;
+  if (concurrentRun) {
+    skipCtpTest = true;
+  }
   const consumerMode = process.env.CONSUMER_MODE || 'all';
   const produceToSecondTopic = process.env.PRODUCE_TO_SECOND_TOPIC ? process.env.PRODUCE_TO_SECOND_TOPIC === 'true' : false;
   const produceToSecondTopicParam = produceToSecondTopic ? '--produce-to-second-topic' : '';
@@ -119,11 +122,11 @@ async function main() {
 
   console.log('Running Confluent CTP test...');
   const outputConfluentCtp = skipCtpTest ? '' :
-    runCommand('MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp');
+    (await runCommand('MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp'));
 
   console.log('Running KafkaJS CTP test...');
   const outputKjsCtp = skipCtpTest ? '' :
-    runCommand('MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp');
+    (await runCommand('MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp'));
 
   // Extract Confluent results
   let ctpConfluent, ctpKjs;
@@ -278,9 +281,11 @@ async function main() {
     console.log(`Consumption time (eachMessage): confluent ${consumerConfluentTime}, kafkajs ${consumerKjsTime}`);
     console.log(`Average RSS (eachMessage): confluent ${consumerConfluentMessageAverageRSS}, kafkajs ${consumerKjsMessageAverageRSS}`);
     console.log(`Max RSS (eachMessage): confluent ${consumerConfluentMessageMaxRSS}, kafkajs ${consumerKjsMessageMaxRSS}`);
-    console.log(`Average broker lag (eachMessage): confluent ${consumerConfluentMessageAverageBrokerLag}, kafkajs ${consumerKjsMessageAverageBrokerLag}`);
-    console.log(`Max broker lag (eachMessage): confluent ${consumerConfluentMessageMaxBrokerLag}, kafkajs ${consumerKjsMessageMaxBrokerLag}`);
-    console.log(`Sample size for broker lag measurement (eachMessage): confluent ${consumerConfluentMessageTotalLagMeasurements}, kafkajs ${consumerKjsMessageTotalLagMeasurements}`);
+    if (concurrentRun) {
+      console.log(`Average broker lag (eachMessage): confluent ${consumerConfluentMessageAverageBrokerLag}, kafkajs ${consumerKjsMessageAverageBrokerLag}`);
+      console.log(`Max broker lag (eachMessage): confluent ${consumerConfluentMessageMaxBrokerLag}, kafkajs ${consumerKjsMessageMaxBrokerLag}`);
+      console.log(`Sample size for broker lag measurement (eachMessage): confluent ${consumerConfluentMessageTotalLagMeasurements}, kafkajs ${consumerKjsMessageTotalLagMeasurements}`);
+    }
   }
   if (consumerModeAll || consumerModeEachBatch) {
     console.log(`Consumer rates MB/s (eachBatch): confluent ${consumerConfluentBatch}, kafkajs ${consumerKjsBatch}`);
@@ -297,9 +302,11 @@ async function main() {
     console.log(`Average eachBatch size: confluent ${consumerConfluentBatchAverageSize}, kafkajs ${consumerKjsBatchAverageSize}`);
     console.log(`Average RSS (eachBatch): confluent ${consumerConfluentBatchAverageRSS}, kafkajs ${consumerKjsBatchAverageRSS}`);
     console.log(`Max RSS (eachBatch): confluent ${consumerConfluentBatchMaxRSS}, kafkajs ${consumerKjsBatchMaxRSS}`);
-    console.log(`Average broker lag (eachBatch): confluent ${consumerConfluentBatchAverageBrokerLag}, kafkajs ${consumerKjsBatchAverageBrokerLag}`);
-    console.log(`Max broker lag (eachBatch): confluent ${consumerConfluentBatchMaxBrokerLag}, kafkajs ${consumerKjsBatchMaxBrokerLag}`);
-    console.log(`Sample size for broker lag measurement (eachBatch): confluent ${consumerConfluentBatchTotalLagMeasurements}, kafkajs ${consumerKjsBatchTotalLagMeasurements}`);
+    if (concurrentRun) {
+      console.log(`Average broker lag (eachBatch): confluent ${consumerConfluentBatchAverageBrokerLag}, kafkajs ${consumerKjsBatchAverageBrokerLag}`);
+      console.log(`Max broker lag (eachBatch): confluent ${consumerConfluentBatchMaxBrokerLag}, kafkajs ${consumerKjsBatchMaxBrokerLag}`);
+      console.log(`Sample size for broker lag measurement (eachBatch): confluent ${consumerConfluentBatchTotalLagMeasurements}, kafkajs ${consumerKjsBatchTotalLagMeasurements}`);
+    }
   }
   if (!concurrentRun) {
     console.log(`Average RSS: confluent ${consumerConfluentAverageRSS}, kafkajs ${consumerKjsAverageRSS}`);

examples/performance/performance-consolidated.js

@@ -207,7 +207,9 @@ function logParameters(parameters) {
         console.log(`  ProduceTopic: ${topic2}`);
         console.log(`  Message Count: ${messageCount}`);
         // Seed the topic with messages
-        await runProducer(parameters, topic, batchSize, warmupMessages, messageCount, messageSize, compression);
+        await runProducerCKJS(parameters, topic, batchSize,
+            warmupMessages, messageCount, messageSize, compression,
+            randomness, limitRPS);
         startTrackingMemory();
         const ctpRate = await runConsumeTransformProduce(parameters, topic, topic2, warmupMessages, messageCount, messageProcessTimeMs, ctpConcurrency);
         endTrackingMemory('consume-transform-produce', `consume-transform-produce-${mode}.json`);

examples/performance/performance-primitives-common.js

@@ -14,7 +14,7 @@ else {
     }
 }
 
-function installHandlers() {
+function installHandlers(useTerminateTimeout) {
     const handlers = {
         terminationRequested: false,
         terminateTimeout: null,
@@ -26,8 +26,10 @@ function installHandlers() {
     process.on('SIGINT', terminationRequestedCallback);
     process.on('SIGTERM', terminationRequestedCallback);
     handlers.terminationRequestedCallback = terminationRequestedCallback;
-    handlers.terminateTimeout = setTimeout(terminationRequestedCallback,
-                                  TERMINATE_TIMEOUT_MS);
+    if (useTerminateTimeout) {
+        handlers.terminateTimeout = setTimeout(terminationRequestedCallback,
+            TERMINATE_TIMEOUT_MS);
+    }
     return handlers;
 }
 
@@ -57,7 +59,7 @@ function genericProduceToTopic(producer, topic, messages) {
 }
 
 async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eachBatch, partitionsConsumedConcurrently, stats, actionOnMessages) {
-    const handlers = installHandlers();
+    const handlers = installHandlers(totalMessageCnt === -1);
     await consumer.connect();
     await consumer.subscribe({ topic });
 
@@ -179,12 +181,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
                     if (!startTime) {
                         startTime = hrtime.bigint();
                     } else if (totalMessageCnt > 0 && messagesMeasured >= totalMessageCnt) {
-                        let durationNanos = Number(hrtime.bigint() - startTime);
-                        durationSeconds = durationNanos / 1e9;
-                        rate = durationNanos === 0 ? Infinity :
-                            (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
-                        console.log(`Recvd ${messagesMeasured} messages in ${durationSeconds} seconds, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
-                        consumer.pause([{ topic }]);
+                        stopConsuming();
                     }
                 }
 
@@ -243,7 +240,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
 }
 
 async function runProducer(producer, topic, batchSize, warmupMessages, totalMessageCnt, msgSize, compression, randomness, limitRPS) {
-    const handlers = installHandlers();
+    const handlers = installHandlers(totalMessageCnt === -1);
     let totalMessagesSent = 0;
     let totalBytesSent = 0;
 
@@ -347,7 +344,7 @@ async function runProducer(producer, topic, batchSize, warmupMessages, totalMess
 }
 
 async function runLagMonitoring(admin, topic) {
-    const handlers = installHandlers();
+    const handlers = installHandlers(true);
     let groupId = process.env.GROUPID_MONITOR;
     if (!groupId) {
         throw new Error("GROUPID_MONITOR environment variable not set");