Skip to content

Commit 8ce7cfb

Browse files
emasabemasab
authored
Performance measurement improvements
CKJS performance improved through fetch.queue.backoff.ms
1 parent 0d8400b commit 8ce7cfb

File tree

3 files changed

+115
-126
lines changed

3 files changed

+115
-126
lines changed

‎examples/performance/performance-consolidated.js‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ const ctpConcurrency = process.env.CONSUME_TRANSFORM_PRODUCE_CONCURRENCY ? +proc
6464
console.log(` Message Count: ${messageCount}`);
6565
// Seed the topic with messages
6666
await runProducer(brokers, topic, batchSize, warmupMessages, messageCount, messageSize, compression);
67-
const ctpRate = await runConsumeTransformProduce(brokers, topic, topic2, messageCount, messageProcessTimeMs, ctpConcurrency);
67+
const ctpRate = await runConsumeTransformProduce(brokers, topic, topic2, warmupMessages, messageCount, messageProcessTimeMs, ctpConcurrency);
6868
console.log("=== Consume-Transform-Produce Rate: ", ctpRate);
6969
}
7070

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

Lines changed: 56 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -118,44 +118,38 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
118118
await consumer.subscribe({ topic, fromBeginning: true });
119119

120120
let messagesReceived = 0;
121+
let messagesMeasured = 0;
121122
let totalMessageSize = 0;
122123
let startTime;
123124
let rate;
125+
const skippedMessages = 100;
126+
127+
console.log("Starting consumer.");
128+
124129
consumer.run({
125130
autoCommit: false,
126131
eachMessage: async ({ topic, partition, message }) => {
127132
messagesReceived++;
128-
totalMessageSize += message.value.length;
129-
if (messagesReceived === 1) {
130-
consumer.pause([{ topic }]);
131-
} else if (messagesReceived === 2) {
132-
startTime = hrtime();
133-
} else if (messagesReceived === totalMessageCnt) {
134-
let elapsed = hrtime(startTime);
135-
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
136-
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
137-
console.log(`Recvd ${messagesReceived} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
138-
consumer.pause([{ topic }]);
139-
// } else if (messagesReceived % 100 == 0) {
140-
// console.log(`Recvd ${messagesReceived} messages, ${totalMessageSize} bytes`);
133+
134+
if (messagesReceived >= skippedMessages) {
135+
messagesMeasured++;
136+
totalMessageSize += message.value.length;
137+
138+
if (messagesReceived === skippedMessages) {
139+
startTime = hrtime();
140+
} else if (messagesMeasured === totalMessageCnt) {
141+
let elapsed = hrtime(startTime);
142+
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
143+
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
144+
console.log(`Recvd ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
145+
consumer.pause([{ topic }]);
146+
}
141147
}
142148
}
143149
});
144150

