@epiclabs/kafka-node
Advanced tools
Client for Apache Kafka v0.8.1+, v0.9.x, and v0.10.x. Refurbished by Epic
Kafka-node is a Node.js client with Zookeeper integration for Apache Kafka 0.8.1 and later. At Epiclabs we branch it out from version 3.0.1 to fix a known issue with Listeners. See section Client for new configuration.
Follow the instructions on the Kafka wiki to build Kafka 0.8 and get a test broker up and running.
New KafkaClient connects directly to Kafka brokers instead of connecting to zookeeper for broker discovery.
Client KafkaClient will not emit socket errors it will do its best to recover and only emit errors when it has exhausted its recovery attemptsready event is only emitted after successful connection to a broker and metadata request to that brokerClient uses zookeeper to discover the SSL kafka host/port since we connect directly to the broker this host/port for SSL need to be correctkafkaHost : 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: 10000requestTimeout : in ms for a kafka request to timeout default: 30000autoConnect : automatically connect when KafkaClient is instantiated otherwise you need to manually call connect default: trueconnectRetryOptions : 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 reconnect after being disconnected). The value is elapsed time in ms without any data written to the TCP socket. default: 5 minutesmaxAsyncRequests : maximum async operations at a time toward the kafka cluster. default: 10sslOptions: 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+)const client = new kafka.KafkaClient({kafkaHost: '10.3.100.196:9092'});
⚠️ This zookeeper based client has been deprecated and is likely to be removed in the future. Please use the KafkaClient instead.
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-client
noAckBatchOptions: 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, custom = 4), 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', // 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 and KafkaClient only)
}
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 requires Kafka 0.10+.
topics: Array, array of topicscb: Function, the callbackExample:
var kafka = require('kafka-node');
var client = new kafka.KafkaClient();
var topicsToCreate = [{
topic: 'topic1',
partitions: 1,
replicationFactor: 2
},
{
topic: 'topic2',
partitions: 5,
replicationFactor: 3
}];
client.createTopics(topics, (error, result) => {
// result is an array of any errors if a given topic could not be created
});
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
}
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', // 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 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
Requires: Kafka v0.9+
highWaterMark size of write buffer (Default: 100)kafkaClient options see KafkaClientproducer options for Producer see HighLevelProducerIn 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
stdinand 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
ConsumerGroupStreamto read from this topic and transform the data and feed the result of into theRebalanceTopicTopic.
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);
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,{
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',
keyEncoding: '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
Consumer implemented using node's Readable stream interface. Read more about streams here.
kafka-node each chunk is a kafka message100 messages and can be changed through the highWaterMark optionSimilar API as Consumer with some exceptions. Methods like pause and resume in ConsumerStream respects the toggling of flow mode in a Stream. In Consumer calling pause() just paused the fetch cycle and will continue to emit message events. Pausing in a ConsumerStream should immediately stop emitting data events.
⚠️ This consumer has been deprecated and is likely to be removed in the future. Please use the ConsumerGroup instead.
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, default `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 callbackExample:
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
The new consumer group uses Kafka broker coordinators instead of Zookeeper to manage consumer groups. This is supported in Kafka version 0.9 and above only.
API is very similar to HighLevelConsumer since it extends directly from HLC so many of the same options will apply with some exceptions noted below:
client this is done inside the ConsumerGroupConsumerGroup does not emit a registered eventvar options = {
host: 'zookeeper:2181', // zookeeper host omit if connecting directly to broker (see kafkaHost below)
kafkaHost: 'broker:9092', // connect directly to kafka broker (instantiates a KafkaClient)
zk : undefined, // put client zk settings if you need them (see Client)
batch: undefined, // put client batch settings if you need them (see Client)
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'],
// 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
migrateHLC: false, // for details please see Migration section below
migrateRolling: true,
// 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
};
var consumerGroup = new ConsumerGroup(options, ['RebalanceTopic', 'RebalanceTest']);
// Or for a single topic pass in a string
var consumerGroup = new ConsumerGroup(options, 'RebalanceTopic');
You can pass a custom assignment strategy to the protocol array with the interface:
topicPartition
{
"RebalanceTopic": [
"0",
"1",
"2"
],
"RebalanceTest": [
"0",
"1",
"2"
]
}
groupMembers
[
{
"subscription": [
"RebalanceTopic",
"RebalanceTest"
],
"version": 0,
"id": "consumer1-8db1b117-61c6-4f91-867d-20ccd1ad8b3d"
},
{
"subscription": [
"RebalanceTopic",
"RebalanceTest"
],
"version": 0,
"id": "consumer3-bf2d11f4-1c73-4a39-b498-cfe76eb65bea"
},
{
"subscription": [
"RebalanceTopic",
"RebalanceTest"
],
"version": 0,
"id": "consumer2-9781058e-fad4-40e8-a69c-69afbae05184"
}
]
callback(error, result)
result
[
{
"memberId": "consumer3-bf2d11f4-1c73-4a39-b498-cfe76eb65bea",
"topicPartitions": {
"RebalanceTopic": [
"2"
],
"RebalanceTest": [
"2"
]
},
"version": 0
},
{
"memberId": "consumer2-9781058e-fad4-40e8-a69c-69afbae05184",
"topicPartitions": {
"RebalanceTopic": [
"1"
],
"RebalanceTest": [
"1"
]
},
"version": 0
},
{
"memberId": "consumer1-8db1b117-61c6-4f91-867d-20ccd1ad8b3d",
"topicPartitions": {
"RebalanceTopic": [
"0"
],
"RebalanceTest": [
"0"
]
},
"version": 0
}
]
We have two options for automatic migration from existing highLevelConsumer group. This is useful to preserve the previous committed offsets for your group.
We support two use cases:
For case 1 use below setting:
{
migrateHLC: true, // default is false
migrateRolling: false // default is true
}
For case 2 setting migrateRolling to true will allow the ConsumerGroup to start monitoring zk nodes for when topic ownership are relinquished by the old HLC consumer. Once this is done the ConsumerGroup will connect and the previous HLC offsets from zookeeper will be migrated automatically to the new Kafka broker based coordinator.
fromOffset settingThe ConsumerGroup wrapped with a Readable stream interface. Read more about consuming Readable streams here.
Same notes in the Notes section of ConsumerStream applies to this stream.
ConsumerGroupStream manages auto commits differently than ConsumerGroup. Whereas the ConsumerGroup would automatically commit offsets of fetched messages the ConsumerGroupStream will only commit offsets of consumed messages from the stream buffer. This will be better for most users since it more accurately represents what was actually "Consumed". The interval at which auto commit fires off is still controlled by the autoCommitIntervalMs option and this feature can be disabled by setting autoCommit to false.
consumerGroupOptions same options to initialize a ConsumerGrouptopics a single or array of topics to subscribe toThis method can be used to commit manually when autoCommit is set to false.
message the original message or an object with {topic, partition, offset}force a commit even if there's a pending commitcallback (optional)Closes the ConsumerGroup. Calls callback when complete. If autoCommit is enabled calling close will also commit offsets consumed from the buffer.
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] } }
});
⚠️WARNING: commits are made to zookeeper and is only compatible with
HighLevelConsumerand will NOT with the newConsumerGroup
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) {
});
⚠️WARNING: commits are from zookeeper and is only compatible with
HighLevelConsumerand will NOT with the newConsumerGroup
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) {
});
⚠️WARNING: commits are from the broker and is only compatible with the new
ConsumerGroupand will NOT with the oldHighLevelConsumer
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.fetchCommitsV1('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]);
});
Example
var partition = 0;
var topic = 't';
offset.fetchEarliestOffsets([topic], function (error, offsets) {
if (error)
return handleError(error);
console.log(offsets[topic][partition]);
});
This class provides administrative APIs can be used to monitor and administer the Kafka cluster.
kafkaClient: client which keeps a connection with the Kafka server. (KafkaClient only, client not supported)List the consumer groups managed by the kafka cluster.
cb: Function, the callbackExample:
const client = new kafka.KafkaClient();
const admin = new kafka.Admin(client); // client must be KafkaClient
admin.listGroups((err, res) => {
console.log('consumerGroups', res);
});
Result:
consumerGroups { 'console-consumer-87148': 'consumer',
'console-consumer-2690': 'consumer',
'console-consumer-7439': 'consumer'
}
Fetch consumer group information from the cluster. See result for detailed information.
consumerGroups: Array, array of consumer groups (which can be gathered from listGroups)cb: Function, the callbackExample:
admin.describeGroups(['console-consumer-2690'], (err, res) => {
console.log(JSON.stringify(res,null,1));
})
Result:
{
"console-consumer-2690": {
"members": [
{
"memberId": "consumer-1-20195e12-cb3b-4ba4-9076-e7da8ed0d57a",
"clientId": "consumer-1",
"clientHost": "/192.168.61.1",
"memberMetadata": {
"subscription": [
"twice-tt"
],
"version": 0,
"userData": "JSON parse error",
"id": "consumer-1-20195e12-cb3b-4ba4-9076-e7da8ed0d57a"
},
"memberAssignment": {
"partitions": {
"twice-tt": [
0,
1
]
},
"version": 0,
"userData": "JSON Parse error"
}
}
],
"error": null,
"groupId": "console-consumer-2690",
"state": "Stable",
"protocolType": "consumer",
"protocol": "range",
"brokerId": "4"
}
}
var topics = [{
topic: 'topic1',
partitions: 1,
replicationFactor: 2
}];
admin.createTopics(topics, (err, res) => {
// result is an array of any errors if a given topic could not be created
})
See createTopics
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:*
Call client.loadMetadataForTopics with a blank topic array to get the entire list of available topics (and available brokers).
client.once('connect', function () {
client.loadMetadataForTopics([], function (error, results) {
if (error) {
return console.error(error);
}
console.log('%j', _.get(results, '1.metadata'));
});
});
If you are using the new ConsumerGroup simply set 'latest' to fromOffset option.
Otherwise:
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 can 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.In the consumer set the encoding option to buffer.
Set the messages attribute in the payload to a Buffer. TypedArrays such as Uint8Array are not supported and need to be converted to a Buffer.
{
messages: Buffer.from(data.buffer)
}
Snappy is a optional compression library. Windows users have reported issues with installing it while running npm install. It's optional in kafka-node and can be skipped by using the --no-optional flag (though errors from it should not fail the install).
npm install kafka-node --no-optional --save
Keep in mind if you try to use snappy without installing it kafka-node will throw a runtime exception.
By default, kafka-node uses debug to log important information. To integrate kafka-node's log output into an application, it is possible to set a logger provider. This enables filtering of log levels and easy redirection of output streams.
A logger provider is a function which takes the name of a logger and returns a logger implementation. For instance, the following code snippet shows how a logger provider for the global console object could be written:
function consoleLoggerProvider (name) {
// do something with the name
return {
debug: console.debug.bind(console),
info: console.info.bind(console),
warn: console.warn.bind(console),
error: console.error.bind(console)
};
}
The logger interface with its debug, info, warn and error methods expects format string support as seen in debug or the JavaScript console object. Many commonly used logging implementations cover this API, e.g. bunyan, pino or winston.
For performance reasons, initialization of the kafka-node module creates all necessary loggers. This means that custom logger providers need to be set before requiring the kafka-node module. The following example shows how this can be done:
// first configure the logger provider
const kafkaLogging = require('kafka-node/logging');
kafkaLogging.setLoggerProvider(consoleLoggerProvider);
// then require kafka-node and continue as normal
const kafka = require('kafka-node');
On the Mac install Docker for Mac.
npm test
Achieved using the KAFKA_VERSION environment variable.
# Runs "latest" kafka on docker hub*
npm test
# Runs test against other versions:
KAFKA_VERSION=0.8 npm test
KAFKA_VERSION=0.9 npm test
KAFKA_VERSION=0.10 npm test
KAFKA_VERSION=0.11 npm test
*See Docker hub tags entry for which version is considered latest.
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.8.1+, v0.9.x, and v0.10.x. Refurbished by Epic
We found that @epiclabs/kafka-node demonstrated a not healthy version release cadence and project activity because the last version was released a year ago. It has 5 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
A malicious npm package disguised as a WhatsApp client is exploiting authentication flows with a remote kill switch to exfiltrate data and destroy files.
Security News
PyPI now supports digital attestations, enhancing security and trust by allowing package maintainers to verify the authenticity of Python packages.
Security News
GitHub removed 27 malicious pull requests attempting to inject harmful code across multiple open source repositories, in another round of low-effort attacks.