Node-kafka-wrapper

Kafka-node-wrapper is a custom wrapper built around the npm module kafka-node. It tries to simplify the complexities of implementing the boiler-plate and helps in focusing on the development of the application. It aims at a configure and use kind of strategy where in the developer just needs to setup the configuration of the client, producer/consumer.

Features

• Consumer Groups
• High Level Producer
• SSL based connections on Kafka 0.9 +
• SASL/PLAIN based connections Kafka 0.10 +

Kafka setup and installation

Kafka can be installed and setup by following the official Kafka Guide.

Usage

Example

const  kafka  =  require('node-kafka-wrapper');
function  sample() {
    kafka.producer.sendMessage(client_options, producer_options, ['sample_message'], 'sample_topic', () =>  console.log('Done'));
    kafka.consumer.subscribe(client_options, consumer_options, ['sample_topic'], (message) => {
    // Do some operations
});
}
sample();

Methods

sendMessage : behaves as a producer to send data to the broker
subscribe : behaves as a consumer group to subscribe to messages

Producer

Connects to the broker as a HighLevelProducer to publish data to a topic. The message can either be a string or keyed message.

The syntax for the producer is as follows.

const kafka = require('node-kafka-wrapper');
kafka.producer.sendMessage(client_options, producer_options, [string_message], topic_name, callback);

Consumer

Connects to the broker as a ConsumerGroup to subscribe to data from a topic.

The syntax for the consumer is as follows.

const kafka = require('node-kafka-wrapper');
kafka.consumer.subscribe(client_options, consumer_options, ['topic1', 'topic2', 'topic3'], callback);

Parameters

All the client, producer/consumer options compatible with the kafka-node library is accepted.

client_options

Example

{
  kafkaHost: 'www.kafka-domain.com',
  ssl: true,
  requestTimeout:  100000,
  sslOptions: {
    host:  'www.kafka-domain.com',
    ca: ['CA certificate'],
  },
}

• 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:9093default: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+)

producer_options

{
    // 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
}

consumer_options

var options = {
  kafkaHost: 'broker:9092', // connect directly to kafka broker (instantiates a KafkaClient)
  batch: undefined, // put client batch settings if you need them
  ssl: true, // optional (defaults to false) or tls options hash
  groupId: 'ExampleTestGroup',
  sessionTimeout: 15000,
  // An array of partition assignment protocols ordered by preference.
  // 'roundrobin' or 'range' string for built ins (see below to pass in custom assignment protocol)
  protocol: ['roundrobin'],
  encoding: 'utf8', // default is utf8, use 'buffer' for binary data
  // Offsets to use for new groups other options could be 'earliest' or 'none' (none will emit an error if no offsets were saved)
  // equivalent to Java client's auto.offset.reset
  fromOffset: 'latest', // default
  commitOffsetsOnFirstJoin: true, // on the very first time this consumer group subscribes to a topic, record the offset returned in fromOffset (latest/earliest)
  // how to recover from OutOfRangeOffset error (where save offset is past server retention) accepts same value as fromOffset
  outOfRangeOffset: 'earliest', // default
  // Callback to allow consumers with autoCommit false a chance to commit before a rebalance finishes
  // isAlreadyMember will be false on the first connection, and true on rebalances triggered after that
  onRebalance: (isAlreadyMember, callback) => { callback(); } // or null
};

Credits

This is just a wrapper to simplify the implementation of kafka-node.