@firefly-exchange/kafka-client v4.1.9
kafka-client
The Kafka Client library is a tiny wrapper over kafka.js to produce and consume messages from topics.
Usage
Installation
Install kafka-client using yarn:
yarn add @firefly-exchange/kafka-client
Publishing a message
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 nodeslogger: pino.Logger
supply your customised pino logger function hereconnectionTimeout: number
(optional) connection timeout in ms, default to 10000ms = 10sclientId: 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 topictopic: string
name of topic to publish data tokey: string
(optional) partition keyacks: 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
Consuming a message
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.
9 days ago
2 months ago
5 months ago
10 months ago
9 months ago
9 months ago
7 months ago
8 months ago
10 months ago
10 months ago
9 months ago
10 months ago
11 months ago
11 months ago
11 months ago
11 months ago
11 months ago
11 months ago
11 months ago
11 months ago
11 months ago
11 months ago
12 months ago
12 months ago
1 year ago
1 year ago