Kafka-node is a Node.js client with Zookeeper integration for Apache Kafka 0.8.1 and later.
The Zookeeper integration does the following jobs:
- Loads broker metadata from Zookeeper before we can communicate with the Kafka server
- Watches broker state, if broker changes, the client will refresh broker and topic metadata stored in the client
Follow the instructions on the Kafka wiki to build Kafka 0.8 and get a test broker up and running.
connectionString: Zookeeper connection string, defaultlocalhost:2181/clientId: This is a user-supplied identifier for the client application, defaultkafka-node-clientzkOptions: Object, Zookeeper options, see node-zookeeper-clientnoAckBatchOptions: Object, when requireAcks is disabled on Producer side we can define the batch properties, 'noAckBatchSize' in bytes and 'noAckBatchAge' in milliseconds. The default value is{ noAckBatchSize: null, noAckBatchAge: null }and it acts as if there was no batch
Closes the connection to Zookeeper and the brokers so that the node process can exit gracefully.
cb: Function, the callback
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), default 0
partitionerType: 2
}var kafka = require('kafka-node'),
Producer = kafka.Producer,
client = new kafka.Client(),
producer = new Producer(client);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.
payloads: Array,array ofProduceRequest,ProduceRequestis 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', // only needed when using keyed partitioner
partition: 0, // default 0
attributes: 2 // default: 0
}cb: Function, the callback
attributes controls compression of the message set. It supports the following values:
0: No compression1: Compress using GZip2: Compress using snappy
Example:
var kafka = require('kafka-node'),
Producer = kafka.Producer,
KeyedMessage = kafka.KeyedMessage,
client = new kafka.Client(),
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) {})This method is used to create topics on the Kafka server. It only works 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 topicsasync: Boolean, async or synccb: Function, the callback
Example:
var kafka = require('kafka-node'),
Producer = kafka.Producer,
client = new kafka.Client(),
producer = new Producer(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 argclient: client which keeps a connection with the Kafka server. Round-robins produce requests to the available topic partitionsoptions: 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), default 2
partitionerType: 3
}var kafka = require('kafka-node'),
HighLevelProducer = kafka.HighLevelProducer,
client = new kafka.Client(),
producer = new HighLevelProducer(client);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.
payloads: Array,array ofProduceRequest,ProduceRequestis a JSON object like:
{
topic: 'topicName',
messages: ['message body'], // multi messages should be a array, single message can be just a string,
key: 'theKey', // only needed when using keyed partitioner
attributes: 1
}cb: Function, the callback
Example:
var kafka = require('kafka-node'),
HighLevelProducer = kafka.HighLevelProducer,
client = new kafka.Client(),
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);
});
});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 topicsasync: Boolean,async or synccb: Function,the callback
Example:
var kafka = require('kafka-node'),
HighLevelProducer = kafka.HighLevelProducer,
client = new kafka.Client(),
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 argclient: client which keeps a connection with the Kafka server. Note: it's recommend that create new client for different consumers.payloads: Array,array ofFetchRequest,FetchRequestis a JSON object like:
{
topic: 'topicName',
offset: 0, //default 0
}options: options for consumer,
{
groupId: 'kafka-node-group',//consumer group id, default `kafka-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: 'utf8'
}Example:
var kafka = require('kafka-node'),
Consumer = kafka.Consumer,
client = new kafka.Client(),
consumer = new Consumer(
client,
[
{ topic: 't', partition: 0 }, { topic: 't1', partition: 1 }
],
{
autoCommit: false
}
);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);
});Add topics to current consumer, if any topic to be added not exists, return error
topics: Array, array of topics to addcb: Function,the callbackfromOffset: 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);topics: Array, array of topics to removecb: Function, the callback
Example:
consumer.removeTopics(['t1', 't2'], function (err, removed) {
});Commit offset of the current topics manually, this method should be called when a consumer leaves
cb: Function, the callback
Example:
consumer.commit(function(err, data) {
});Set offset of the given topic
-
topic: String -
partition: Number -
offset: Number
Example:
consumer.setOffset('topic', 0, 0);Pause the consumer
Resume the consumer
Pause specify topics
consumer.pauseTopics([
'topic1',
{ topic: 'topic2', partition: 0 }
]);
Resume specify topics
consumer.resumeTopics([
'topic1',
{ topic: 'topic2', partition: 0 }
]);
force: Boolean, if set to true, it forces the consumer to commit the current offset before closing, defaultfalse
Example
consumer.close(true, cb);
consumer.close(cb); //force is disabledclient: client which keeps a connection with the Kafka server.payloads: Array,array ofFetchRequest,FetchRequestis a JSON object like:
{
topic: 'topicName'
}options: options for consumer,
{
// Consumer group id, deafult `kafka-node-group`
groupId: 'kafka-node-group',
// Consumer id, defaults to `groupId`
id: 'my-consumer-id',
// 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: 'utf8'
}Example:
var kafka = require('kafka-node'),
HighLevelConsumer = kafka.HighLevelConsumer,
client = new kafka.Client(),
consumer = new HighLevelConsumer(
client,
[
{ topic: 't' }, { topic: 't1' }
],
{
groupId: 'my-group'
}
);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);
});Add topics to current consumer, if any topic to be added not exists, return error
topics: Array, array of topics to addcb: Function,the callbackfromOffset: 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);topics: Array, array of topics to removecb: Function, the callback
Example:
consumer.removeTopics(['t1', 't2'], function (err, removed) {
});Commit offset of the current topics manually, this method should be called when a consumer leaves
cb: Function, the callback
Example:
consumer.commit(function(err, data) {
});Set offset of the given topic
-
topic: String -
partition: Number -
offset: Number
Example:
consumer.setOffset('topic', 0, 0);Pause the consumer
Resume the consumer
force: Boolean, if set to true, it forces the consumer to commit the current offset before closing, defaultfalse
Example:
consumer.close(true, cb);
consumer.close(cb); //force is disabledclient: client which keeps a connection with the Kafka server.
ready: when zookeeper is readyconnectwhen broker is ready
Fetch the available offset of a specific topic-partition
payloads: Array,array ofOffsetRequest,OffsetRequestis a JSON object like:
{
topic: 'topicName',
partition: 0, //default 0
// time:
// Used to ask for all messages before a certain time (ms), default Date.now(),
// Specify -1 to receive the latest offsets and -2 to receive the earliest available offset.
time: Date.now(),
maxNum: 1 //default 1
}cb: Function, the callback
Example
var kafka = require('kafka-node'),
client = new kafka.Client(),
offset = new kafka.Offset(client);
offset.fetch([
{ topic: 't', partition: 0, time: Date.now(), maxNum: 1 }
], function (err, data) {
// data
// { 't': { '0': [999] } }
});groupId: consumer grouppayloads: Array,array ofOffsetCommitRequest,OffsetCommitRequestis a JSON object like:
{
topic: 'topicName',
partition: 0, //default 0
offset: 1,
metadata: 'm', //default 'm'
}Example
var kafka = require('kafka-node'),
client = new kafka.Client(),
offset = new kafka.Offset(client);
offset.commit('groupId', [
{ topic: 't', partition: 0, offset: 10 }
], function (err, data) {
});Fetch the last committed offset in a topic of a specific consumer group
groupId: consumer grouppayloads: Array,array ofOffsetFetchRequest,OffsetFetchRequestis a JSON object like:
{
topic: 'topicName',
partition: 0 //default 0
}Example
var kafka = require('kafka-node'),
client = new kafka.Client(),
offset = new kafka.Offset(client);
offset.fetchCommits('groupId', [
{ topic: 't', partition: 0 }
], function (err, data) {
});- Compression: gzip & snappy (√)
Copyright (c) 2015 Sohu.com
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.