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:

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

var topicsToCreate = [{
  topic: 'topic1',
  partitions: 1,
  replicationFactor: 2
},
{
  topic: 'topic2',
  partitions: 5,
  replicationFactor: 3,
  // Optional set of config entries
  configEntries: [
    {
      name: 'compression.type',
      value: 'gzip'
    },
    {
      name: 'min.compaction.lag.ms',
      value: '50'
    }
  ],
  // Optional explicit partition / replica assignment
  // When this property exists, partitions and replicationFactor properties are ignored
  replicaAssignment: [
    {
      partition: 0,
      replicas: [3, 4]
    },
    {
      partition: 1,
      replicas: [2, 1]
    }
  ]
}];

client.createTopics(topicsToCreate, (error, result) => {
  // result is an array of any errors if a given topic could not be created
});

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.

const ProducerStream = require('./lib/producerStream');
const ConsumerGroupStream = require('./lib/consumerGroupStream');
const resultProducer = new ProducerStream();

const consumerOptions = {
  kafkaHost: '127.0.0.1:9092',
  groupId: 'ExampleTestGroup',
  sessionTimeout: 15000,
  protocol: ['roundrobin'],
  asyncPush: false,
  id: 'consumer1',
  fromOffset: 'latest'
};

const consumerGroup = new ConsumerGroupStream(consumerOptions, 'ExampleTopic');

const messageTransform = new Transform({
  objectMode: true,
  decodeStrings: true,
  transform (message, encoding, callback) {
    console.log(`Received message ${message.value} transforming input`);
    callback(null, {
      topic: 'RebalanceTopic',
      messages: `You have been (${message.value}) made an example of`
    });
  }
});

consumerGroup.pipe(messageTransform).pipe(resultProducer);

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,

``js { groupId: 'kafka-node-group',//consumer group id, defaultkafka-node-group` // Auto commit config autoCommit: true, autoCommitIntervalMs: 5000, // The max wait time is the maximum amount of time in milliseconds to block waiting if insufficient data is available at the time the request is issued, default 100ms fetchMaxWaitMs: 100, // This is the minimum number of bytes of messages that must be available to give a response, default 1 byte fetchMinBytes: 1, // The maximum bytes to include in the message set for this partition. This helps bound the size of the response. fetchMaxBytes: 1024 * 1024, // If set true, consumer will fetch message from the given offset in the payloads fromOffset: false, // If set to 'buffer', values will be returned as raw buffer objects. encoding: 'utf