Skip to content

Commit ed0fb03

Browse files
emasabemasab
committed
Add broker lag monitoring
1 parent 9a5db79 commit ed0fb03

File tree

4 files changed

+149
-4
lines changed

4 files changed

+149
-4
lines changed

‎ci/tests/run_perf_test.js‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -79,15 +79,24 @@ async function main() {
7979

8080
if (concurrentRun) {
8181
console.log(`Running ${modeLabel} Producer/Consumer test (concurrently)...`);
82+
const TERMINATE_TIMEOUT_MS = process.env.TERMINATE_TIMEOUT_MS ? +process.env.TERMINATE_TIMEOUT_MS : 600000;
83+
// Wait 2s more to see if all lag is caught up
84+
const TERMINATE_TIMEOUT_MS_CONSUMERS = TERMINATE_TIMEOUT_MS + 2000;
8285

8386
await runCommand(`MODE=${mode} node performance-consolidated.js --create-topics`);
8487
const allPromises = [];
8588
allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} node performance-consolidated.js --producer`));
8689
if (consumerModeAll || consumerModeEachMessage) {
87-
allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} GROUPID_MESSAGE=${groupIdEachMessage} node performance-consolidated.js --consumer-each-message ${produceToSecondTopicParam}`));
90+
allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MESSAGE=${groupIdEachMessage} node performance-consolidated.js --consumer-each-message ${produceToSecondTopicParam}`));
8891
}
8992
if (consumerModeAll || consumerModeEachBatch) {
90-
allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --consumer-each-batch ${produceToSecondTopicParam}`));
93+
allPromises.push(runCommand(`MODE=${mode} MESSAGE_COUNT=${messageCount} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_BATCH=${groupIdEachBatch} node performance-consolidated.js --consumer-each-batch ${produceToSecondTopicParam}`));
94+
}
95+
if (consumerModeAll || consumerModeEachMessage) {
96+
allPromises.push(runCommand(`MODE=${mode} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MONITOR=${groupIdEachMessage} node performance-consolidated.js --monitor-lag`));
97+
}
98+
if (consumerModeAll || consumerModeEachBatch) {
99+
allPromises.push(runCommand(`MODE=${mode} TERMINATE_TIMEOUT_MS=${TERMINATE_TIMEOUT_MS_CONSUMERS} GROUPID_MONITOR=${groupIdEachBatch} node performance-consolidated.js --monitor-lag`));
91100
}
92101
const results = await Promise.allSettled(allPromises);
93102
return results.map(r => r.status === 'fulfilled' ? r.value : '').join('\n');
@@ -123,6 +132,10 @@ async function main() {
123132
let consumerConfluentTime;
124133
let consumerConfluentMessageAverageRSS;
125134
let consumerConfluentMessageMaxRSS;
135+
let consumerConfluentMessageAverageBrokerLag;
136+
let consumerConfluentMessageMaxBrokerLag;
137+
let consumerConfluentMessageTotalLagMeasurements;
138+
126139
let consumerConfluentBatch;
127140
let consumerConfluentBatchRate;
128141
let consumerConfluentBatchAvgLatency;
@@ -133,6 +146,9 @@ async function main() {
133146
let consumerConfluentBatchAverageSize;
134147
let consumerConfluentBatchAverageRSS;
135148
let consumerConfluentBatchMaxRSS;
149+
let consumerConfluentBatchAverageBrokerLag;
150+
let consumerConfluentBatchMaxBrokerLag;
151+
let consumerConfluentBatchTotalLagMeasurements;
136152

137153
const producerConfluent = extractValue(outputConfluentProducerConsumer, '=== Producer Rate:');
138154
const producerConfluentAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Average producer RSS KB:');
@@ -145,6 +161,9 @@ async function main() {
145161
consumerConfluentTime = extractValue(outputConfluentProducerConsumer, '=== Consumption time (eachMessage):');
146162
consumerConfluentMessageAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Average consumer-each-message RSS KB:');
147163
consumerConfluentMessageMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max consumer-each-message RSS KB:');
164+
consumerConfluentMessageAverageBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Average broker lag (${groupIdEachMessageConfluent}):`);
165+
consumerConfluentMessageMaxBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Max broker lag (${groupIdEachMessageConfluent}):`);
166+
consumerConfluentMessageTotalLagMeasurements = extractValue(outputConfluentProducerConsumer, `=== Total broker lag measurements (${groupIdEachMessageConfluent}):`);
148167
}
149168
if (consumerModeAll || consumerModeEachBatch) {
150169
consumerConfluentBatch = extractValue(outputConfluentProducerConsumer, '=== Consumer Rate MB/s (eachBatch):');
@@ -157,6 +176,9 @@ async function main() {
157176
consumerConfluentBatchAverageSize = extractValue(outputConfluentProducerConsumer, '=== Average eachBatch size:');
158177
consumerConfluentBatchAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Average consumer-each-batch RSS KB:');
159178
consumerConfluentBatchMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max consumer-each-batch RSS KB:');
179+
consumerConfluentBatchAverageBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Average broker lag (${groupIdEachBatchConfluent}):`);
180+
consumerConfluentBatchMaxBrokerLag = extractValue(outputConfluentProducerConsumer, `=== Max broker lag (${groupIdEachBatchConfluent}):`);
181+
consumerConfluentBatchTotalLagMeasurements = extractValue(outputConfluentProducerConsumer, `=== Total broker lag measurements (${groupIdEachBatchConfluent}):`);
160182
}
161183
const consumerConfluentAverageRSS = extractValue(outputConfluentProducerConsumer, '=== Max Average RSS across tests:');
162184
const consumerConfluentMaxRSS = extractValue(outputConfluentProducerConsumer, '=== Max RSS across tests:');
@@ -172,6 +194,10 @@ async function main() {
172194
let consumerKjsTime;
173195
let consumerKjsMessageAverageRSS;
174196
let consumerKjsMessageMaxRSS;
197+
let consumerKjsMessageAverageBrokerLag;
198+
let consumerKjsMessageMaxBrokerLag;
199+
let consumerKjsMessageTotalLagMeasurements;
200+
175201
let consumerKjsBatch;
176202
let consumerKjsBatchRate;
177203
let consumerKjsBatchAvgLatency;
@@ -182,6 +208,9 @@ async function main() {
182208
let consumerKjsBatchAverageSize;
183209
let consumerKjsBatchAverageRSS;
184210
let consumerKjsBatchMaxRSS;
211+
let consumerKjsBatchAverageBrokerLag;
212+
let consumerKjsBatchMaxBrokerLag;
213+
let consumerKjsBatchTotalLagMeasurements;
185214

186215
const producerKjs = extractValue(outputKjsProducerConsumer, '=== Producer Rate:');
187216
const producerKjsAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average producer RSS KB:');
@@ -193,6 +222,9 @@ async function main() {
193222
consumerKjsMessageMaxLatency = extractValue(outputKjsProducerConsumer, '=== Consumer max E2E latency (eachMessage):');
194223
consumerKjsMessageAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average consumer-each-message RSS KB:');
195224
consumerKjsMessageMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max consumer-each-message RSS KB:');
225+
consumerKjsMessageAverageBrokerLag = extractValue(outputKjsProducerConsumer, `=== Average broker lag (${groupIdEachMessageKafkaJS}):`);
226+
consumerKjsMessageMaxBrokerLag = extractValue(outputKjsProducerConsumer, `=== Max broker lag (${groupIdEachMessageKafkaJS}):`);
227+
consumerKjsMessageTotalLagMeasurements = extractValue(outputKjsProducerConsumer, `=== Total broker lag measurements (${groupIdEachMessageKafkaJS}):`);
196228
}
197229
if (consumerModeAll || consumerModeEachBatch) {
198230
consumerKjsTime = extractValue(outputKjsProducerConsumer, '=== Consumption time (eachMessage):');
@@ -206,6 +238,9 @@ async function main() {
206238
consumerKjsBatchAverageSize = extractValue(outputKjsProducerConsumer, '=== Average eachBatch size:');
207239
consumerKjsBatchAverageRSS = extractValue(outputKjsProducerConsumer, '=== Average consumer-each-batch RSS KB:');
208240
consumerKjsBatchMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max consumer-each-batch RSS KB:');
241+
consumerKjsBatchAverageBrokerLag = extractValue(outputKjsProducerConsumer, `=== Average broker lag (${groupIdEachBatchKafkaJS}):`);
242+
consumerKjsBatchMaxBrokerLag = extractValue(outputKjsProducerConsumer, `=== Max broker lag (${groupIdEachBatchKafkaJS}):`);
243+
consumerKjsBatchTotalLagMeasurements = extractValue(outputKjsProducerConsumer, `=== Total broker lag measurements (${groupIdEachBatchKafkaJS}):`);
209244
}
210245
const consumerKjsAverageRSS = extractValue(outputKjsProducerConsumer, '=== Max Average RSS across tests:');
211246
const consumerKjsMaxRSS = extractValue(outputKjsProducerConsumer, '=== Max RSS across tests:');
@@ -225,6 +260,9 @@ async function main() {
225260
console.log(`Consumption time (eachMessage): confluent ${consumerConfluentTime}, kafkajs ${consumerKjsTime}`);
226261
console.log(`Average RSS (eachMessage): confluent ${consumerConfluentMessageAverageRSS}, kafkajs ${consumerKjsMessageAverageRSS}`);
227262
console.log(`Max RSS (eachMessage): confluent ${consumerConfluentMessageMaxRSS}, kafkajs ${consumerKjsMessageMaxRSS}`);
263+
console.log(`Average broker lag (eachMessage): confluent ${consumerConfluentMessageAverageBrokerLag}, kafkajs ${consumerKjsMessageAverageBrokerLag}`);
264+
console.log(`Max broker lag (eachMessage): confluent ${consumerConfluentMessageMaxBrokerLag}, kafkajs ${consumerKjsMessageMaxBrokerLag}`);
265+
console.log(`Total broker lag measurements (eachMessage): confluent ${consumerConfluentMessageTotalLagMeasurements}, kafkajs ${consumerKjsMessageTotalLagMeasurements}`);
228266
}
229267
if (consumerModeAll || consumerModeEachBatch) {
230268
console.log(`Consumer rates MB/s (eachBatch): confluent ${consumerConfluentBatch}, kafkajs ${consumerKjsBatch}`);
@@ -237,6 +275,9 @@ async function main() {
237275
console.log(`Average eachBatch size: confluent ${consumerConfluentBatchAverageSize}, kafkajs ${consumerKjsBatchAverageSize}`);
238276
console.log(`Average RSS (eachBatch): confluent ${consumerConfluentBatchAverageRSS}, kafkajs ${consumerKjsBatchAverageRSS}`);
239277
console.log(`Max RSS (eachBatch): confluent ${consumerConfluentBatchMaxRSS}, kafkajs ${consumerKjsBatchMaxRSS}`);
278+
console.log(`Average broker lag (eachBatch): confluent ${consumerConfluentBatchAverageBrokerLag}, kafkajs ${consumerKjsBatchAverageBrokerLag}`);
279+
console.log(`Max broker lag (eachBatch): confluent ${consumerConfluentBatchMaxBrokerLag}, kafkajs ${consumerKjsBatchMaxBrokerLag}`);
280+
console.log(`Total broker lag measurements (eachBatch): confluent ${consumerConfluentBatchTotalLagMeasurements}, kafkajs ${consumerKjsBatchTotalLagMeasurements}`);
240281
}
241282
if (!concurrentRun) {
242283
console.log(`Average RSS: confluent ${consumerConfluentAverageRSS}, kafkajs ${consumerKjsAverageRSS}`);

‎examples/performance/performance-consolidated.js‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,13 @@ const mode = process.env.MODE ? process.env.MODE : 'confluent';
33

44
let runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics, runProducerConsumerTogether;
55
if (mode === 'confluent') {
6-
({ runProducer, runConsumer, runConsumeTransformProduce, runCreateTopics, runProducerConsumerTogether } = require('./performance-primitives'));
6+
({ runProducer, runConsumer, runConsumeTransformProduce,
7+
runCreateTopics, runLagMonitoring,
8+
runProducerConsumerTogether } = require('./performance-primitives'));
79
} else {
810
({ runProducer, runConsumer, runConsumeTransformProduce, runProducerConsumerTogether } = require('./performance-primitives-kafkajs'));
911
/* createTopics is more reliable in CKJS */
10-
({ runCreateTopics } = require('./performance-primitives'));
12+
({ runCreateTopics, runLagMonitoring } = require('./performance-primitives'));
1113
}
1214

1315
const brokers = process.env.KAFKA_BROKERS || 'localhost:9092';
@@ -58,6 +60,7 @@ function logParameters(parameters) {
5860
const produceConsumeLatency = process.argv.includes('--latency');
5961
const all = process.argv.includes('--all');
6062
const createTopics = process.argv.includes('--create-topics');
63+
const monitorLag = process.argv.includes('--monitor-lag');
6164
let maxAverageRSSKB, maxMaxRSSKB;
6265
const stats = {};
6366

@@ -106,6 +109,21 @@ function logParameters(parameters) {
106109
await runCreateTopics(parameters, topic, topic2, numPartitions);
107110
}
108111

112+
if (monitorLag) {
113+
console.log("=== Starting Lag Monitoring:");
114+
logParameters(parameters);
115+
console.log(` Topic: ${topic}`);
116+
const {
117+
averageLag,
118+
maxLag,
119+
totalMeasurements
120+
} = await runLagMonitoring(parameters, topic);
121+
const monitoredGroupId = process.env.GROUPID_MONITOR;
122+
console.log(`=== Average broker lag (${monitoredGroupId}): `, averageLag);
123+
console.log(`=== Max broker lag (${monitoredGroupId}): `, maxLag);
124+
console.log(`=== Total broker lag measurements (${monitoredGroupId}): `, totalMeasurements);
125+
}
126+
109127
if (producer || all) {
110128
console.log("=== Running Basic Producer Performance Test:")
111129
logParameters(parameters);

‎examples/performance/performance-primitives-common.js‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -306,9 +306,86 @@ async function runProducer(producer, topic, batchSize, warmupMessages, totalMess
306306
return rate;
307307
}
308308

309+
async function runLagMonitoring(admin, topic) {
310+
const handlers = installHandlers();
311+
let groupId = process.env.GROUPID_MONITOR;
312+
if (!groupId) {
313+
throw new Error("GROUPID_MONITOR environment variable not set");
314+
}
315+
316+
await admin.connect();
317+
318+
const fetchTotalLag = async () => {
319+
const partitionCompleteLag = {};
320+
const partitionHWM = {};
321+
const partitionLag = {};
322+
let totalLag = 0n;
323+
const operations = [
324+
admin.fetchTopicOffsets(topic, { timeout: 30000 }),
325+
admin.fetchOffsets({ groupId, topics: [topic], timeout: 30000 }),
326+
]
327+
let [topicOffsets, groupOffsets] = await Promise.all(operations);
328+
groupOffsets = groupOffsets[0];
329+
330+
for (const partitionOffset of topicOffsets) {
331+
partitionHWM[partitionOffset.partition] = BigInt(partitionOffset.high);
332+
partitionCompleteLag[partitionOffset.partition] = BigInt(partitionOffset.high) - BigInt(partitionOffset.low);
333+
}
334+
335+
if (groupOffsets && groupOffsets.partitions) {
336+
for (const partitionOffset of groupOffsets.partitions) {
337+
const partition = partitionOffset.partition;
338+
const hwm = partitionHWM[partition];
339+
if (hwm && partitionOffset.offset && hwm >= BigInt(partitionOffset.offset)) {
340+
const currentLag = hwm - BigInt(partitionOffset.offset);
341+
partitionLag[partition] = currentLag;
342+
totalLag += currentLag;
343+
}
344+
}
345+
} else {
346+
throw new Error(`No offsets found for group ${groupId} on topic ${topic}`);
347+
}
348+
for (const partition of Object.keys(partitionHWM)) {
349+
if (partitionLag[partition] === undefined) {
350+
const currentLag = partitionCompleteLag[partition];
351+
partitionLag[partition] = currentLag;
352+
totalLag += currentLag;
353+
}
354+
}
355+
return totalLag;
356+
}
357+
358+
let totalAverageLag = 0n;
359+
let maxLag = 0n;
360+
let totalMeasurements = 0;
361+
362+
while (!handlers.terminationRequested) {
363+
try {
364+
const lag = await fetchTotalLag();
365+
totalAverageLag += lag;
366+
maxLag = lag > maxLag ? lag : maxLag;
367+
totalMeasurements++;
368+
} catch (e) {
369+
console.error(`Error fetching lag: ${e}`);
370+
}
371+
await new Promise(resolve => setTimeout(resolve, 100));
372+
}
373+
const averageLag = totalMeasurements > 0 ? (Number(totalAverageLag) / totalMeasurements) : NaN;
374+
maxLag = Number(maxLag);
375+
376+
await admin.disconnect();
377+
removeHandlers(handlers);
378+
return {
379+
averageLag,
380+
maxLag,
381+
totalMeasurements
382+
};
383+
}
384+
309385
module.exports = {
310386
runConsumer,
311387
runProducer,
388+
runLagMonitoring,
312389
genericProduceToTopic,
313390
getAutoCommit,
314391
};

‎examples/performance/performance-primitives.js‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ const { hrtime } = require('process');
44
const {
55
runConsumer: runConsumerCommon,
66
runProducer: runProducerCommon,
7+
runLagMonitoring: runLagMonitoringCommon,
78
genericProduceToTopic,
89
getAutoCommit,
910
} = require('./performance-primitives-common');
@@ -14,6 +15,7 @@ module.exports = {
1415
runConsumeTransformProduce,
1516
runCreateTopics,
1617
runProducerConsumerTogether,
18+
runLagMonitoring,
1719
};
1820

1921
function baseConfiguration(parameters) {
@@ -65,6 +67,13 @@ async function runCreateTopics(parameters, topic, topic2, numPartitions) {
6567
await admin.disconnect();
6668
}
6769

70+
function runLagMonitoring(parameters, topic) {
71+
const kafka = new Kafka(baseConfiguration(parameters));
72+
const admin = kafka.admin();
73+
74+
return runLagMonitoringCommon(admin, topic);
75+
}
76+
6877
class CompatibleProducer {
6978
constructor(producer) {
7079
this.producer = producer;

0 commit comments

Comments
 (0)