The Kafka Client library is a tiny wrapper over kafka.js to produce and consume messages from topics.
Install kafka-client using yarn:
yarn add @firefly-exchange/kafka-client
To publish a message first create a new instance of the KafkaProducer
class using its
static create
method. The method accepts the following parameters
-
bootstrapServers: string
a comma separated list of Kafka nodes -
logger: pino.Logger
supply your customised pino logger function here -
connectionTimeout: number
(optional) connection timeout in ms, default to 10000ms = 10s -
clientId: string
(optional) kafka client id, default to anonymous-client
code example:
kafkaProducer = KafkaProducer.create({
bootstrapServers: "localhost:9092,localhost:9093",
logger: pino()
});
Now to send message, call use send method. It accepts the following parameters
-
payload: any
payload to publish to topic -
topic: string
name of topic to publish data to -
key: string
(optional) partition key -
acks: int
ack mode, by default is set to -1 (In kafkajs -1 = ALL)
await kafkaProducer.send({
payload: payload,
topic: topic,
key: "order-hash-1", // optional
acks: -1 // optional
});
NOTE: To read about kafka ack please refer https://betterprogramming.pub/kafka-acks-explained-c0515b3b707e
Now to create a consumer to consume messages from a topic:
const consumer = await KafkaConsumer.create(messageBrokerURL);
await consumer.consumeTopic(topicName, callback);
await consumer.stop();
There are two concepts in Kafka autocommit and manual commit, autocommit means that once consumer have given you the messages it means its done, consumer dont care if your message is processed sucessfully on your end or not. However on manual commit, you specify after processing that yes you have received the message to kafka.
how to specify it here, Note that by default AUTOCOMMIT IS TRUE. For autocommit you can use this approach
kafkaConsumer = await KafkaConsumer.create({
bootstrapServers: broker,
logger: logger,
consumerGroupId: consumerGroup,
autoCommit: true
});
// The first argument is topic name, second argument is the fromBegining=true or false. the third argument is the callback func
await consumer.consumeTopic(topicName,true ,callback);
await consumer.stop();
for using manual commit, you can use this approach:
kafkaConsumer = await KafkaConsumer.create({
bootstrapServers: broker,
logger: logger,
consumerGroupId: consumerGroup,
autoCommit: false
});
// The first argument is topic name, second argument is the fromBegining=true or false. the third argument is the callback func
kafkaConsumer.consumeTopic(topic,true, async function (msg) {
// Do some processing
// once processing is done then commit the message
// commit the message
await kafkaConsumer.commit({
topic: topic,
partition: msg.partition,
offset: msg.offset
});
});
await consumer.stop();
The consumeTopic message run the handlerFunction with an await meaning that it wont process next message until one is processed. If you DO not want that then you can use this approach.
kafkaConsumer.consumeTopicAsync(topic,true, async function (msg) {
// Do some processing
// once processing is done then commit the message
// commit the message
await kafkaConsumer.commit({
topic: topic,
partition: msg.partition,
offset: msg.offset
});
});
await consumer.stop();
Please also note that in manual commit, the code will commit after your callback function is finished.