Fix use warmup messages with headers in case messages from other partitions are consumed first

Author: emasab

ci/tests/run_perf_test.js

@@ -97,7 +97,8 @@ async function main() {
       const TERMINATE_TIMEOUT_MS_CONSUMERS = TERMINATE_TIMEOUT_MS + INITIAL_DELAY_MS + 2000;
       const TERMINATE_TIMEOUT_MS_LAG_MONITORING = TERMINATE_TIMEOUT_MS + 1000;
 
-      await runCommand(`MODE=${mode} node performance-consolidated.js --create-topics`);
+      const createTopicResult = await runCommand(`MODE=${mode} node performance-consolidated.js --create-topics`);
+      let ret = [createTopicResult];
 
       console.log(`Waiting 10s after topic creation before starting producer and consumers...`);
       await new Promise(resolve => setTimeout(resolve, 10000));
@@ -118,7 +119,7 @@ async function main() {
         allPromises.push(runCommand(`MODE=${mode} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_LAG_MONITORING} GROUPID_MONITOR=${groupIdEachBatch} node performance-consolidated.js --monitor-lag`));
       }
       const results = await Promise.allSettled(allPromises);
-      return results.map(r => r.status === 'fulfilled' ? r.value : '').join('\n');
+      return ret.concat(results.map(r => r.status === 'fulfilled' ? r.value : '')).join('\n');
     } else {
       console.log(`Running ${modeLabel} Producer/Consumer test...`);
       return runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} GROUPID_MESSAGE=${groupIdEachMessage} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --create-topics ${consumerParam} ${produceToSecondTopicParam} --producer`);

examples/performance/performance-consolidated.js

@@ -28,7 +28,7 @@ const randomness = process.env.RANDOMNESS ? +process.env.RANDOMNESS : 0.5;
 const initialDelayMs = process.env.INITIAL_DELAY_MS ? +process.env.INITIAL_DELAY_MS : 0;
 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 warmupMessages = process.env.WARMUP_MESSAGES ? +process.env.WARMUP_MESSAGES : ((batchSize * 10 > 1024) ? 1024 : (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;
 const consumerProcessingTime = process.env.CONSUMER_PROCESSING_TIME ? +process.env.CONSUMER_PROCESSING_TIME : 100;

examples/performance/performance-primitives-common.js

@@ -172,7 +172,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
     const skippedMessages = warmupMessages;
 
     const updateLatency = (receivedAt, numMessages, message, isT0T2) => {
-        if (!stats)
+        if (!stats || numMessages < 1)
             return;
 
         if (!message.headers || !message.headers['timestamp']) {
@@ -402,11 +402,18 @@ async function runProducer(producer, topic, batchSize, warmupMessages, totalMess
 
     console.log('Sending ' + warmupMessages + ' warmup messages.');
     while (warmupMessages > 0) {
+        const toProduce = Math.min(batchSize, warmupMessages);
         await producer.send({
             topic,
-            messages: messages.slice(0, batchSize)
+            messages: messages.slice(0, toProduce).map((msg) => ({
+                key: msg.key,
+                value: msg.value,
+                headers: {
+                    'timestamp': Date.now().toString(),
+                }
+            }))
         }, compression);
-        warmupMessages -= batchSize;
+        warmupMessages -= toProduce;
     }
     console.log('Sent warmup messages');