node-rdkafka

node-rdkafka - Node.js wrapper for Kafka C/C++ library

Copyright (c) 2016 Blizzard Entertainment.

https://github.com/blizzard/node-rdkafka

Looking for Collaborators!

I am looking for your help to make this project even better! If you're interested, check this out

Overview

The node-rdkafka library is a high-performance NodeJS client for Apache Kafka that wraps the native librdkafka library. All the complexity of balancing writes across partitions and managing (possibly ever-changing) brokers should be encapsulated in the library.

This library currently uses librdkafka version 2.12.0.

Reference Docs

To view the reference docs for the current version, go here

Contributing

For guidelines on contributing please see CONTRIBUTING.md

Code of Conduct

Play nice; Play fair.

Requirements

• Apache Kafka >=0.9
• Node.js >=16
• Linux/Mac
• Windows?! See below
• OpenSSL

Mac OS High Sierra / Mojave

OpenSSL has been upgraded in High Sierra and homebrew does not overwrite default system libraries. That means when building node-rdkafka, because you are using openssl, you need to tell the linker where to find it:

export CPPFLAGS=-I/usr/local/opt/openssl/include
export LDFLAGS=-L/usr/local/opt/openssl/lib

Then you can run npm install on your application to get it to build correctly.

NOTE: From the librdkafka docs

WARNING: Due to a bug in Apache Kafka 0.9.0.x, the ApiVersionRequest (as sent by the client when connecting to the broker) will be silently ignored by the broker causing the request to time out after 10 seconds. This causes client-broker connections to stall for 10 seconds during connection-setup before librdkafka falls back on the broker.version.fallback protocol features. The workaround is to explicitly configure api.version.request to false on clients communicating with <=0.9.0.x brokers.

Alpine

Using Alpine Linux? Check out the docs.

Windows

Windows build is not compiled from librdkafka source but it is rather linked against the appropriate version of NuGet librdkafka.redist static binary that gets downloaded from https://globalcdn.nuget.org/packages/librdkafka.redist.2.12.0.nupkg during installation. This download link can be changed using the environment variable NODE_RDKAFKA_NUGET_BASE_URL that defaults to https://globalcdn.nuget.org/packages/ when it's no set.

Requirements:

• node-gyp for Windows

Note: I still do not recommend using node-rdkafka in production on Windows. This feature was in high demand and is provided to help develop, but we do not test against Windows, and windows support may lag behind Linux/Mac support because those platforms are the ones used to develop this library. Contributors are welcome if any Windows issues are found :)

Tests

This project includes two types of unit tests in this project:

• end-to-end integration tests
• unit tests

You can run both types of tests by using Makefile. Doing so calls mocha in your locally installed node_modules directory.

• Before you run the tests, be sure to init and update the submodules:
  1. git submodule init
  2. git submodule update
• To run the unit tests, you can run make lint or make test.
• To run the integration tests, you must have a running Kafka installation available. By default, the test tries to connect to localhost:9092; however, you can supply the KAFKA_HOST environment variable to override this default behavior. Run make e2e.

Usage

You can install the node-rdkafka module like any other module:

npm install node-rdkafka

To use the module, you must require it.

const Kafka = require('node-rdkafka');

Configuration

You can pass many configuration options to librdkafka. A full list can be found in librdkafka's Configuration.md

Configuration keys that have the suffix _cb are designated as callbacks. Some of these keys are informational and you can choose to opt-in (for example, dr_cb). Others are callbacks designed to return a value, such as partitioner_cb.

Not all of these options are supported. The library will throw an error if the value you send in is invalid.

The library currently supports the following callbacks:

• partitioner_cb
• dr_cb or dr_msg_cb
• event_cb
• rebalance_cb (see Rebalancing)
• offset_commit_cb (see Commits)

Librdkafka Methods

This library includes two utility functions for detecting the status of your installation. Please try to include these when making issue reports where applicable.

You can get the features supported by your compile of librdkafka by reading the variable "features" on the root of the node-rdkafka object.

const Kafka = require('node-rdkafka');
console.log(Kafka.features);

// #=> [ 'gzip', 'snappy', 'ssl', 'sasl', 'regex', 'lz4' ]

You can also get the version of librdkafka

const Kafka = require('node-rdkafka');
console.log(Kafka.librdkafkaVersion);

// #=> 2.12.0

Sending Messages

A Producer sends messages to Kafka. The Producer constructor takes a configuration object, as shown in the following example:

const producer = new Kafka.Producer({
  'metadata.broker.list': 'kafka-host1:9092,kafka-host2:9092'
});

A Producer requires only metadata.broker.list (the Kafka brokers) to be created. The values in this list are separated by commas. For other configuration options, see the Configuration.md file described previously.

The following example illustrates a list with several librdkafka options set.

const producer = new Kafka.Producer({
  'client.id': 'kafka',
  'metadata.broker.list': 'localhost:9092',
  'compression.codec': 'gzip',
  'retry.backoff.ms': 200,
  'message.send.max.retries': 10,
  'socket.keepalive.enable': true,
  'queue.buffering.max.messages': 100000,
  'queue.buffering.max.ms': 1000,
  'batch.num.messages': 1000000,
  'dr_cb': true
});

Stream API

You can easily use the Producer as a writable stream immediately after creation (as shown in the following example):

…

If you do not want your code to crash when an error happens, ensure you have an error listener on the stream. Most errors are not necessarily fatal, but the ones that are will immediately destroy the stream. If you use autoClose, the stream will close itself at the first sign of a problem.

Standard API

The Standard API is more performant, particularly when handling high volumes of messages. However, it requires more manual setup to use. The following example illustrates its use:

…

To see the configuration options available to you, see the Configuration section.

Methods

Events

Some configuration properties that end in _cb indicate that an event should be generated for that option. You can either:

• provide a value of true and react to the event
• provide a callback function directly

The following example illustrates an event:

const producer = new Kafka.Producer({
  'client.id': 'my-client', // Specifies an identifier to use to help trace activity in Kafka
  'metadata.broker.list': 'localhost:9092', // Connect to a Kafka instance on localhost
  'dr_cb': true // Specifies that we want a delivery-report event to be generated
});

// Poll for events every 100 ms
producer.setPollInterval(100);

producer.on('delivery-report', (err, report) => {
  // Report of delivery statistics here:
  //
  console.log(report);
});

The following table describes types of events.

Higher Level Producer

The higher level producer is a variant of the producer which can propagate callbacks to you upon message delivery.

const producer = new Kafka.HighLevelProducer({
  'metadata.broker.list': 'localhost:9092',
});

This will enrich the produce call so it will have a callback to tell you when the message has been delivered. You lose the ability to specify opaque tokens.

producer.produce(topicName, null, Buffer.from('alliance4ever'), null, Date.now(), (err, offset) => {
  // The offset if our acknowledgement level allows us to receive delivery offsets
  console.log(offset);
});

Additionally you can add serializers to modify the value of a produce for a key or value befor