Purpose and scope of this Document
This document provides an overview of consuming data from a Kafka topic using one of the Kafka clients in Node.js. It aims to guide readers through the configuration and execution of a Kafka consumer script, ensuring a clear understanding of the necessary steps involved in the process.
Pre-Read: Get started with different Kafka clients
Pre-Req Info:
- bootstrap_server = 'YOUR_BOOTSTRAP_SERVER' # from cluster settings
Note: For playground env you can always use pkc-zd997.eu-central-1.aws.confluent.cloud:9092
- api_key = 'YOUR_API_KEY' # from API key management, "API key" which is shared by KaaS Team
- api_secret = 'YOUR_API_SECRET' # from API key management, "API secret" which is shared by KaaS Team
- topic_name = 'YOUR_TOPIC_NAME' # your topic name configured as per naming standard
- groupId/clientId = 'GROUPID/CLIENTID' # It is mandatory to have a prefix as "sa-[yourserviceaccountname]-" for groupID/ClientID
Node.js Script
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49 | const fs = require("fs");
const { Kafka } = require("@confluentinc/kafka-javascript").KafkaJS;
// Read configuration from client.properties
function readConfig(fileName) {
const data = fs.readFileSync(fileName, "utf8").toString().split("\n");
return data.reduce((config, line) => {
const [key, value] = line.split("=");
if (key && value) {
config[key] = value;
}
return config;
}, {});
}
async function consume(topic, config) {
const consumer = new Kafka().consumer(config);
// Setup graceful shutdown
const disconnect = () => {
consumer.commitOffsets().finally(() => {
consumer.disconnect();
});
};
process.on("SIGTERM", disconnect);
process.on("SIGINT", disconnect);
// Connect the consumer to the broker
await consumer.connect();
// Subscribe to the topic
await consumer.subscribe({ topics: [topic] });
// Consume messages from the topic
consumer.run({
eachMessage: async ({ topic, partition, message }) => {
console.log(`Consumed message from topic ${topic}, partition ${partition}: key = ${message.key.toString()}, value = ${message.value.toString()}`);
},
});
}
// Main function to run the consumer
async function main() {
const config = readConfig("client.properties");
const topic = "YOUR_TOPIC_NAME"; // Replace with your actual topic
await consume(topic, config);
}
main().catch(console.error);
|