kafka-node

Kafka-node

Kafka-node is a Node.js client for Apache Kafka 0.9 and later.

Table of Contents

• Features
• Install Kafka
• API
  • KafkaClient
  • Producer
  • HighLevelProducer
  • ProducerStream
  • Consumer
  • ConsumerStream
  • ConsumerGroup
  • ConsumerGroupStream
  • Offset
  • Admin
• Troubleshooting / FAQ
  • HighLevelProducer with KeyedPartitioner errors on first send
  • How do I debug an issue?
  • For a new consumer how do I start consuming from the latest message in a partition?
  • ConsumerGroup does not consume on all partitions
  • How to throttle messages / control the concurrency of processing messages
  • How do I produce and consume binary data?
  • What are these node-gyp and snappy errors?
  • How do I configure the log output?
  • Error: Not a message set. Magic byte is 2
• Running Tests
• LICENSE - "MIT"

Features

• Consumer
• Producer and High Level Producer
• Node Stream Producer (Kafka 0.9+)
• Node Stream Consumers (ConsumerGroupStream Kafka 0.9+)
• Manage topic Offsets
• SSL connections to brokers (Kafka 0.9+)
• SASL/PLAIN Authentication (Kafka 0.10+)
• Consumer Groups managed by Kafka coordinator (Kafka 0.9+)
• Connect directly to brokers (Kafka 0.9+)
• Administrative APIs
  • List Groups
  • Describe Groups
  • Create Topics

Install Kafka

Follow the instructions on the Kafka wiki to build Kafka and get a test broker up and running.

API

KafkaClient

New KafkaClient connects directly to Kafka brokers.

Options

• kafkaHost : A string of kafka broker/host combination delimited by comma for example: kafka-1.us-east-1.myapp.com:9093,kafka-2.us-east-1.myapp.com:9093,kafka-3.us-east-1.myapp.com:9093 default: localhost:9092.
• connectTimeout : in ms it takes to wait for a successful connection before moving to the next host default: 10000
• requestTimeout : in ms for a kafka request to timeout default: 30000
• autoConnect : automatically connect when KafkaClient is instantiated otherwise you need to manually call connect default: true
• connectRetryOptions : object hash that applies to the initial connection. see retry module for these options.
• idleConnection : allows the broker to disconnect an idle connection from a client (otherwise the clients continues to O after being disconnected). The value is elapsed time in ms without any data written to the TCP socket. default: 5 minutes
• reconnectOnIdle : when the connection is closed due to client idling, client will attempt to auto-reconnect. default: true
• maxAsyncRequests : maximum async operations at a time toward the kafka cluster. default: 10
• sslOptions: Object, options to be passed to the tls broker sockets, ex. { rejectUnauthorized: false } (Kafka 0.9+)
• sasl: Object, SASL authentication configuration (only SASL/PLAIN is currently supported), ex. { mechanism: 'plain', username: 'foo', password: 'bar' } (Kafka 0.10+)

Example

const client = new kafka.KafkaClient({kafkaHost: '10.3.100.196:9092'});

Producer

Producer(KafkaClient, [options], [customPartitioner])

• client: client which keeps a connection with the Kafka server.
• options: options for producer,

{
    // Configuration for when to consider a message as acknowledged, default 1
    requireAcks: 1,
    // The amount of time in milliseconds to wait for all acks before considered, default 100ms
    ackTimeoutMs: 100,
    // Partitioner type (default = 0, random = 1, cyclic = 2, keyed = 3, custom = 4), default 0
    partitionerType: 2
}

var kafka = require('kafka-node'),
    Producer = kafka.Producer,
    client = new kafka.KafkaClient(),
    producer = new Producer(client);

Events

• ready: this event is emitted when producer is ready to send messages.
• error: this is the error event propagates from internal client, producer should always listen it.

send(payloads, cb)

• payloads: Array,array of ProduceRequest, ProduceRequest is a JSON object like:

{
   topic: 'topicName',
   messages: ['message body'], // multi messages should be a array, single message can be just a string or a KeyedMessage instance
   key: 'theKey', // string or buffer, only needed when using keyed partitioner
   partition: 0, // default 0
   attributes: 2, // default: 0
   timestamp: Date.now() // <-- defaults to Date.now() (only available with kafka v0.10+)
}

• cb: Function, the callback

attributes controls compression of the message set. It supports the following values:

• 0: No compression
• 1: Compress using GZip
• 2: Compress using snappy

Example:

var kafka = require('kafka-node'),
    Producer = kafka.Producer,
    KeyedMessage = kafka.KeyedMessage,
    client = new kafka.KafkaClient(),
    producer = new Producer(client),
    km = new KeyedMessage('key', 'message'),
    payloads = [
        { topic: 'topic1', messages: 'hi', partition: 0 },
        { topic: 'topic2', messages: ['hello', 'world', km] }
    ];
producer.on('ready', function () {
    producer.send(payloads, function (err, data) {
        console.log(data);
    });
});

producer.on('error', function (err) {})

createTopics(topics, cb)

This method is used to create topics on the Kafka server. It requires Kafka 0.10+.

