kafka-node
Advanced tools
The kafka-node package is a client library for Apache Kafka, a distributed streaming platform. It allows you to produce and consume messages, manage Kafka topics, and handle Kafka streams in Node.js applications.
Producer
This feature allows you to produce messages to a Kafka topic. The code sample demonstrates how to create a Kafka producer, connect to a Kafka broker, and send a message to a specified topic.
const kafka = require('kafka-node');
const client = new kafka.KafkaClient({kafkaHost: 'localhost:9092'});
const producer = new kafka.Producer(client);
const payloads = [
{ topic: 'topic1', messages: 'hello world', partition: 0 }
];
producer.on('ready', function () {
producer.send(payloads, function (err, data) {
console.log(data);
});
});
producer.on('error', function (err) {
console.error('Producer error:', err);
});Consumer
This feature allows you to consume messages from a Kafka topic. The code sample demonstrates how to create a Kafka consumer, connect to a Kafka broker, and listen for messages on a specified topic.
const kafka = require('kafka-node');
const client = new kafka.KafkaClient({kafkaHost: 'localhost:9092'});
const consumer = new kafka.Consumer(
client,
[
{ topic: 'topic1', partition: 0 }
],
{
autoCommit: true
}
);
consumer.on('message', function (message) {
console.log(message);
});
consumer.on('error', function (err) {
console.error('Consumer error:', err);
});Admin
This feature allows you to perform administrative tasks such as listing topics. The code sample demonstrates how to create an admin client and list all topics in the Kafka cluster.
const kafka = require('kafka-node');
const client = new kafka.KafkaClient({kafkaHost: 'localhost:9092'});
const admin = new kafka.Admin(client);
admin.listTopics((err, res) => {
console.log(res);
});KafkaJS is a modern Apache Kafka client for Node.js. It is fully written in JavaScript and provides a more idiomatic and modern API compared to kafka-node. KafkaJS is known for its simplicity, better documentation, and active maintenance.
node-rdkafka is a high-performance Node.js client for Apache Kafka based on the C/C++ library librdkafka. It offers more advanced features and better performance compared to kafka-node, but it requires native module compilation.
Kafka-node is a Node.js client with Zookeeper integration for Apache Kafka 0.8.1 and later.
Follow the instructions on the Kafka wiki to build Kafka 0.8 and get a test broker up and running.
connectionString: Zookeeper connection string, default localhost:2181/clientId: This is a user-supplied identifier for the client application, default kafka-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 batchsslOptions: Object, options to be passed to the tls broker sockets, ex. { rejectUnauthorized: false } (Kafka +0.9)Closes the connection to Zookeeper and the brokers so that the node process can exit gracefully.
cb: Function, the callbackclient: 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 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', // only needed when using keyed partitioner
partition: 0, // default 0
attributes: 2 // default: 0
}
cb: Function, the callbackattributes controls compression of the message set. It supports the following values:
0: No compression1: Compress using GZip2: Compress using snappyExample:
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) {})
⚠️WARNING: Batch multiple messages of the same topic/partition together as an array on the
messagesattribute otherwise you may lose messages!
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 callbackExample:
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 arg
client: 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 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', // only needed when using keyed partitioner
attributes: 1
}
cb: Function, the callbackExample:
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);
});
});
⚠️WARNING: Batch multiple messages of the same topic/partition together as an array on the
messagesattribute otherwise you may lose messages!
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 callbackExample:
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 arg
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
}
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 comesExample:
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 callbackExample:
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 callbackExample:
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. Calling pause does not automatically stop messages from being emitted. This is because pause just stops the kafka consumer fetch loop. Each iteration of the fetch loop can obtain a batch of messages (limited by fetchMaxBytes).
Resume the consumer. Resumes the fetch loop.
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, default falseExample
consumer.close(true, cb);
consumer.close(cb); //force is disabled
client: client which keeps a connection with the Kafka server.payloads: Array,array of FetchRequest, FetchRequest is a JSON object like:{
topic: 'topicName'
}
options: options for consumer,{
// Consumer group id, deafult `kafka-node-group`
groupId: 'kafka-node-group',
// Optional consumer id, defaults to groupId + uuid
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 comesExample:
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 callbackExample:
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 callbackExample:
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. Calling pause does not automatically stop messages from being emitted. This is because pause just stops the kafka consumer fetch loop. Each iteration of the fetch loop can obtain a batch of messages (limited by fetchMaxBytes).
Resume the consumer. Resumes the fetch loop.
force: Boolean, if set to true, it forces the consumer to commit the current offset before closing, default falseExample:
consumer.close(true, cb);
consumer.close(cb); //force is disabled
client: client which keeps a connection with the Kafka server.ready: when zookeeper is readyconnect when broker is readyFetch the available offset of a specific topic-partition
payloads: Array,array of OffsetRequest, OffsetRequest is 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 callbackExample
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 of OffsetCommitRequest, OffsetCommitRequest is 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 of OffsetFetchRequest, OffsetFetchRequest is 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) {
});
Example
var partition = 0;
var topic = 't';
offset.fetchLatestOffsets([topic], function (error, offsets) {
if (error)
return handleError(error);
console.log(offsets[topic][partition]);
});
Error:
BrokerNotAvailableError: Could not find the leader
Call client.refreshMetadata() before sending the first message. Reference issue #354
This module uses the debug module so you can just run below before starting your app.
export DEBUG=kafka-node:*
offset.fetchLatestOffsets to get fetch the latest offsetReference issue #342
This error can occur when a HLC is killed and restarted quickly. The ephemeral nodes linked to the previous session are not relinquished in zookeeper when SIGINT is sent and instead relinquished when zookeeper session timeout is reached. The timeout can be adjusted using the sessionTimeout zookeeper option when the Client is created (the default is 30000ms).
Example handler:
process.on('SIGINT', function () {
highLevelConsumer.close(true, function () {
process.exit();
});
});
Alternatively, you to avoid this issue entirely by omitting the HLC's id and a unique one will be generated for you.
Reference issue #90
Your partition will be stuck if the fetchMaxBytes is smaller than the message produced. Increase fetchMaxBytes value should resolve this issue.
Reference to issue #339
async.queue with message processor and concurrency of one (the message processor itself is wrapped with setImmediate so it will not freeze up the event loop)queue.drain to resume the consumermessage event pauses the consumer and pushes the message to the queue.On the Mac you can either install docker-machine or Docker for Mac.
Docker machine:
brew install docker docker-machine docker-compose
docker-machine create --driver virtualbox dev
npm test
npm run stopDocker
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.
FAQs
Client for Apache Kafka v0.9.x, v0.10.x and v0.11.x
The npm package kafka-node receives a total of 110,460 weekly downloads. As such, kafka-node popularity was classified as popular.
We found that kafka-node demonstrated a not healthy version release cadence and project activity because the last version was released a year ago. It has 3 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.
Security News
The Ultralytics' PyPI Package was compromised four times in one weekend through GitHub Actions cache poisoning and failure to rotate previously compromised API tokens.