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

Message compression

Importing @firefly-exchange/kafka-client registers compression codecs into kafkajs.CompressionCodecs as a side effect, so consumers and producers can read and write messages in the org's standard compression algorithms without extra setup:

| Algorithm | Source | Notes | | --------- | ------------------- | ------------------------------------------------ | | gzip | kafkajs built-in | Always available. | | lz4 | kafkajs-lz4 | Unblocks turning on compression for the matching-engine events producer — see note below. | | snappy | kafkajs-snappy | Available as a freebie; not currently in heavy use. | | zstd | not registered | Pending — see "zstd" note below. |

Why lz4 specifically. In pro-mono, the matching-engine-v3 events producer that publishes pro-trading-engine.events (rust/service/matching-engine-v3/src/boundary/kafka/producer.rs:153) runs uncompressed today, with the comment on line 176:

// If our TS clients ever upgrade, we can enable compression here for even better performance.

That comment is exactly what this codec registration resolves. Once consumers of this wrapper have picked up ^4.4.0, the events producer can set compression.type=lz4 and TS consumers will decode it transparently.

zstd. The @kafkajs/zstd codec wraps the native cppzst module, which requires install scripts to build the native binary. Services that run with enableScripts: false (supply-chain hardening) cannot install it. A separate follow-up will add zstd once a pure-JS/WASM zstd codec is in place or an agreed install-scripts exception path exists.

lz4 install-scripts caveat. [email protected] depends on the lz4 Node package, which builds a native binary via node-gyp at install time. If your service runs with enableScripts: false, you'll need an explicit yarnrc exception for lz4 (and its build-tooling deps nan, node-gyp). Alternatively, the [email protected] line uses pure-WASM lz4-asm and needs no install scripts but is currently a beta release.

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.