You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
3.3 KiB
100 lines
3.3 KiB
const Kafka = require("../lib/index.js");
|
|
|
|
const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
|
|
|
const sendData = async (producer, totalMessages) => {
|
|
const topic = "node";
|
|
const msg = "dkfljaskldfajkldsjfklasdjfalk;dsjfkl;asjfdskl;fjda;lkfjsdklfsajlkfjdsklfajsklfjsklanklsalkjkljkasfak";
|
|
const buffer = Buffer.from(msg);
|
|
const key = "test";
|
|
for (let n = 0; n < totalMessages; ++n) {
|
|
let bufferIsFull = false;
|
|
do {
|
|
bufferIsFull = false;
|
|
try {
|
|
producer.produce(topic, -1, buffer, key, null, n);
|
|
}
|
|
catch (error) {
|
|
// Based on config, and messages, this will execute once
|
|
if (error.code === Kafka.CODES.ERRORS.ERR__QUEUE_FULL) {
|
|
producer.poll();
|
|
// The wait introduces 11-12 seconds of latency when dr_cb is true
|
|
const start = process.hrtime();
|
|
await wait(50);
|
|
const latency = process.hrtime(start);
|
|
console.info(`Wait took ${latency[0]} seconds`);
|
|
bufferIsFull = true;
|
|
} else {
|
|
throw error;
|
|
}
|
|
}
|
|
} while (bufferIsFull);
|
|
}
|
|
console.log("Finished producing");
|
|
};
|
|
|
|
const verifyReports = async (reports, reportsComplete, totalMessages) => {
|
|
const reportsTimeout = new Promise((resolve, reject) => {
|
|
setTimeout(() => {
|
|
reject("Delivery report timed out");
|
|
}, 10000);
|
|
});
|
|
await Promise.race([reportsComplete, reportsTimeout]);
|
|
await wait(500); // wait for some more delivery reports.
|
|
if (reports.length === totalMessages) {
|
|
console.log("Reports count match");
|
|
} else {
|
|
console.error("Reports count doesn't match");
|
|
return;
|
|
}
|
|
for(let n = 0; n < totalMessages; ++n) {
|
|
if(reports[n].opaque !== n) {
|
|
console.error("Expect message number does not match");
|
|
}
|
|
}
|
|
};
|
|
|
|
const run = async () => {
|
|
const reports = [];
|
|
const totalMessages = 1000100;
|
|
const producer = new Kafka.Producer({
|
|
"batch.num.messages": 50000,
|
|
"compression.codec": "lz4",
|
|
"delivery.report.only.error": false,
|
|
"dr_cb": true,
|
|
"metadata.broker.list": "localhost:9092",
|
|
"message.send.max.retries": 10000000,
|
|
"queue.buffering.max.kbytes": 2000000,
|
|
"queue.buffering.max.messages": 1000000,
|
|
"queue.buffering.max.ms": 0,
|
|
"socket.keepalive.enable": true,
|
|
}, {});
|
|
|
|
producer.setPollInterval(100);
|
|
producer.on("event.log", (obj) => console.log(obj));
|
|
const reportsComplete = new Promise((resolve) => {
|
|
producer.on("delivery-report", (err, report) => {
|
|
reports.push(report);
|
|
if(reports.length === totalMessages) {
|
|
resolve();
|
|
}
|
|
});
|
|
});
|
|
|
|
const readyPromise = new Promise((resolve) => {
|
|
producer.on("ready", async () => {
|
|
console.log("Producer is ready");
|
|
resolve();
|
|
});
|
|
producer.connect();
|
|
});
|
|
await readyPromise;
|
|
|
|
await sendData(producer, totalMessages);
|
|
await verifyReports(reports, reportsComplete, totalMessages);
|
|
process.exit(0);
|
|
};
|
|
|
|
run().catch((err) => {
|
|
console.error(err);
|
|
});
|