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 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

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.