• topics: Array, array of topics
• cb: Function, the callback

Example:

…

HighLevelProducer

HighLevelProducer(KafkaClient, [options], [customPartitioner])

• client: client which keeps a connection with the Kafka server. Round-robins produce requests to the available topic partitions
• options: options for producer,

{
    // Configuration for when to consider a message as acknowledged, default 1
    requireAcks: 1,
    // The amount of time in milliseconds to wait for all acks before considered, default 100ms
    ackTimeoutMs: 100
}

var kafka = require('kafka-node'),
    HighLevelProducer = kafka.HighLevelProducer,
    client = new kafka.KafkaClient(),
    producer = new HighLevelProducer(client);

Events

• ready: this event is emitted when producer is ready to send messages.
• error: this is the error event propagates from internal client, producer should always listen it.

send(payloads, cb)

• payloads: Array,array of ProduceRequest, ProduceRequest is a JSON object like:

{
   topic: 'topicName',
   messages: ['message body'], // multi messages should be a array, single message can be just a string,
   key: 'theKey', // string or buffer, only needed when using keyed partitioner
   attributes: 1,
   timestamp: Date.now() // <-- defaults to Date.now() (only available with kafka v0.10 and KafkaClient only)
}

• cb: Function, the callback

Example:

var kafka = require('kafka-node'),
    HighLevelProducer = kafka.HighLevelProducer,
    client = new kafka.KafkaClient(),
    producer = new HighLevelProducer(client),
    payloads = [
        { topic: 'topic1', messages: 'hi' },
        { topic: 'topic2', messages: ['hello', 'world'] }
    ];
producer.on('ready', function () {
    producer.send(payloads, function (err, data) {
        console.log(data);
    });
});

createTopics(topics, async, cb)

This method is used to create topics on the Kafka server. It only work when auto.create.topics.enable, on the Kafka server, is set to true. Our client simply sends a metadata request to the server which will auto create topics. When async is set to false, this method does not return until all topics are created, otherwise it returns immediately.

• topics: Array,array of topics
• async: Boolean,async or sync
• cb: Function,the callback

Example:

var kafka = require('kafka-node'),
    HighLevelProducer = kafka.HighLevelProducer,
    client = new kafka.KafkaClient(),
    producer = new HighLevelProducer(client);
// Create topics sync
producer.createTopics(['t','t1'], false, function (err, data) {
    console.log(data);
});
// Create topics async
producer.createTopics(['t'], true, function (err, data) {});
producer.createTopics(['t'], function (err, data) {});// Simply omit 2nd arg

ProducerStream

ProducerStream (options)

Options

• highWaterMark size of write buffer (Default: 100)
• kafkaClient options see KafkaClient
• producer options for Producer see HighLevelProducer

Streams Example

In this example we demonstrate how to stream a source of data (from stdin) to kafka (ExampleTopic topic) for processing. Then in a separate instance (or worker process) we consume from that kafka topic and use a Transform stream to update the data and stream the result to a different topic using a ProducerStream.

Stream text from stdin and write that into a Kafka Topic

const Transform = require('stream').Transform;
const ProducerStream = require('./lib/producerStream');
const _ = require('lodash');
const producer = new ProducerStream();

const stdinTransform = new Transform({
  objectMode: true,
  decodeStrings: true,
  transform (text, encoding, callback) {
    text = _.trim(text);
    console.log(`pushing message ${text} to ExampleTopic`);
    callback(null, {
      topic: 'ExampleTopic',
      messages: text
    });
  }
});

process.stdin.setEncoding('utf8');
process.stdin.pipe(stdinTransform).pipe(producer);

Use ConsumerGroupStream to read from this topic and transform the data and feed the result of into the RebalanceTopic Topic.

…

Consumer

Consumer(client, payloads, options)

• client: client which keeps a connection with the Kafka server. Note: it's recommend that create new client for different consumers.
• payloads: Array,array of FetchRequest, FetchRequest is a JSON object like:

{
   topic: 'topicName',
   offset: 0, //default 0
   partition: 0 // default 0
}

• options: options for consumer,

…

Example:

var kafka = require('kafka-node'),
    Consumer = kafka.Consumer,
    client = new kafka.KafkaClient(),
    consumer = new Consumer(
        client,
        [
            { topic: 't', partition: 0 }, { topic: 't1', partition: 1 }
        ],
        {
            autoCommit: false
        }
    );

on('message', onMessage);

By default, we will consume messages from the last committed offset of the current group

• onMessage: Function, callback when new message comes

Example:

consumer.on('message', function (message) {
    console.log(message);
});

on('error', function (err) {})

on('offsetOutOfRange', function (err) {})

addTopics(topics, cb, fromOffset)

Add topics to current consumer, if any topic to be added not exists, return error

• topics: Array, array of topics to add
• cb: Function,the callback
• fromOffset: Boolean, if true, the consumer will fetch message from the specified offset, otherwise it will fetch message from the last commited offset of the topic.

Example:

consumer.addTopics(['t1', 't2'], function (err, added) {
});

or

consumer.addTopics([{ topic: 't1', offset: 10 }], function (err, added) {
}, true);