145-
// Wait until the first message is received
146-
await new Promise((resolve) => {
147-
let interval = setInterval(() => {
148-
if (messagesReceived > 0) {
149-
clearInterval(interval);
150-
resolve();
151-
}
152-
}, 100);
153-
});
154-
155-
console.log("Starting consumer.")
156-
157151
totalMessageSize = 0;
158-
consumer.resume([{ topic }]);
152+
159153
await new Promise((resolve) => {
160154
let interval = setInterval(() => {
161155
if (messagesReceived >= totalMessageCnt) {
@@ -169,7 +163,7 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
169163
return rate;
170164
}
171165

172-
async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
166+
async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
173167
const kafka = new Kafka({
174168
clientId: 'kafka-test-performance',
175169
brokers: brokers.split(','),
@@ -185,56 +179,56 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, t
185179
await consumer.subscribe({ topic: consumeTopic, fromBeginning: true });
186180

187181
let messagesReceived = 0;
182+
let messagesMeasured = 0;
188183
let totalMessageSize = 0;
189184
let startTime;
190185
let rate;
186+
const skippedMessages = warmupMessages;
187+
188+
console.log("Starting consume-transform-produce.");
189+
191190
consumer.run({
192191
autoCommit: false,
193192
partitionsConsumedConcurrently: ctpConcurrency,
194193
eachMessage: async ({ topic, partition, message }) => {
195-
/* Simulate message processing for messageProcessTimeMs */
196-
if (messageProcessTimeMs > 0) {
197-
await new Promise((resolve) => setTimeout(resolve, messageProcessTimeMs));
198-
}
199-
await producer.send({
200-
topic: produceTopic,
201-
messages: [{ value: message.value }],
202-
})
203194
messagesReceived++;
204-
totalMessageSize += message.value.length;
205-
if (messagesReceived === 1) {
206-
consumer.pause([{ topic }]);
207-
} else if (messagesReceived === 2) {
208-
startTime = hrtime();
209-
} else if (messagesReceived === totalMessageCnt) {
210-
let elapsed = hrtime(startTime);
211-
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
212-
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
213-
console.log(`Recvd, transformed and sent ${messagesReceived} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
214-
consumer.pause([{ topic }]);
215-
// } else if (messagesReceived % 1 == 0) {
216-
// console.log(`Recvd ${messagesReceived} messages, ${totalMessageSize} bytes`);
217-
}
218-
}
219-
});
220195

221-
// Wait until the first message is received
222-
await new Promise((resolve) => {
223-
let interval = setInterval(() => {
224-
if (messagesReceived > 0) {
225-
clearInterval(interval);
226-
resolve();
196+
if (messagesReceived >= skippedMessages) {
197+
messagesMeasured++;
198+
totalMessageSize += message.value.length;
199+
200+
if (messagesReceived === skippedMessages)
201+
startTime = hrtime();
202+
203+
/* Simulate message processing for messageProcessTimeMs */
204+
if (messageProcessTimeMs > 0) {
205+
await new Promise((resolve) => setTimeout(resolve, messageProcessTimeMs));
206+
}
207+
await producer.send({
208+
topic: produceTopic,
209+
messages: [{ value: message.value }],
210+
})
211+
212+
if (messagesMeasured === totalMessageCnt) {
213+
let elapsed = hrtime(startTime);
214+
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
215+
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
216+
console.log(`Recvd, transformed and sent ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
217+
consumer.pause([{ topic }]);
218+
}
219+
} else {
220+
await producer.send({
221+
topic: produceTopic,
222+
messages: [{ value: message.value }],
223+
})
227224
}
228-
}, 100);
225+
}
229226
});
230227

231-
console.log("Starting consume-transform-produce.")
232-
233228
totalMessageSize = 0;
234-
consumer.resume([{ topic: consumeTopic }]);
235229
await new Promise((resolve) => {
236230
let interval = setInterval(() => {
237-
if (messagesReceived >= totalMessageCnt) {
231+
if (messagesMeasured >= totalMessageCnt) {
238232
clearInterval(interval);
239233
resolve();
240234
}

‎examples/performance/performance-primitives.js‎

Lines changed: 58 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -120,51 +120,45 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
120120
'group.id': 'test-group' + Math.random(),
121121
'enable.auto.commit': false,
122122
'auto.offset.reset': 'earliest',
123+
'fetch.queue.backoff.ms': '100',
123124
});
124125
await consumer.connect();
125126
await consumer.subscribe({ topic });
126127

127128
let messagesReceived = 0;
129+
let messagesMeasured = 0;
128130
let totalMessageSize = 0;
129131
let startTime;
130132
let rate;
133+
const skippedMessages = 100;
134+
135+
console.log("Starting consumer.");
136+
131137
consumer.run({
132138
eachMessage: async ({ topic, partition, message }) => {
133139
messagesReceived++;
134-
totalMessageSize += message.value.length;
135-
if (messagesReceived === 1) {
136-
consumer.pause([{ topic }]);
137-
} else if (messagesReceived === 2) {
138-
startTime = hrtime();
139-
} else if (messagesReceived === totalMessageCnt) {
140-
let elapsed = hrtime(startTime);
141-
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
142-
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
143-
console.log(`Recvd ${messagesReceived} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
144-
consumer.pause([{ topic }]);
145-
// } else if (messagesReceived % 100 == 0) {
146-
// console.log(`Recvd ${messagesReceived} messages, ${totalMessageSize} bytes`);
147-
}
148-
}
149-
});
140+
141+
if (messagesReceived >= skippedMessages) {
142+
messagesMeasured++;
143+
totalMessageSize += message.value.length;
150144

151-
// Wait until the first message is received
152-
await new Promise((resolve) => {
153-
let interval = setInterval(() => {
154-
if (messagesReceived > 0) {
155-
clearInterval(interval);
156-
resolve();
145+
if (messagesReceived === skippedMessages) {
146+
startTime = hrtime();
147+
} else if (messagesMeasured === totalMessageCnt) {
148+
let elapsed = hrtime(startTime);
149+
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
150+
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
151+
console.log(`Recvd ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
152+
consumer.pause([{ topic }]);
153+
}
157154
}
158-
}, 100);
155+
}
159156
});
160157

161-
console.log("Starting consumer.")
162-
163158
totalMessageSize = 0;
164-
consumer.resume([{ topic }]);
165159
await new Promise((resolve) => {
166160
let interval = setInterval(() => {
167-
if (messagesReceived >= totalMessageCnt) {
161+
if (messagesMeasured >= totalMessageCnt) {
168162
clearInterval(interval);
169163
resolve();
170164
}
@@ -175,7 +169,8 @@ async function runConsumer(brokers, topic, totalMessageCnt) {
175169
return rate;
176170
}
177171

178-
async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
172+
async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, warmupMessages, totalMessageCnt, messageProcessTimeMs, ctpConcurrency) {
173+
console.log("here");
179174
const kafka = new Kafka({
180175
'client.id': 'kafka-test-performance',
181176
'metadata.broker.list': brokers,
@@ -205,55 +200,55 @@ async function runConsumeTransformProduce(brokers, consumeTopic, produceTopic, t
205200
await consumer.subscribe({ topic: consumeTopic });
206201

207202
let messagesReceived = 0;
203+
let messagesMeasured = 0;
208204
let totalMessageSize = 0;
209205
let startTime;
210206
let rate;
207+
const skippedMessages = warmupMessages;
208+
209+
console.log("Starting consume-transform-produce.");
210+
211211
consumer.run({
212212
partitionsConsumedConcurrently: ctpConcurrency,
213213
eachMessage: async ({ topic, partition, message }) => {
214-
/* Simulate message processing for messageProcessTimeMs */
215-
if (messageProcessTimeMs > 0) {
216-
await new Promise((resolve) => setTimeout(resolve, messageProcessTimeMs));
217-
}
218-
await producer.send({
219-
topic: produceTopic,
220-
messages: [{ value: message.value }],
221-
})
222214
messagesReceived++;
223-
totalMessageSize += message.value.length;
224-
if (messagesReceived === 1) {
225-
consumer.pause([{ topic }]);
226-
} else if (messagesReceived === 2) {
227-
startTime = hrtime();
228-
} else if (messagesReceived === totalMessageCnt) {
229-
let elapsed = hrtime(startTime);
230-
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
231-
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
232-
console.log(`Recvd, transformed and sent ${messagesReceived} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
233-
consumer.pause([{ topic }]);
234-
// } else if (messagesReceived % 1 == 0) {
235-
// console.log(`Recvd ${messagesReceived} messages, ${totalMessageSize} bytes`);
236-
}
237-
}
238-
});
239215

240-
// Wait until the first message is received
241-
await new Promise((resolve) => {
242-
let interval = setInterval(() => {
243-
if (messagesReceived > 0) {
244-
clearInterval(interval);
245-
resolve();
216+
if (messagesReceived >= skippedMessages) {
217+
messagesMeasured++;
218+
totalMessageSize += message.value.length;
219+
220+
if (messagesReceived === skippedMessages)
221+
startTime = hrtime();
222+
223+
/* Simulate message processing for messageProcessTimeMs */
224+
if (messageProcessTimeMs > 0) {
225+
await new Promise((resolve) => setTimeout(resolve, messageProcessTimeMs));
226+
}
227+
await producer.send({
228+
topic: produceTopic,
229+
messages: [{ value: message.value }],
230+
})
231+
232+
if (messagesMeasured === totalMessageCnt) {
233+
let elapsed = hrtime(startTime);
234+
let durationNanos = elapsed[0] * 1e9 + elapsed[1];
235+
rate = (totalMessageSize / durationNanos) * 1e9 / (1024 * 1024); /* MB/s */
236+
console.log(`Recvd, transformed and sent ${messagesMeasured} messages, ${totalMessageSize} bytes; rate is ${rate} MB/s`);
237+
consumer.pause([{ topic }]);
238+
}
239+
} else {
240+
await producer.send({
241+
topic: produceTopic,
242+
messages: [{ value: message.value }],
243+
})
246244
}
247-
}, 100);
245+
}
248246
});
249247

250-
console.log("Starting consume-transform-produce.")
251-
252248
totalMessageSize = 0;
253-
consumer.resume([{ topic: consumeTopic }]);
254249
await new Promise((resolve) => {
255250
let interval = setInterval(() => {
256-
if (messagesReceived >= totalMessageCnt) {
251+
if (messagesMeasured >= totalMessageCnt) {
257252
clearInterval(interval);
258253
resolve();
259254
}

0 commit comments

Comments
 (0)