Configurable batch size #389
+482
−252
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -39,11 +39,23 @@ | |
|
|
||
|
|
||
|
|
||
| async function main() { | ||
| // Run performance tests and store outputs in memory | ||
| const messageCount = process.env.MESSAGE_COUNT ? +process.env.MESSAGE_COUNT : 50000; | ||
| 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; | ||
| } | ||
| 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'; | ||
| } | ||
| if (!process.env.COMPRESSION) { | ||
| process.env.COMPRESSION = 'GZIP'; | ||
| } | ||
| 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' : ''; | ||
|
|
@@ -58,10 +70,10 @@ | |
| } | ||
| let outputConfluentProducerConsumer; | ||
| let outputKjsProducerConsumer; | ||
| const groupIdEachMessageConfluent = `test-group-confluent-message-` + Math.random(); | ||
| const groupIdEachBatchConfluent = `test-group-confluent-batch-` + Math.random(); | ||
| const groupIdEachMessageKafkaJS = `test-group-kafkajs-message-` + Math.random(); | ||
| const groupIdEachBatchKafkaJS = `test-group-kafkajs-batch-` + Math.random(); | ||
| if (consumerModeAll || consumerModeEachMessage) { | ||
| console.log(`Confluent eachMessage group id: ${groupIdEachMessageConfluent}`); | ||
| console.log(`KafkaJS eachMessage group id: ${groupIdEachMessageKafkaJS}`); | ||
|
|
@@ -78,24 +90,32 @@ | |
|
|
||
| if (concurrentRun) { | ||
| console.log(`Running ${modeLabel} Producer/Consumer test (concurrently)...`); | ||
| const INITIAL_DELAY_MS = 10000; | ||
| const TERMINATE_TIMEOUT_MS = process.env.TERMINATE_TIMEOUT_MS ? +process.env.TERMINATE_TIMEOUT_MS : 600000; | ||
| // Wait INITIAL_DELAY_MS more to see if all lag is caught up, start earlier than the producer to check | ||
| // E2E latencies more accurately. | ||
| 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`); | ||
|
|
||
| console.log(`Waiting 10s after topic creation before starting producer and consumers...`); | ||
| await new Promise(resolve => setTimeout(resolve, 10000)); | ||
|
|
||
| console.log(`Starting producer and consumers...`); | ||
| const allPromises = []; | ||
| allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} node performance-consolidated.js --producer`)); | ||
| if (consumerModeAll || consumerModeEachMessage) { | ||
| allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} INITIAL_DELAY_MS=0 TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MESSAGE=${groupIdEachMessage} node performance-consolidated.js --consumer-each-message ${produceToSecondTopicParam}`)); | ||
| } | ||
| if (consumerModeAll || consumerModeEachBatch) { | ||
| allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} INITIAL_DELAY_MS=0 TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --consumer-each-batch ${produceToSecondTopicParam}`)); | ||
| } | ||
| if (consumerModeAll || consumerModeEachMessage) { | ||
| allPromises.push(runCommand(`MODE=${mode} INITIAL_DELAY_MS=${INITIAL_DELAY_MS} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_LAG_MONITORING} GROUPID_MONITOR=${groupIdEachMessage} node performance-consolidated.js --monitor-lag`)); | ||
| } | ||
| if (consumerModeAll || consumerModeEachBatch) { | ||
| 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'); | ||
|
|
@@ -119,11 +139,11 @@ | |
|
|
||
| console.log('Running Confluent CTP test...'); | ||
| const outputConfluentCtp = skipCtpTest ? '' : | ||
| (await runCommand('MODE=confluent MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp')); | ||
|
|
||
| console.log('Running KafkaJS CTP test...'); | ||
| const outputKjsCtp = skipCtpTest ? '' : | ||
| (await runCommand('MODE=kafkajs MESSAGE_COUNT=5000 node performance-consolidated.js --create-topics --ctp')); | ||
|
|
||
| // Extract Confluent results | ||
| let ctpConfluent, ctpKjs; | ||
|
|
@@ -230,22 +250,22 @@ | |
| consumerKjsMessageRate = extractValue(outputKjsProducerConsumer, '=== Consumer Rate msg/s (eachMessage):'); | ||
| consumerKjsMessageAvgLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T1 (eachMessage):'); | ||
| consumerKjsMessageMaxLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T1 (eachMessage):'); | ||
| consumerKjsMessageAvgLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T2 (eachMessage):'); | ||
|
Check failure on line 253 in ci/tests/run_perf_test.js
|
||
| consumerKjsMessageMaxLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T2 (eachMessage):'); | ||
|
Check failure on line 254 in ci/tests/run_perf_test.js
|
||
| consumerKjsTime = extractValue(outputKjsProducerConsumer, '=== Consumption time (eachMessage):'); | ||
| consumerKjsMessageAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average consumer-each-message RSS KB:'); | ||
| consumerKjsMessageMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max consumer-each-message RSS KB:'); | ||
| consumerKjsMessageAverageBrokerLag = extractValue(outputKjsProducerConsumer, `=== Average broker lag (${groupIdEachMessageKafkaJS}):`); | ||
| consumerKjsMessageMaxBrokerLag = extractValue(outputKjsProducerConsumer, `=== Max broker lag (${groupIdEachMessageKafkaJS}):`); | ||
| consumerKjsMessageTotalLagMeasurements = extractValue(outputKjsProducerConsumer, `=== Sample size for broker lag measurement (${groupIdEachMessageKafkaJS}):`); | ||
| } | ||
| if (consumerModeAll || consumerModeEachBatch) { | ||
| consumerKjsBatch = extractValue(outputKjsProducerConsumer, '=== Consumer Rate MB/s (eachBatch):'); | ||
| consumerKjsBatchRate = extractValue(outputKjsProducerConsumer, '=== Consumer Rate msg/s (eachBatch):'); | ||
| consumerKjsBatchAvgLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T1 (eachBatch):'); | ||
| consumerKjsBatchMaxLatencyT0T1 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T1 (eachBatch):'); | ||
| consumerKjsBatchAvgLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer average E2E latency T0-T2 (eachBatch):'); | ||
|
Check failure on line 267 in ci/tests/run_perf_test.js
|
||
| consumerKjsBatchMaxLatencyT0T2 = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency T0-T2 (eachBatch):'); | ||
|
Check failure on line 268 in ci/tests/run_perf_test.js
|
||
| consumerKjsBatchTime = extractValue(outputKjsProducerConsumer, '=== Consumption time (eachBatch):'); | ||
| consumerKjsBatchAverageLag = extractValue(outputKjsProducerConsumer, '=== Average eachBatch lag:'); | ||
| consumerKjsBatchMaxLag = extractValue(outputKjsProducerConsumer, '=== Max eachBatch lag:'); | ||
|
|
@@ -269,37 +289,41 @@ | |
| if (consumerModeAll || consumerModeEachMessage) { | ||
| console.log(`Consumer rates MB/s (eachMessage): confluent ${consumerConfluentMessage}, kafkajs ${consumerKjsMessage}`); | ||
| console.log(`Consumer rates msg/s (eachMessage): confluent ${consumerConfluentMessageRate}, kafkajs ${consumerKjsMessageRate}`); | ||
| 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}`); | ||
| if (concurrentRun) { | ||
| console.log(`Consumer average E2E latency T0-T1 (eachMessage): confluent ${consumerConfluentMessageAvgLatencyT0T1}, kafkajs ${consumerKjsMessageAvgLatencyT0T1}`); | ||
| console.log(`Consumer max E2E latency T0-T1 (eachMessage): confluent ${consumerConfluentMessageMaxLatencyT0T1}, kafkajs ${consumerKjsMessageMaxLatencyT0T1}`); | ||
| if (produceToSecondTopic) { | ||
| console.log(`Consumer average E2E latency T0-T2 (eachMessage): confluent ${consumerConfluentMessageAvgLatencyT0T2}, kafkajs ${consumerKjsMessageAvgLatencyT0T2}`); | ||
| console.log(`Consumer max E2E latency T0-T2 (eachMessage): confluent ${consumerConfluentMessageMaxLatencyT0T2}, kafkajs ${consumerKjsMessageMaxLatencyT0T2}`); | ||
| } | ||
| 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}`); | ||
| console.log(`Consumer rates msg/s (eachBatch): confluent ${consumerConfluentBatchRate}, kafkajs ${consumerKjsBatchRate}`); | ||
| console.log(`Consumption time (eachBatch): confluent ${consumerConfluentBatchTime}, kafkajs ${consumerKjsBatchTime}`); | ||
| 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}`); | ||
| if (concurrentRun) { | ||
| console.log(`Consumer average E2E latency T0-T1 (eachBatch): confluent ${consumerConfluentBatchAvgLatencyT0T1}, kafkajs ${consumerKjsBatchAvgLatencyT0T1}`); | ||
| console.log(`Consumer max E2E latency T0-T1 (eachBatch): confluent ${consumerConfluentBatchMaxLatencyT0T1}, kafkajs ${consumerKjsBatchMaxLatencyT0T1}`); | ||
| if (produceToSecondTopic) { | ||
| console.log(`Consumer average E2E latency T0-T2 (eachBatch): confluent ${consumerConfluentBatchAvgLatencyT0T2}, kafkajs ${consumerKjsBatchAvgLatencyT0T2}`); | ||
| console.log(`Consumer max E2E latency T0-T2 (eachBatch): confluent ${consumerConfluentBatchMaxLatencyT0T2}, kafkajs ${consumerKjsBatchMaxLatencyT0T2}`); | ||
| } | ||
| console.log(`Average eachBatch lag: confluent ${consumerConfluentBatchAverageLag}, kafkajs ${consumerKjsBatchAverageLag}`); | ||
| console.log(`Max eachBatch lag: confluent ${consumerConfluentBatchMaxLag}, kafkajs ${consumerKjsBatchMaxLag}`); | ||
| 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}`); | ||
|
|
@@ -321,7 +345,7 @@ | |
| if (consumerModeAll || consumerModeEachMessage) { | ||
| if (belowThreshold(consumerConfluentMessage, consumerKjsMessage, maxPerformanceDifference)) { | ||
| console.log(`Consumer rates MB/s (eachMessage) differ by more than 30%: confluent ${consumerConfluentMessage}, kafkajs ${consumerKjsMessage}`); | ||
| // FIXME: improve consumer performance at least to KafkaJS level | ||
| errcode = 0; | ||
| } | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -89,7 +89,7 @@ function getPackageVersion(tag, branch) { | |
|
|
||
| // publish with a -devel suffix for EA and RC releases. | ||
| if (tag.prerelease.length > 0) { | ||
| baseVersion += '-' + tag.prerelease.join('.'); | ||
| } | ||
|
|
||
| console.log(`Package version is "${baseVersion}"`); | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: Let's make it 1.7.0