1. module kafka-node
   1. function kafka-node.BrokerNotAvailableError (message)
   2. function kafka-node.BrokerReadable (options)
   3. function kafka-node.BrokerTransform (options)
   4. function kafka-node.BrokerWrapper (socket, noAckBatchOptions)
   5. function kafka-node.Client (connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions)
   6. function kafka-node.Consumer (client, topics, options)
   7. function kafka-node.ConsumerGroup (memberOptions, topics)
   8. function kafka-node.CustomPartitioner (partitioner)
   9. function kafka-node.CyclicPartitioner ()
   10. function kafka-node.DefaultPartitioner ()
   11. function kafka-node.FailedToRebalanceConsumerError (message)
   12. function kafka-node.FailedToRegisterConsumerError (message, nested)
   13. function kafka-node.GroupCoordinatorNotAvailableError (message)
   14. function kafka-node.GroupLoadInProgressError (message)
   15. function kafka-node.HeartbeatTimeoutError (message)
   16. function kafka-node.HighLevelConsumer (client, topics, options)
   17. function kafka-node.HighLevelProducer (client, options, customPartitioner)
   18. function kafka-node.HighLevelProducer.super_ (client, options, defaultPartitionerType, customPartitioner)
   19. function kafka-node.IllegalGenerationError (message)
   20. function kafka-node.InvalidConfigError (message)
   21. function kafka-node.InvalidConsumerOffsetError (message, nested)
   22. function kafka-node.KafkaBuffer (batchSize, batchAge)
   23. function kafka-node.KeyedMessage (key, value)
   24. function kafka-node.KeyedPartitioner ()
   25. function kafka-node.NotCoordinatorForGroupError (message)
   26. function kafka-node.Offset (client)
   27. function kafka-node.Producer (client, options, customPartitioner)
   28. function kafka-node.RandomPartitioner ()
   29. function kafka-node.RebalanceInProgressError (message)
   30. function kafka-node.TopicsNotExistError (topics)
   31. function kafka-node.UnknownMemberIdError (message)
   32. function kafka-node.consumerGroupMigrator (consumerGroup)
   33. function kafka-node.consumerGroupRecovery (consumerGroup)
   34. object kafka-node.BrokerReadable.prototype
   35. object kafka-node.BrokerTransform.prototype
   36. object kafka-node.BrokerWrapper.prototype
   37. object kafka-node.Client.prototype
   38. object kafka-node.Consumer.prototype
   39. object kafka-node.ConsumerGroup.prototype
   40. object kafka-node.CyclicPartitioner.prototype
   41. object kafka-node.DefaultPartitioner.prototype
   42. object kafka-node.HighLevelConsumer.prototype
   43. object kafka-node.HighLevelProducer.super_.prototype
   44. object kafka-node.KafkaBuffer.prototype
   45. object kafka-node.KeyedPartitioner.prototype
   46. object kafka-node.Offset.prototype
   47. object kafka-node.RandomPartitioner.prototype
   48. object kafka-node.consumerGroupMigrator.prototype
   49. object kafka-node.consumerGroupRecovery.prototype
   50. object kafka-node.logging
   51. object kafka-node.partitioner
   52. object kafka-node.protocol
   53. object kafka-node.protocol_struct
   54. object kafka-node.range
   55. object kafka-node.roundrobin
   56. object kafka-node.snappy
   57. object kafka-node.utils
   58. object kafka-node.zookeeper
2. module kafka-node.BrokerNotAvailableError
   1. function kafka-node.BrokerNotAvailableError (message)
   2. function kafka-node.BrokerNotAvailableError.super_ ()
3. module kafka-node.BrokerReadable
   1. function kafka-node.BrokerReadable (options)
   2. function kafka-node.BrokerReadable.super_ (options)
4. module kafka-node.BrokerReadable.prototype
   1. function kafka-node.BrokerReadable.prototype._read (size)
5. module kafka-node.BrokerTransform
   1. function kafka-node.BrokerTransform (options)
   2. function kafka-node.BrokerTransform.super_ (options)
6. module kafka-node.BrokerTransform.prototype
   1. function kafka-node.BrokerTransform.prototype._transform (chunk, enc, done)
   2. function kafka-node.BrokerTransform.prototype._transformNext ()
7. module kafka-node.BrokerWrapper
   1. function kafka-node.BrokerWrapper (socket, noAckBatchOptions)
8. module kafka-node.BrokerWrapper.prototype
   1. function kafka-node.BrokerWrapper.prototype.write (buffer)
   2. function kafka-node.BrokerWrapper.prototype.writeAsync (buffer)
9. module kafka-node.Client
   1. function kafka-node.Client (connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions)
   2. function kafka-node.Client.super_ ()
10. module kafka-node.Client.prototype
    1. function kafka-node.Client.prototype.addTopics (topics, cb)
    2. function kafka-node.Client.prototype.brokerForLeader (leader, longpolling)
    3. function kafka-node.Client.prototype.checkMetadatas (payloads)
    4. function kafka-node.Client.prototype.clearCallbackQueue (socket, error)
    5. function kafka-node.Client.prototype.close (cb)
    6. function kafka-node.Client.prototype.closeBrokers (brokers)
    7. function kafka-node.Client.prototype.connect ()
    8. function kafka-node.Client.prototype.createBroker (host, port, longpolling)
    9. function kafka-node.Client.prototype.createTopics (topics, isAsync, cb)
    10. function kafka-node.Client.prototype.getBrokers (longpolling)
    11. function kafka-node.Client.prototype.handleReceivedData (socket)
    12. function kafka-node.Client.prototype.hasMetadata (topic, partition)
    13. function kafka-node.Client.prototype.leaderByPartition (topic, partition)
    14. function kafka-node.Client.prototype.loadMetadataForTopics (topics, cb)
    15. function kafka-node.Client.prototype.nextId ()
    16. function kafka-node.Client.prototype.nextSocketId ()
    17. function kafka-node.Client.prototype.payloadsByLeader (payloads)
    18. function kafka-node.Client.prototype.queueCallback (socket, id, data)
    19. function kafka-node.Client.prototype.reconnectBroker (oldSocket)
    20. function kafka-node.Client.prototype.refreshBrokers ()
    21. function kafka-node.Client.prototype.refreshMetadata (topicNames, cb)
    22. function kafka-node.Client.prototype.removeTopicMetadata (topics, cb)
    23. function kafka-node.Client.prototype.send (payloads, encoder, decoder, cb)
    24. function kafka-node.Client.prototype.sendFetchRequest (consumer, payloads, fetchMaxWaitMs, fetchMinBytes, maxTickMessages)
    25. function kafka-node.Client.prototype.sendGroupCoordinatorRequest (groupId, cb)
    26. function kafka-node.Client.prototype.sendGroupRequest (encode, decode, requestArgs)
    27. function kafka-node.Client.prototype.sendHeartbeatRequest (groupId, generationId, memberId, cb)
    28. function kafka-node.Client.prototype.sendJoinGroupRequest (groupId, memberId, sessionTimeout, groupProtocol, cb)
    29. function kafka-node.Client.prototype.sendLeaveGroupRequest (groupId, memberId, cb)
    30. function kafka-node.Client.prototype.sendOffsetCommitRequest (group, payloads, cb)
    31. function kafka-node.Client.prototype.sendOffsetCommitV2Request (group, generationId, memberId, payloads, cb)
    32. function kafka-node.Client.prototype.sendOffsetFetchRequest (group, payloads, cb)
    33. function kafka-node.Client.prototype.sendOffsetFetchV1Request (group, payloads, cb)
    34. function kafka-node.Client.prototype.sendOffsetRequest (payloads, cb)
    35. function kafka-node.Client.prototype.sendProduceRequest (payloads, requireAcks, ackTimeoutMs, cb)
    36. function kafka-node.Client.prototype.sendSyncGroupRequest (groupId, generationId, memberId, groupAssignment, cb)
    37. function kafka-node.Client.prototype.sendToBroker (payloads, encoder, decoder, cb)
    38. function kafka-node.Client.prototype.setupBroker (host, port, longpolling, brokers)
    39. function kafka-node.Client.prototype.setupBrokerProfiles (brokers)
    40. function kafka-node.Client.prototype.topicExists (topics, cb)
    41. function kafka-node.Client.prototype.unqueueCallback (socket, id)
    42. function kafka-node.Client.prototype.updateMetadatas (metadatas)
11. module kafka-node.Consumer
    1. function kafka-node.Consumer (client, topics, options)
    2. function kafka-node.Consumer.super_ ()
12. module kafka-node.Consumer.prototype
    1. function kafka-node.Consumer.prototype.addTopics (topics, cb, fromOffset)
    2. function kafka-node.Consumer.prototype.autoCommit (force, cb)
    3. function kafka-node.Consumer.prototype.buildPayloads (payloads)
    4. function kafka-node.Consumer.prototype.close (force, cb)
    5. function kafka-node.Consumer.prototype.commit (force, cb)
    6. function kafka-node.Consumer.prototype.connect ()
    7. function kafka-node.Consumer.prototype.fetch ()
    8. function kafka-node.Consumer.prototype.fetchOffset (payloads, cb)
    9. function kafka-node.Consumer.prototype.init ()
    10. function kafka-node.Consumer.prototype.pause ()
    11. function kafka-node.Consumer.prototype.pauseTopics (topics)
    12. function kafka-node.Consumer.prototype.removeTopics (topics, cb)
    13. function kafka-node.Consumer.prototype.resume ()
    14. function kafka-node.Consumer.prototype.resumeTopics (topics)
    15. function kafka-node.Consumer.prototype.setOffset (topic, partition, offset)
    16. function kafka-node.Consumer.prototype.updateOffsets (topics, initing)
13. module kafka-node.ConsumerGroup
    1. function kafka-node.ConsumerGroup (memberOptions, topics)
    2. function kafka-node.ConsumerGroup.super_ (client, topics, options)
14. module kafka-node.ConsumerGroup.prototype
    1. function kafka-node.ConsumerGroup.prototype.assignPartitions (protocol, groupMembers, callback)
    2. function kafka-node.ConsumerGroup.prototype.close (force, cb)
    3. function kafka-node.ConsumerGroup.prototype.connect ()
    4. function kafka-node.ConsumerGroup.prototype.fetchOffset (payloads, cb)
    5. function kafka-node.ConsumerGroup.prototype.getDefaultOffset (tp, defaultOffset)
    6. function kafka-node.ConsumerGroup.prototype.getOffset ()
    7. function kafka-node.ConsumerGroup.prototype.handleJoinGroup (joinGroupResponse, callback)
    8. function kafka-node.ConsumerGroup.prototype.handleSyncGroup (syncGroupResponse, callback)
    9. function kafka-node.ConsumerGroup.prototype.leaveGroup (callback)
    10. function kafka-node.ConsumerGroup.prototype.saveDefaultOffsets (topicPartitionList, callback)
    11. function kafka-node.ConsumerGroup.prototype.scheduleReconnect (timeout)
    12. function kafka-node.ConsumerGroup.prototype.sendHeartbeat ()
    13. function kafka-node.ConsumerGroup.prototype.sendOffsetCommitRequest (commits, cb)
    14. function kafka-node.ConsumerGroup.prototype.setCoordinatorId (coordinatorId)
    15. function kafka-node.ConsumerGroup.prototype.setupProtocols (protocols)
    16. function kafka-node.ConsumerGroup.prototype.startHeartbeats ()
    17. function kafka-node.ConsumerGroup.prototype.stopHeartbeats ()
15. module kafka-node.CustomPartitioner
    1. function kafka-node.CustomPartitioner (partitioner)
    2. function kafka-node.CustomPartitioner.super_ ()
16. module kafka-node.CyclicPartitioner
    1. function kafka-node.CyclicPartitioner ()
    2. function kafka-node.CyclicPartitioner.super_ ()
17. module kafka-node.CyclicPartitioner.prototype
    1. function kafka-node.CyclicPartitioner.prototype.getPartition (partitions)
18. module kafka-node.DefaultPartitioner
    1. function kafka-node.DefaultPartitioner ()
    2. function kafka-node.DefaultPartitioner.super_ ()
19. module kafka-node.DefaultPartitioner.prototype
    1. function kafka-node.DefaultPartitioner.prototype.getPartition (partitions)
20. module kafka-node.FailedToRebalanceConsumerError
    1. function kafka-node.FailedToRebalanceConsumerError (message)
    2. function kafka-node.FailedToRebalanceConsumerError.super_ ()
21. module kafka-node.FailedToRegisterConsumerError
    1. function kafka-node.FailedToRegisterConsumerError (message, nested)
    2. function kafka-node.FailedToRegisterConsumerError.super_ (message, nested)
22. module kafka-node.GroupCoordinatorNotAvailableError
    1. function kafka-node.GroupCoordinatorNotAvailableError (message)
    2. function kafka-node.GroupCoordinatorNotAvailableError.super_ ()
23. module kafka-node.GroupLoadInProgressError
    1. function kafka-node.GroupLoadInProgressError (message)
    2. function kafka-node.GroupLoadInProgressError.super_ ()
24. module kafka-node.HeartbeatTimeoutError
    1. function kafka-node.HeartbeatTimeoutError (message)
    2. function kafka-node.HeartbeatTimeoutError.super_ ()
25. module kafka-node.HighLevelConsumer
    1. function kafka-node.HighLevelConsumer (client, topics, options)
    2. function kafka-node.HighLevelConsumer.super_ ()
26. module kafka-node.HighLevelConsumer.prototype
    1. function kafka-node.HighLevelConsumer.prototype._releasePartitions (topicPayloads, callback)
    2. function kafka-node.HighLevelConsumer.prototype.addTopics (topics, cb)
    3. function kafka-node.HighLevelConsumer.prototype.autoCommit (force, cb)
    4. function kafka-node.HighLevelConsumer.prototype.buildPayloads (payloads)
    5. function kafka-node.HighLevelConsumer.prototype.buildTopicPayloads (topics)
    6. function kafka-node.HighLevelConsumer.prototype.close (force, cb)
    7. function kafka-node.HighLevelConsumer.prototype.commit (force, cb)
    8. function kafka-node.HighLevelConsumer.prototype.connect ()
    9. function kafka-node.HighLevelConsumer.prototype.fetch ()
    10. function kafka-node.HighLevelConsumer.prototype.fetchOffset (payloads, cb)
    11. function kafka-node.HighLevelConsumer.prototype.getTopicPayloads ()
    12. function kafka-node.HighLevelConsumer.prototype.init ()
    13. function kafka-node.HighLevelConsumer.prototype.leaveGroup (cb)
    14. function kafka-node.HighLevelConsumer.prototype.offsetRequest (payloads, cb)
    15. function kafka-node.HighLevelConsumer.prototype.pause ()
    16. function kafka-node.HighLevelConsumer.prototype.rebalanceAttempt (oldTopicPayloads, cb)
    17. function kafka-node.HighLevelConsumer.prototype.registerConsumer (cb)
    18. function kafka-node.HighLevelConsumer.prototype.removeTopics (topics, cb)
    19. function kafka-node.HighLevelConsumer.prototype.resume ()
    20. function kafka-node.HighLevelConsumer.prototype.sendOffsetCommitRequest (commits, cb)
    21. function kafka-node.HighLevelConsumer.prototype.setOffset (topic, partition, offset)
    22. function kafka-node.HighLevelConsumer.prototype.stop (cb)
    23. function kafka-node.HighLevelConsumer.prototype.updateOffsets (topics, initing)
27. module kafka-node.HighLevelProducer
    1. function kafka-node.HighLevelProducer (client, options, customPartitioner)
    2. function kafka-node.HighLevelProducer.super_ (client, options, defaultPartitionerType, customPartitioner)
    3. object kafka-node.HighLevelProducer.PARTITIONER_TYPES
28. module kafka-node.HighLevelProducer.super_
    1. function kafka-node.HighLevelProducer.super_ ()
    2. object kafka-node.HighLevelProducer.super_.PARTITIONER_TYPES
29. module kafka-node.HighLevelProducer.super_.prototype
    1. function kafka-node.HighLevelProducer.super_.prototype.buildPayloads (payloads, topicMetadata)
    2. function kafka-node.HighLevelProducer.super_.prototype.close (cb)
    3. function kafka-node.HighLevelProducer.super_.prototype.connect ()
    4. function kafka-node.HighLevelProducer.super_.prototype.createTopics (topics, async, cb)
    5. function kafka-node.HighLevelProducer.super_.prototype.send (payloads, cb)
30. module kafka-node.IllegalGenerationError
    1. function kafka-node.IllegalGenerationError (message)
    2. function kafka-node.IllegalGenerationError.super_ ()
31. module kafka-node.InvalidConfigError
    1. function kafka-node.InvalidConfigError (message)
    2. function kafka-node.InvalidConfigError.super_ ()
32. module kafka-node.InvalidConsumerOffsetError
    1. function kafka-node.InvalidConsumerOffsetError (message, nested)
    2. function kafka-node.InvalidConsumerOffsetError.super_ (message, nested)
33. module kafka-node.KafkaBuffer
    1. function kafka-node.KafkaBuffer (batchSize, batchAge)
34. module kafka-node.KafkaBuffer.prototype
    1. function kafka-node.KafkaBuffer.prototype._setupTimer (callback)
    2. function kafka-node.KafkaBuffer.prototype.addChunk (buffer, callback)
    3. function kafka-node.KafkaBuffer.prototype.getBatch ()
    4. function kafka-node.KafkaBuffer.prototype.truncateBatch ()
35. module kafka-node.KeyedPartitioner
    1. function kafka-node.KeyedPartitioner ()
    2. function kafka-node.KeyedPartitioner.super_ ()
36. module kafka-node.KeyedPartitioner.prototype
    1. function kafka-node.KeyedPartitioner.prototype.getPartition (partitions, key)
    2. function kafka-node.KeyedPartitioner.prototype.hashCode (string)
37. module kafka-node.NotCoordinatorForGroupError
    1. function kafka-node.NotCoordinatorForGroupError (message)
    2. function kafka-node.NotCoordinatorForGroupError.super_ ()
38. module kafka-node.Offset
    1. function kafka-node.Offset (client)
    2. function kafka-node.Offset.super_ ()
39. module kafka-node.Offset.prototype
    1. function kafka-node.Offset.prototype.buildPayloads (payloads)
    2. function kafka-node.Offset.prototype.commit (groupId, payloads, cb)
    3. function kafka-node.Offset.prototype.fetch (payloads, cb)
    4. function kafka-node.Offset.prototype.fetchCommits (groupId, payloads, cb)
    5. function kafka-node.Offset.prototype.fetchEarliestOffsets (topics, cb)
    6. function kafka-node.Offset.prototype.fetchLatestOffsets (topics, cb)
40. module kafka-node.Producer
    1. function kafka-node.Producer (client, options, customPartitioner)
    2. function kafka-node.Producer.super_ (client, options, defaultPartitionerType, customPartitioner)
    3. object kafka-node.Producer.PARTITIONER_TYPES
41. module kafka-node.RandomPartitioner
    1. function kafka-node.RandomPartitioner ()
    2. function kafka-node.RandomPartitioner.super_ ()
42. module kafka-node.RandomPartitioner.prototype
    1. function kafka-node.RandomPartitioner.prototype.getPartition (partitions)
43. module kafka-node.RebalanceInProgressError
    1. function kafka-node.RebalanceInProgressError (message)
    2. function kafka-node.RebalanceInProgressError.super_ ()
44. module kafka-node.TopicsNotExistError
    1. function kafka-node.TopicsNotExistError (topics)
    2. function kafka-node.TopicsNotExistError.super_ ()
45. module kafka-node.UnknownMemberIdError
    1. function kafka-node.UnknownMemberIdError (message)
    2. function kafka-node.UnknownMemberIdError.super_ ()
46. module kafka-node.consumerGroupMigrator
    1. function kafka-node.consumerGroupMigrator (consumerGroup)
    2. function kafka-node.consumerGroupMigrator.super_ ()
47. module kafka-node.consumerGroupMigrator.prototype
    1. function kafka-node.consumerGroupMigrator.prototype.checkForOwners (topics, listenForChange)
    2. function kafka-node.consumerGroupMigrator.prototype.checkForOwnersAndListenForChange (topics)
    3. function kafka-node.consumerGroupMigrator.prototype.connectConsumerGroup ()
    4. function kafka-node.consumerGroupMigrator.prototype.filterByExistingZkTopics (callback)
    5. function kafka-node.consumerGroupMigrator.prototype.getOffset (tp, defaultOffset)
    6. function kafka-node.consumerGroupMigrator.prototype.saveHighLevelConsumerOffsets (topicPartitionList, callback)
48. module kafka-node.consumerGroupRecovery
    1. function kafka-node.consumerGroupRecovery (consumerGroup)
49. module kafka-node.consumerGroupRecovery.prototype
    1. function kafka-node.consumerGroupRecovery.prototype.clearError ()
    2. function kafka-node.consumerGroupRecovery.prototype.getRetryTimeout (error)
    3. function kafka-node.consumerGroupRecovery.prototype.tryToRecoverFrom (error, source)
50. module kafka-node.logging
    1. function kafka-node.logging.setLoggerProvider (provider)
51. module kafka-node.partitioner
    1. function kafka-node.partitioner.CustomPartitioner (partitioner)
    2. function kafka-node.partitioner.CyclicPartitioner ()
    3. function kafka-node.partitioner.DefaultPartitioner ()
    4. function kafka-node.partitioner.KeyedPartitioner ()
    5. function kafka-node.partitioner.RandomPartitioner ()
52. module kafka-node.protocol
    1. function kafka-node.protocol.decodeFetchResponse (cb, maxTickMessages)
    2. function kafka-node.protocol.decodeGroupCoordinatorResponse (resp)
    3. function kafka-node.protocol.decodeGroupHeartbeat (resp)
    4. function kafka-node.protocol.decodeJoinGroupResponse (resp)
    5. function kafka-node.protocol.decodeLeaveGroupResponse (resp)
    6. function kafka-node.protocol.decodeMetadataResponse (resp)
    7. function kafka-node.protocol.decodeOffsetCommitResponse (resp)
    8. function kafka-node.protocol.decodeOffsetFetchResponse (resp)
    9. function kafka-node.protocol.decodeOffsetFetchV1Response (resp)
    10. function kafka-node.protocol.decodeOffsetResponse (resp)
    11. function kafka-node.protocol.decodeProduceResponse (resp)
    12. function kafka-node.protocol.decodeSyncGroupResponse (resp)
    13. function kafka-node.protocol.encodeFetchRequest (maxWaitMs, minBytes)
    14. function kafka-node.protocol.encodeGroupCoordinatorRequest (clientId, correlationId, groupId)
    15. function kafka-node.protocol.encodeGroupHeartbeat (clientId, correlationId, groupId, generationId, memberId)
    16. function kafka-node.protocol.encodeJoinGroupRequest (clientId, correlationId, groupId, memberId, sessionTimeout, groupProtocols)
    17. function kafka-node.protocol.encodeLeaveGroupRequest (clientId, correlationId, groupId, memberId)
    18. function kafka-node.protocol.encodeMessageSet (messageSet)
    19. function kafka-node.protocol.encodeMetadataRequest (clientId, correlationId, topics)
    20. function kafka-node.protocol.encodeOffsetCommitRequest (group)
    21. function kafka-node.protocol.encodeOffsetCommitV2Request (clientId, correlationId, group, generationId, memberId, payloads)
    22. function kafka-node.protocol.encodeOffsetFetchRequest (group)
    23. function kafka-node.protocol.encodeOffsetFetchV1Request (clientId, correlationId, group, payloads)
    24. function kafka-node.protocol.encodeOffsetRequest (clientId, correlationId, payloads)
    25. function kafka-node.protocol.encodeProduceRequest (requireAcks, ackTimeoutMs)
    26. function kafka-node.protocol.encodeSyncGroupRequest (clientId, correlationId, groupId, generationId, memberId, groupAssignment)
53. module kafka-node.protocol_struct
    1. function kafka-node.protocol_struct.FetchRequest ()
    2. function kafka-node.protocol_struct.FetchResponse ()
    3. function kafka-node.protocol_struct.KeyedMessage (key, value)
    4. function kafka-node.protocol_struct.Message ()
    5. function kafka-node.protocol_struct.OffsetCommitRequest ()
    6. function kafka-node.protocol_struct.OffsetCommitResponse ()
    7. function kafka-node.protocol_struct.PartitionMetadata ()
    8. function kafka-node.protocol_struct.ProduceRequest ()
    9. function kafka-node.protocol_struct.Request ()
    10. function kafka-node.protocol_struct.TopicAndPartition ()
    11. object kafka-node.protocol_struct.ERROR_CODE
    12. object kafka-node.protocol_struct.GROUP_ERROR
    13. object kafka-node.protocol_struct.KEYS
    14. object kafka-node.protocol_struct.REQUEST_TYPE
54. module kafka-node.range
    1. function kafka-node.range.assign (topicPartition, groupMembers, callback)
    2. number kafka-node.range.version
    3. string kafka-node.range.name
55. module kafka-node.roundrobin
    1. function kafka-node.roundrobin.assign (topicPartition, groupMembers, callback)
    2. number kafka-node.roundrobin.version
    3. string kafka-node.roundrobin.name
56. module kafka-node.snappy
    1. function kafka-node.snappy.decode (buffer, cb)
    2. function kafka-node.snappy.encode (input, callback)
57. module kafka-node.utils
    1. function kafka-node.utils.createTopicPartitionList (topicPartitions)
    2. function kafka-node.utils.groupPartitionsByTopic (topicPartitions)
    3. function kafka-node.utils.validateConfig (property, value)
    4. function kafka-node.utils.validateTopicNames (topics)
    5. function kafka-node.utils.validateTopics (topics)
58. module kafka-node.zookeeper
    1. function kafka-node.zookeeper.Zookeeper (connectionString, options)
    2. function kafka-node.zookeeper.ZookeeperConsumerMappings ()

module kafka-node

function kafka-node.BrokerNotAvailableError (message)

• description and source-code

  BrokerNotAvailableError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  ...
  
    requestArgs.unshift(this.clientId, correlationId);
  
    var request = encode.apply(null, requestArgs);
    var broker = this.brokerForLeader(this.coordinatorId);
  
    if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
      return cb(new errors.BrokerNotAvailableError('Broker not available'));
    }
  
    this.queueCallback(broker.socket, correlationId, [decode, cb]);
    broker.write(request);
  };
  
  Client.prototype.sendGroupCoordinatorRequest = function (groupId, cb) {
  ...

function kafka-node.BrokerReadable (options)

• description and source-code

  BrokerReadable = function (options) {
    Readable.call(this, options);
  }

• example usage

  n/a

function kafka-node.BrokerTransform (options)

• description and source-code

  BrokerTransform = function (options) {
    Transform.call(this, options);
    this.noAckBatchSize = options ? options.noAckBatchSize : null;
    this.noAckBatchAge = options ? options.noAckBatchAge : null;
    this._KafkaBuffer = new KafkaBuffer(this.noAckBatchSize, this.noAckBatchAge);
  }

• example usage

  n/a

function kafka-node.BrokerWrapper (socket, noAckBatchOptions)

• description and source-code

  BrokerWrapper = function (socket, noAckBatchOptions) {
    this.socket = socket;
  
    var self = this;
    var readable = new BrokerReadable();
    var transform = new BrokerTransform(noAckBatchOptions);
  
    readable.pipe(transform);
  
    transform.on('readable', function () {
      var bulkMessage = null;
      while (bulkMessage = transform.read()) { // eslint-disable-line no-cond-assign
        self.socket.write(bulkMessage);
      }
    });
  
    this.readableSocket = readable;
  }

• example usage

  n/a

function kafka-node.Client (connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions)

• description and source-code

  Client = function (connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions) {
    if (this instanceof Client === false) {
      return new Client(connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions);
    }
  
    this.sslOptions = sslOptions;
    this.ssl = !!sslOptions;
  
    if (clientId) {
      validateConfig('clientId', clientId);
    }
  
    this.connectionString = connectionString || 'localhost:2181/';
    this.clientId = clientId || 'kafka-node-client';
    this.zkOptions = zkOptions;
    this.noAckBatchOptions = noAckBatchOptions;
    this.brokers = {};
    this.longpollingBrokers = {};
    this.topicMetadata = {};
    this.topicPartitions = {};
    this.correlationId = 0;
    this._socketId = 0;
    this.cbqueue = {};
    this.brokerMetadata = {};
    this.ready = false;
    this.connect();
  }

• example usage

  ...
      partitionerType: 2
  }
  ```
  
  ``` js
  var kafka = require('kafka-node'),
      Producer = kafka.Producer,
      client = new kafka.Client(),
      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.
  ...

function kafka-node.Consumer (client, topics, options)

• description and source-code

  Consumer = function (client, topics, options) {
    if (!topics) {
      throw new Error('Must have payloads');
    }
  
    utils.validateTopics(topics);
  
    this.fetchCount = 0;
    this.client = client;
    this.options = _.defaults((options || {}), DEFAULTS);
    this.ready = false;
    this.paused = this.options.paused;
    this.id = nextId();
    this.payloads = this.buildPayloads(topics);
    this.connect();
    this.encoding = this.options.encoding;
  
    if (this.options.groupId) {
      utils.validateConfig('options.groupId', this.options.groupId);
    }
  }

• example usage

  n/a

function kafka-node.ConsumerGroup (memberOptions, topics)

• description and source-code

  function ConsumerGroup(memberOptions, topics) {
    EventEmitter.call(this);
    const self = this;
    this.options = _.defaults((memberOptions || {}), DEFAULTS);
  
    if (!this.options.heartbeatInterval) {
      this.options.heartbeatInterval = Math.floor(this.options.sessionTimeout / 3);
    }
  
    if (memberOptions.ssl === true) {
      memberOptions.ssl = {};
    }
  
    if (!(this.options.fromOffset in ACCEPTED_FROM_OFFSET)) {
      throw new Error(`fromOffset ${this.options.fromOffset} should be either: ${Object.keys(ACCEPTED_FROM_OFFSET).join(', ')}`);
    }
  
    if (!(this.options.outOfRangeOffset in ACCEPTED_FROM_OFFSET)) {
      throw new Error(`outOfRangeOffset ${this.options.outOfRangeOffset} should be either: ${Object.keys(ACCEPTED_FROM_OFFSET).join
  (', ')}`);
    }
  
    this.client = new Client(memberOptions.host, memberOptions.id, memberOptions.zk,
      memberOptions.batch, memberOptions.ssl);
  
    if (_.isString(topics)) {
      topics = [topics];
    }
  
    assert(Array.isArray(topics), 'Array of topics is required');
  
    this.topics = topics;
  
    this.recovery = new ConsumerGroupRecovery(this);
  
    this.setupProtocols(this.options.protocol);
  
    if (this.options.connectOnReady && !this.options.migrateHLC) {
      this.client.once('ready', this.connect.bind(this));
    }
  
    if (this.options.migrateHLC) {
      const ConsumerGroupMigrator = require('./consumerGroupMigrator');
      this.migrator = new ConsumerGroupMigrator(this);
      this.migrator.on('error', function (error) {
        self.emit('error', error);
      });
    }
  
    this.client.on('error', function (err) {
      logger.error('Error from %s', self.client.clientId, err);
      self.emit('error', err);
    });
  
    const recoverFromBrokerChange = _.debounce(function () {
      logger.debug('brokersChanged refreshing metadata');
      self.client.refreshMetadata(self.topics, function (error) {
        if (error) {
          self.emit(error);
          return;
        }
        self.paused = false;
        if (!self.ready && !self.connecting) {
          if (self.reconnectTimer) {
            // brokers changed so bypass backoff retry and reconnect now
            clearTimeout(self.reconnectTimer);
            self.reconnectTimer = null;
          }
          self.connect();
        } else if (!self.connecting) {
          self.fetch();
        }
      });
    }, 200);
  
    this.client.on('brokersChanged', function () {
      self.pause();
      recoverFromBrokerChange();
    });
  
    this.client.on('reconnect', function (lastError) {
      self.fetch();
    });
  
    this.on('offsetOutOfRange', topic => {
      this.pause();
      if (this.options.outOfRangeOffset === 'none') {
        this.emit('error', new errors.InvalidConsumerOffsetError(`Offset out of range for topic "${topic.topic}" partition ${topic
  .partition}`));
        return;
      }
  
      topic.time = ACCEPTED_FROM_OFFSET[this.options.outOfRangeOffset];
  
      this.getOffset().fetch([topic], (error, result) => {
        if (error) {
          this.emit('error', new errors.InvalidConsumerOffsetError(`Fetching ${this.options.outOfRangeOffset} offset failed`, error
  ));
          return;
        }
        const offset = _.head(result[topic.topic][topic.partition]);
        const oldOffset = _.find(this.topicPayloads, {topic: topic.topic, partition: topic.partition}).offset;
  
        logger.debug('replacing %s-%s stale offset of %d with %d', topic.topic, topic.partition, oldOffset, offset);
  
        this.setOffset(topic.topic, topic.partition, offset);
        this.resume();
      });
    });
  
    // 'done' will be emit when a message fetch request complete
    this.on('done', function (topics) {
      self.updateOffsets(topics);
      if (!self.paused) {
        setImmediate(function () {
          self.fetch();
        });
      }
    });
  
    if (this.options.groupId) {
      validateConfig('options.groupId', this.options.groupId);
    }
  
    this.isLeader = false;
    this.coordinatorId = null;
    this.generationId = null;
    this.ready = false;
    this.topicPayloads = [];
  }

• example usage

  n/a

function kafka-node.CustomPartitioner (partitioner)

• description and source-code

  CustomPartitioner = function (partitioner) {
    this.getPartition = partitioner;
  }

• example usage

  n/a

function kafka-node.CyclicPartitioner ()

• description and source-code

  CyclicPartitioner = function () {
    this.c = 0;
  }

• example usage

  n/a

function kafka-node.DefaultPartitioner ()

• description and source-code

  DefaultPartitioner = function () {}

• example usage

  n/a

function kafka-node.FailedToRebalanceConsumerError (message)

• description and source-code

  FailedToRebalanceConsumerError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  ...
        operation.attempt(function (currentAttempt) {
  self.rebalanceAttempt(oldTopicPayloads, function (err) {
    if (operation.retry(err)) {
      return;
    }
    if (err) {
      self.rebalancing = false;
      return self.emit('error', new errors.FailedToRebalanceConsumerError(operation
  .mainError().toString()));
    } else {
      var topicNames = self.topicPayloads.map(function (p) {
        return p.topic;
      });
      self.client.refreshMetadata(topicNames, function (err) {
        register();
        if (err) {
  ...

function kafka-node.FailedToRegisterConsumerError (message, nested)

• description and source-code

  FailedToRegisterConsumerError = function (message, nested) {
    NestedError.call(this, message, nested);
    this.message = message;
  }

• example usage

  ...
            } else {
              callback(new Error(util.format('Consumer %s is not registered in group %s', self.id, self.options.groupId)));
            }
          });
        }
      ], function (error) {
        if (error) {
          self.emit('error', new errors.FailedToRegisterConsumerError(error.toString
  (), error));
        }
      });
    }
  }, 20000);
  
  function fetchAndUpdateOffsets (cb) {
    self.fetchOffset(self.topicPayloads, function (err, topics) {
  ...

function kafka-node.GroupCoordinatorNotAvailableError (message)

• description and source-code

  GroupCoordinatorNotAvailableError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.GroupLoadInProgressError (message)

• description and source-code

  GroupLoadInProgressError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.HeartbeatTimeoutError (message)

• description and source-code

  HeartbeatTimeoutError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.HighLevelConsumer (client, topics, options)

• description and source-code

  HighLevelConsumer = function (client, topics, options) {
    if (!topics) {
      throw new Error('Must have payloads');
    }
    this.fetchCount = 0;
    this.client = client;
    this.options = _.defaults((options || {}), DEFAULTS);
    this.initialised = false;
    this.ready = false;
    this.closing = false;
    this.paused = this.options.paused;
    this.rebalancing = false;
    this.pendingRebalances = 0;
    this.committing = false;
    this.needToCommit = false;
    this.id = this.options.id || this.options.groupId + '_' + uuid.v4();
    this.payloads = this.buildPayloads(topics);
    this.topicPayloads = this.buildTopicPayloads(topics);
    this.connect();
  
    if (this.options.groupId) {
      validateConfig('options.groupId', this.options.groupId);
    }
  }

• example usage

  n/a

function kafka-node.HighLevelProducer (client, options, customPartitioner)

• description and source-code

  function HighLevelProducer(client, options, customPartitioner) {
    BaseProducer.call(this, client, options, BaseProducer.PARTITIONER_TYPES.cyclic, customPartitioner);
  }

• example usage

  n/a

function kafka-node.HighLevelProducer.super_ (client, options, defaultPartitionerType, customPartitioner)

• description and source-code

  function BaseProducer(client, options, defaultPartitionerType, customPartitioner) {
    options = options || {};
  
    this.ready = false;
    this.client = client;
  
    this.requireAcks = options.requireAcks === undefined
      ? DEFAULTS.requireAcks
      : options.requireAcks;
    this.ackTimeoutMs = options.ackTimeoutMs === undefined
      ? DEFAULTS.ackTimeoutMs
      : options.ackTimeoutMs;
  
    if (customPartitioner !== undefined && options.partitionerType !== PARTITIONER_TYPES.custom) {
      throw new Error('Partitioner Type must be custom if providing a customPartitioner.');
    } else if (customPartitioner === undefined && options.partitionerType === PARTITIONER_TYPES.custom) {
      throw new Error('No customer partitioner defined');
    }
  
    var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
    // eslint-disable-next-line
    this.partitioner = new partitionerType(customPartitioner);
  
    this.connect();
  }

• example usage

  n/a

function kafka-node.IllegalGenerationError (message)

• description and source-code

  IllegalGenerationError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.InvalidConfigError (message)

• description and source-code

  InvalidConfigError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.InvalidConsumerOffsetError (message, nested)

• description and source-code

  InvalidConsumerOffsetError = function (message, nested) {
    NestedError.apply(this, arguments);
  }

• example usage

  ...
    this.client.on('reconnect', function (lastError) {
  self.fetch();
    });
  
    this.on('offsetOutOfRange', topic => {
  this.pause();
  if (this.options.outOfRangeOffset === 'none') {
    this.emit('error', new errors.InvalidConsumerOffsetError(`Offset out of range
   for topic "${topic.topic}" partition ${topic.partition}`));
    return;
  }
  
  topic.time = ACCEPTED_FROM_OFFSET[this.options.outOfRangeOffset];
  
  this.getOffset().fetch([topic], (error, result) => {
    if (error) {
  ...

function kafka-node.KafkaBuffer (batchSize, batchAge)

• description and source-code

  KafkaBuffer = function (batchSize, batchAge) {
    this._batch_size = batchSize;
    this._batch_age = batchAge;
    this._batch_age_timer = null;
    this._buffer = null;
  }

• example usage

  n/a

function kafka-node.KeyedMessage (key, value)

• description and source-code

  function KeyedMessage(key, value) {
    exports.Message.call(this, 0, 0, key, value);
  }

• example usage

  n/a

function kafka-node.KeyedPartitioner ()

• description and source-code

  KeyedPartitioner = function () {}

• example usage

  n/a

function kafka-node.NotCoordinatorForGroupError (message)

• description and source-code

  NotCoordinatorForGroupError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.Offset (client)

• description and source-code

  Offset = function (client) {
    var self = this;
    this.client = client;
    this.ready = this.client.ready;
    this.client.on('ready', function () {
      self.ready = true;
      self.emit('ready');
    });
    this.client.once('connect', function () {
      self.emit('connect');
    });
    this.client.on('error', function (err) {
      self.emit('error', err);
    });
  }

• example usage

  ...
  * `cb`: *Function*, the callback
  
  Example
  
  ```js
  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] } }
      });
  ```
  ...

function kafka-node.Producer (client, options, customPartitioner)

• description and source-code

  function Producer(client, options, customPartitioner) {
    BaseProducer.call(this, client, options, BaseProducer.PARTITIONER_TYPES.default, customPartitioner);
  }

• example usage

  n/a

function kafka-node.RandomPartitioner ()

• description and source-code

  RandomPartitioner = function () {}

• example usage

  n/a

function kafka-node.RebalanceInProgressError (message)

• description and source-code

  RebalanceInProgressError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.TopicsNotExistError (topics)

• description and source-code

  TopicsNotExistError = function (topics) {
    Error.captureStackTrace(this, this);
    this.topics = topics;
    this.message = 'The topic(s) ' + topics.toString() + ' do not exist';
  }

• example usage

  ...
   */
  Client.prototype.topicExists = function (topics, cb) {
  var notExistsTopics = [];
  var self = this;
  
  async.each(topics, checkZK, function (err) {
    if (err) return cb(err);
    if (notExistsTopics.length) return cb(new errors.TopicsNotExistError(notExistsTopics
  ));
    cb();
  });
  
  function checkZK (topic, cb) {
    self.zk.topicExists(topic, function (err, existed, topic) {
      if (err) return cb(err);
      if (!existed) notExistsTopics.push(topic);
  ...

function kafka-node.UnknownMemberIdError (message)

• description and source-code

  UnknownMemberIdError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.consumerGroupMigrator (consumerGroup)

• description and source-code

  function ConsumerGroupMigrator(consumerGroup) {
    EventEmitter.call(this);
    assert(consumerGroup);
    const self = this;
    this.consumerGroup = consumerGroup;
    this.client = consumerGroup.client;
    var verified = 0;
  
    if (consumerGroup.options.migrateRolling) {
      this.zk = zookeeper.createClient(consumerGroup.client.connectionString, {retries: 10});
      this.zk.on('connected', function () {
        self.filterByExistingZkTopics(function (error, topics) {
          if (error) {
            return self.emit('error', error);
          }
  
          if (topics.length) {
            self.checkForOwnersAndListenForChange(topics);
          } else {
            logger.debug('No HLC topics exist in zookeeper.');
            self.connectConsumerGroup();
          }
        });
      });
  
      this.on('noOwnersForTopics', function (topics) {
        logger.debug('No owners for topics %s reported.', topics);
        if (++verified <= NUMER_OF_TIMES_TO_VERIFY) {
          logger.debug('%s verify %d of %d HLC has given up ownership by checking again in %d', self.client.clientId, verified,
            NUMER_OF_TIMES_TO_VERIFY, VERIFY_WAIT_TIME_MS);
  
          setTimeout(function () {
            self.checkForOwners(topics);
          }, VERIFY_WAIT_TIME_MS);
        } else {
          self.connectConsumerGroup();
        }
      });
  
      this.on('topicOwnerChange', _.debounce(function (topics) {
        verified = 0;
        self.checkForOwnersAndListenForChange(topics);
      }, 250));
  
      this.zk.connect();
    } else {
      this.connectConsumerGroup();
    }
  }

• example usage

  n/a

function kafka-node.consumerGroupRecovery (consumerGroup)

• description and source-code

  function ConsumerGroupRecovery(consumerGroup) {
    this.consumerGroup = consumerGroup;
    this.options = consumerGroup.options;
  }

• example usage

  n/a

module kafka-node.BrokerNotAvailableError

function kafka-node.BrokerNotAvailableError (message)

• description and source-code

  BrokerNotAvailableError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  ...
  
    requestArgs.unshift(this.clientId, correlationId);
  
    var request = encode.apply(null, requestArgs);
    var broker = this.brokerForLeader(this.coordinatorId);
  
    if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
      return cb(new errors.BrokerNotAvailableError('Broker not available'));
    }
  
    this.queueCallback(broker.socket, correlationId, [decode, cb]);
    broker.write(request);
  };
  
  Client.prototype.sendGroupCoordinatorRequest = function (groupId, cb) {
  ...

function kafka-node.BrokerNotAvailableError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.BrokerReadable

function kafka-node.BrokerReadable (options)

• description and source-code

  BrokerReadable = function (options) {
    Readable.call(this, options);
  }

• example usage

  n/a

function kafka-node.BrokerReadable.super_ (options)

• description and source-code

  function Readable(options) {
    if (!(this instanceof Readable))
      return new Readable(options);
  
    this._readableState = new ReadableState(options, this);
  
    // legacy
    this.readable = true;
  
    if (options && typeof options.read === 'function')
      this._read = options.read;
  
    Stream.call(this);
  }

• example usage

  n/a

module kafka-node.BrokerReadable.prototype

function kafka-node.BrokerReadable.prototype._read (size)

• description and source-code

  _read = function (size) {}

• example usage

  n/a

module kafka-node.BrokerTransform

function kafka-node.BrokerTransform (options)

• description and source-code

  BrokerTransform = function (options) {
    Transform.call(this, options);
    this.noAckBatchSize = options ? options.noAckBatchSize : null;
    this.noAckBatchAge = options ? options.noAckBatchAge : null;
    this._KafkaBuffer = new KafkaBuffer(this.noAckBatchSize, this.noAckBatchAge);
  }

• example usage

  n/a

function kafka-node.BrokerTransform.super_ (options)

• description and source-code

  function Transform(options) {
    if (!(this instanceof Transform))
      return new Transform(options);
  
    Duplex.call(this, options);
  
    this._transformState = new TransformState(this);
  
    var stream = this;
  
    // start out asking for a readable event once data is transformed.
    this._readableState.needReadable = true;
  
    // we have implemented the _read method, and done the other things
    // that Readable wants before the first _read call, so unset the
    // sync guard flag.
    this._readableState.sync = false;
  
    if (options) {
      if (typeof options.transform === 'function')
        this._transform = options.transform;
  
      if (typeof options.flush === 'function')
        this._flush = options.flush;
    }
  
    // When the writable side finishes, then flush out anything remaining.
    this.once('prefinish', function() {
      if (typeof this._flush === 'function')
        this._flush(function(er) {
          done(stream, er);
        });
      else
        done(stream);
    });
  }

• example usage

  n/a

module kafka-node.BrokerTransform.prototype

function kafka-node.BrokerTransform.prototype._transform (chunk, enc, done)

• description and source-code

  _transform = function (chunk, enc, done) {
    this._KafkaBuffer.addChunk(chunk, this._transformNext.bind(this));
    done();
  }

• example usage

  n/a

function kafka-node.BrokerTransform.prototype._transformNext ()

• description and source-code

  _transformNext = function () {
    this.push(this._KafkaBuffer.getBatch());
    this._KafkaBuffer.truncateBatch();
  }

• example usage

  n/a

module kafka-node.BrokerWrapper

function kafka-node.BrokerWrapper (socket, noAckBatchOptions)

• description and source-code

  BrokerWrapper = function (socket, noAckBatchOptions) {
    this.socket = socket;
  
    var self = this;
    var readable = new BrokerReadable();
    var transform = new BrokerTransform(noAckBatchOptions);
  
    readable.pipe(transform);
  
    transform.on('readable', function () {
      var bulkMessage = null;
      while (bulkMessage = transform.read()) { // eslint-disable-line no-cond-assign
        self.socket.write(bulkMessage);
      }
    });
  
    this.readableSocket = readable;
  }

• example usage

  n/a

module kafka-node.BrokerWrapper.prototype

function kafka-node.BrokerWrapper.prototype.write (buffer)

• description and source-code

  write = function (buffer) {
    this.socket.write(buffer);
  }

• example usage

  ...
    var broker = this.brokerForLeader(this.coordinatorId);
  
    if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
      return cb(new errors.BrokerNotAvailableError('Broker not available'));
    }
  
    this.queueCallback(broker.socket, correlationId, [decode, cb]);
    broker.write(request);
  };
  
  Client.prototype.sendGroupCoordinatorRequest = function (groupId, cb) {
    this.sendGroupRequest(protocol.encodeGroupCoordinatorRequest, protocol.decodeGroupCoordinatorResponse, arguments);
  };
  
  Client.prototype.sendJoinGroupRequest = function (groupId, memberId, sessionTimeout, groupProtocol, cb) {
  ...

function kafka-node.BrokerWrapper.prototype.writeAsync (buffer)

• description and source-code

  writeAsync = function (buffer) {
    this.readableSocket.push(buffer);
  }

• example usage

  ...
  
      if (longpolling) {
        if (broker.socket.waiting) continue;
        broker.socket.waiting = true;
      }
  
      if (decoder.requireAcks === 0) {
        broker.writeAsync(request);
        cb(null, { result: 'no ack' });
      } else {
        this.queueCallback(broker.socket, correlationId, [decoder, cb]);
        broker.write(request);
      }
    }
  };
  ...

module kafka-node.Client

function kafka-node.Client (connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions)

• description and source-code

  Client = function (connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions) {
    if (this instanceof Client === false) {
      return new Client(connectionString, clientId, zkOptions, noAckBatchOptions, sslOptions);
    }
  
    this.sslOptions = sslOptions;
    this.ssl = !!sslOptions;
  
    if (clientId) {
      validateConfig('clientId', clientId);
    }
  
    this.connectionString = connectionString || 'localhost:2181/';
    this.clientId = clientId || 'kafka-node-client';
    this.zkOptions = zkOptions;
    this.noAckBatchOptions = noAckBatchOptions;
    this.brokers = {};
    this.longpollingBrokers = {};
    this.topicMetadata = {};
    this.topicPartitions = {};
    this.correlationId = 0;
    this._socketId = 0;
    this.cbqueue = {};
    this.brokerMetadata = {};
    this.ready = false;
    this.connect();
  }

• example usage

  ...
      partitionerType: 2
  }
  ```
  
  ``` js
  var kafka = require('kafka-node'),
      Producer = kafka.Producer,
      client = new kafka.Client(),
      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.
  ...

function kafka-node.Client.super_ ()

• description and source-code

  function EventEmitter() {
    EventEmitter.init.call(this);
  }

• example usage

  n/a

module kafka-node.Client.prototype

function kafka-node.Client.prototype.addTopics (topics, cb)

• description and source-code

  addTopics = function (topics, cb) {
    var self = this;
    this.topicExists(topics, function (err) {
      if (err) return cb(err);
      self.loadMetadataForTopics(topics, function (err, resp) {
        if (err) return cb(err);
        self.updateMetadatas(resp);
        cb(null, topics);
      });
    });
  }

• example usage

  ...
  * `topics`: **Array**, array of topics to add
  * `cb`: **Function**,the callback
  * `fromOffset`: **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:
  
  ``` js
  consumer.addTopics(['t1', 't2'], function (err, added) {
  });
  
  or
  
  consumer.addTopics([{ topic: 't1', offset: 10 }], function (err, added) {
  }, true);
  ```
  ...

function kafka-node.Client.prototype.brokerForLeader (leader, longpolling)

• description and source-code

  brokerForLeader = function (leader, longpolling) {
    var addr;
    var brokers = this.getBrokers(longpolling);
    // If leader is not give, choose the first broker as leader
    if (typeof leader === 'undefined') {
      if (!_.isEmpty(brokers)) {
        addr = Object.keys(brokers)[0];
        return brokers[addr];
      } else if (!_.isEmpty(this.brokerMetadata)) {
        leader = Object.keys(this.brokerMetadata)[0];
      } else {
        return;
      }
    }
  
    var broker = _.find(this.brokerProfiles, {id: leader});
  
    if (!broker) {
      return;
    }
  
    addr = broker.host + ':' + broker.port;
  
    return brokers[addr] || this.setupBroker(broker.host, broker.port, longpolling, brokers);
  }

• example usage

  ...
  requestArgs = _.values(requestArgs);
  var cb = requestArgs.pop();
  var correlationId = this.nextId();
  
  requestArgs.unshift(this.clientId, correlationId);
  
  var request = encode.apply(null, requestArgs);
  var broker = this.brokerForLeader(this.coordinatorId);
  
  if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
    return cb(new errors.BrokerNotAvailableError('Broker not available'));
  }
  
  this.queueCallback(broker.socket, correlationId, [decode, cb]);
  broker.write(request);
  ...

function kafka-node.Client.prototype.checkMetadatas (payloads)

• description and source-code

  checkMetadatas = function (payloads) {
    if (_.isEmpty(this.topicMetadata)) return [ [], payloads ];
    // out: [ [metadata exists], [metadata not exists] ]
    var out = [ [], [] ];
    payloads.forEach(function (p) {
      if (this.hasMetadata(p.topic, p.partition)) out[0].push(p);
      else out[1].push(p);
    }.bind(this));
    return out;
  }

• example usage

  ...
  }
  };
  
  Client.prototype.send = function (payloads, encoder, decoder, cb) {
  var self = this;
  var _payloads = payloads;
  // payloads: [ [metadata exists], [metadata not exists] ]
  payloads = this.checkMetadatas(payloads);
  if (payloads[0].length && !payloads[1].length) {
    this.sendToBroker(_.flatten(payloads), encoder, decoder, cb);
    return;
  }
  if (payloads[1].length) {
    var topicNames = payloads[1].map(function (p) { return p.topic; });
    this.loadMetadataForTopics(topicNames, function (err, resp) {
  ...

function kafka-node.Client.prototype.clearCallbackQueue (socket, error)

• description and source-code

  clearCallbackQueue = function (socket, error) {
    var socketId = socket.socketId;
    var longpolling = socket.longpolling;
  
    if (!this.cbqueue.hasOwnProperty(socketId)) {
      return;
    }
  
    var queue = this.cbqueue[socketId];
  
    if (!longpolling) {
      Object.keys(queue).forEach(function (key) {
        var handlers = queue[key];
        var cb = handlers[1];
        cb(error);
      });
    }
    delete this.cbqueue[socketId];
  }

• example usage

  ...
  socket.on('error', function (err) {
    this.error = err;
    self.emit('error', err);
  });
  socket.on('close', function (hadError) {
    self.emit('close', this);
    if (hadError && this.error) {
      self.clearCallbackQueue(this, this.error);
    } else {
      self.clearCallbackQueue(this, new errors.BrokerNotAvailableError('Broker not available'));
    }
    retry(this);
  });
  socket.on('end', function () {
    retry(this);
  ...

function kafka-node.Client.prototype.close (cb)

• description and source-code

  close = function (cb) {
    this.closeBrokers(this.brokers);
    this.closeBrokers(this.longpollingBrokers);
    this.zk.close();
    cb && cb();
  }

• example usage

  ...
  
  ### close(force, cb)
  * `force`: **Boolean**, if set to true, it forces the consumer to commit the current offset before closing, default `false`
  
  Example
  
  ```js
  consumer.close(true, cb);
  consumer.close(cb); //force is disabled
  ```
  
  ## HighLevelConsumer
  ⚠️ ***This consumer has been deprecated in the latest version of Kafka (0.10.1) and is likely to be removed in the future. Please
   use the ConsumerGroup instead.***
  
  ### HighLevelConsumer(client, payloads, options)
  ...

function kafka-node.Client.prototype.closeBrokers (brokers)

• description and source-code

  closeBrokers = function (brokers) {
    _.each(brokers, function (broker) {
      broker.socket.closing = true;
      broker.socket.end();
    });
  }

• example usage

  ...
  
    self.brokerProfiles[addr] = brokerProfile;
    self.brokerProfiles[addr].id = key;
  });
  };
  
  Client.prototype.close = function (cb) {
  this.closeBrokers(this.brokers);
  this.closeBrokers(this.longpollingBrokers);
  this.zk.close();
  cb && cb();
  };
  
  Client.prototype.closeBrokers = function (brokers) {
  _.each(brokers, function (broker) {
  ...

function kafka-node.Client.prototype.connect ()

• description and source-code

  connect = function () {
    var zk = this.zk = new Zookeeper(this.connectionString, this.zkOptions);
    var self = this;
    zk.once('init', function (brokers) {
      try {
        self.ready = true;
        self.brokerMetadata = brokers;
        self.setupBrokerProfiles(brokers);
        Object
            .keys(self.brokerProfiles)
            .some(function (key, index) {
              var broker = self.brokerProfiles[key];
              self.setupBroker(broker.host, broker.port, false, self.brokers);
              // Only connect one broker
              return !index;
            });
        self.emit('ready');
      } catch (error) {
        self.ready = false;
        self.emit('error', error);
      }
    });
    zk.on('brokersChanged', function (brokerMetadata) {
      try {
        self.brokerMetadata = brokerMetadata;
        logger.debug('brokersChanged', brokerMetadata);
        self.setupBrokerProfiles(brokerMetadata);
        self.refreshBrokers();
        // Emit after a 3 seconds
        setTimeout(function () {
          self.emit('brokersChanged');
        }, 3000);
      } catch (error) {
        self.emit('error', error);
      }
    });
    zk.once('disconnected', function () {
      if (!zk.closed) {
        zk.close();
        self.connect();
        self.emit('zkReconnect');
      }
    });
    zk.on('error', function (err) {
      self.emit('error', err);
    });
  }

• example usage

  ...
  }
  
  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);
  
  this.connect();
  }
  
  util.inherits(BaseProducer, events.EventEmitter);
  
  BaseProducer.prototype.connect = function () {
  // emiter...
  var self = this;
  ...

function kafka-node.Client.prototype.createBroker (host, port, longpolling)

• description and source-code

  createBroker = function (host, port, longpolling) {
    var self = this;
    var socket;
    if (self.ssl) {
      socket = tls.connect(port, host, self.sslOptions);
    } else {
      socket = net.createConnection(port, host);
    }
    socket.addr = host + ':' + port;
    socket.host = host;
    socket.port = port;
    socket.socketId = this.nextSocketId();
    if (longpolling) socket.longpolling = true;
  
    socket.on('connect', function () {
      var lastError = this.error;
      this.error = null;
      if (lastError) {
        this.waiting = false;
        self.emit('reconnect');
      } else {
        self.emit('connect');
      }
    });
    socket.on('error', function (err) {
      this.error = err;
      self.emit('error', err);
    });
    socket.on('close', function (hadError) {
      self.emit('close', this);
      if (hadError && this.error) {
        self.clearCallbackQueue(this, this.error);
      } else {
        self.clearCallbackQueue(this, new errors.BrokerNotAvailableError('Broker not available'));
      }
      retry(this);
    });
    socket.on('end', function () {
      retry(this);
    });
    socket.buffer = new Buffer([]);
    socket.on('data', function (data) {
      this.buffer = Buffer.concat([this.buffer, data]);
      self.handleReceivedData(this);
    });
    socket.setKeepAlive(true, 60000);
  
    function retry (s) {
      if (s.retrying || s.closing) return;
      s.retrying = true;
      s.retryTimer = setTimeout(function () {
        if (s.closing) return;
        self.reconnectBroker(s);
      }, 1000);
    }
    return new BrokerWrapper(socket, this.noAckBatchOptions);
  }

• example usage

  ...
  
  Client.prototype.getBrokers = function (longpolling) {
  return longpolling ? this.longpollingBrokers : this.brokers;
  };
  
  Client.prototype.setupBroker = function (host, port, longpolling, brokers) {
  var brokerKey = host + ':' + port;
  brokers[brokerKey] = this.createBroker(host, port, longpolling);
  return brokers[brokerKey];
  };
  
  Client.prototype.createBroker = function (host, port, longpolling) {
  var self = this;
  var socket;
  if (self.ssl) {
  ...

function kafka-node.Client.prototype.createTopics (topics, isAsync, cb)

• description and source-code

  createTopics = function (topics, isAsync, cb) {
    topics = typeof topics === 'string' ? [topics] : topics;
  
    if (typeof isAsync === 'function' && typeof cb === 'undefined') {
      cb = isAsync;
      isAsync = true;
    }
  
    try {
      validateKafkaTopics(topics);
    } catch (e) {
      if (isAsync) return cb(e);
      throw e;
    }
  
    cb = _.once(cb);
  
    const getTopicsFromKafka = (topics, callback) => {
      this.loadMetadataForTopics(topics, function (error, resp) {
        if (error) {
          return callback(error);
        }
        callback(null, Object.keys(resp[1].metadata));
      });
    };
  
    const operation = retry.operation({ minTimeout: 200, maxTimeout: 2000 });
  
    operation.attempt(currentAttempt => {
      logger.debug('create topics currentAttempt', currentAttempt);
      getTopicsFromKafka(topics, function (error, kafkaTopics) {
        if (error) {
          if (operation.retry(error)) {
            return;
          }
        }
  
        logger.debug('kafka reported topics', kafkaTopics);
        const left = _.difference(topics, kafkaTopics);
        if (left.length === 0) {
          logger.debug(`Topics created ${kafkaTopics}`);
          return cb(null, kafkaTopics);
        }
  
        logger.debug(`Topics left ${left.join(', ')}`);
        if (!operation.retry(new Error(`Topics not created ${left}`))) {
          cb(operation.mainError());
        }
      });
    });
  
    if (!isAsync) {
      cb(null);
    }
  }

• example usage

  ...
  
  ``` js
  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
  ```
  ...

function kafka-node.Client.prototype.getBrokers (longpolling)

• description and source-code

  getBrokers = function (longpolling) {
    return longpolling ? this.longpollingBrokers : this.brokers;
  }

• example usage

  ...
  Client.prototype.leaderByPartition = function (topic, partition) {
  var topicMetadata = this.topicMetadata;
  return topicMetadata[topic] && topicMetadata[topic][partition] && topicMetadata[topic][partition].leader;
  };
  
  Client.prototype.brokerForLeader = function (leader, longpolling) {
  var addr;
  var brokers = this.getBrokers(longpolling);
  // If leader is not give, choose the first broker as leader
  if (typeof leader === 'undefined') {
    if (!_.isEmpty(brokers)) {
      addr = Object.keys(brokers)[0];
      return brokers[addr];
    } else if (!_.isEmpty(this.brokerMetadata)) {
      leader = Object.keys(this.brokerMetadata)[0];
  ...

function kafka-node.Client.prototype.handleReceivedData (socket)

• description and source-code

  handleReceivedData = function (socket) {
    var vars = Binary.parse(socket.buffer).word32bu('size').word32bu('correlationId').vars;
    var size = vars.size + 4;
    var correlationId = vars.correlationId;
  
    if (socket.buffer.length >= size) {
      var resp = socket.buffer.slice(0, size);
      var handlers = this.unqueueCallback(socket, correlationId);
  
      if (!handlers) return;
      var decoder = handlers[0];
      var cb = handlers[1];
      var result = decoder(resp);
      (result instanceof Error)
        ? cb.call(this, result)
        : cb.call(this, null, result);
      socket.buffer = socket.buffer.slice(size);
      if (socket.longpolling) socket.waiting = false;
    } else { return; }
  
    if (socket.buffer.length) {
      setImmediate(function () { this.handleReceivedData(socket); }.bind(this));
    }
  }

• example usage

  ...
  });
  socket.on('end', function () {
    retry(this);
  });
  socket.buffer = new Buffer([]);
  socket.on('data', function (data) {
    this.buffer = Buffer.concat([this.buffer, data]);
    self.handleReceivedData(this);
  });
  socket.setKeepAlive(true, 60000);
  
  function retry (s) {
    if (s.retrying || s.closing) return;
    s.retrying = true;
    s.retryTimer = setTimeout(function () {
  ...

function kafka-node.Client.prototype.hasMetadata (topic, partition)

• description and source-code

  hasMetadata = function (topic, partition) {
    var brokerMetadata = this.brokerMetadata;
    var leader = this.leaderByPartition(topic, partition);
  
    return (leader !== undefined) && brokerMetadata[leader];
  }

• example usage

  ...
  };
  
  Client.prototype.checkMetadatas = function (payloads) {
  if (_.isEmpty(this.topicMetadata)) return [ [], payloads ];
  // out: [ [metadata exists], [metadata not exists] ]
  var out = [ [], [] ];
  payloads.forEach(function (p) {
    if (this.hasMetadata(p.topic, p.partition)) out[0].push(p);
    else out[1].push(p);
  }.bind(this));
  return out;
  };
  
  Client.prototype.hasMetadata = function (topic, partition) {
  var brokerMetadata = this.brokerMetadata;
  ...

function kafka-node.Client.prototype.leaderByPartition (topic, partition)

• description and source-code

  leaderByPartition = function (topic, partition) {
    var topicMetadata = this.topicMetadata;
    return topicMetadata[topic] && topicMetadata[topic][partition] && topicMetadata[topic][partition].leader;
  }

• example usage

  ...
    else out[1].push(p);
  }.bind(this));
  return out;
  };
  
  Client.prototype.hasMetadata = function (topic, partition) {
  var brokerMetadata = this.brokerMetadata;
  var leader = this.leaderByPartition(topic, partition);
  
  return (leader !== undefined) && brokerMetadata[leader];
  };
  
  Client.prototype.updateMetadatas = function (metadatas) {
  // _.extend(this.brokerMetadata, metadatas[0])
  _.extend(this.topicMetadata, metadatas[1].metadata);
  ...

function kafka-node.Client.prototype.loadMetadataForTopics (topics, cb)

• description and source-code

  loadMetadataForTopics = function (topics, cb) {
    var correlationId = this.nextId();
    var request = protocol.encodeMetadataRequest(this.clientId, correlationId, topics);
    var broker = this.brokerForLeader();
  
    if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
      return cb(new errors.BrokerNotAvailableError('Broker not available'));
    }
  
    this.queueCallback(broker.socket, correlationId, [protocol.decodeMetadataResponse, cb]);
    broker.write(request);
  }

• example usage

  ...
  
  ## How do I get a list of all topics?
  
  Call `client.loadMetadataForTopics` with a blank topic array to get the entire list of available topics (and available brokers).
  
  ```js
  client.once('connect', function () {
  	client.loadMetadataForTopics([], function (error, results) {
  	  if (error) {
  	  	return console.error(error);
  	  }
  	  console.log('%j', _.get(results, '1.metadata'));
  	});
  });
  ```
  ...

function kafka-node.Client.prototype.nextId ()

• description and source-code

  nextId = function () {
    if (this.correlationId >= MAX_INT32) {
      this.correlationId = 0;
    }
    return this.correlationId++;
  }

• example usage

  ...
  var decoder = protocol.decodeOffsetResponse;
  this.send(payloads, encoder, decoder, cb);
  };
  
  Client.prototype.sendGroupRequest = function (encode, decode, requestArgs) {
  requestArgs = _.values(requestArgs);
  var cb = requestArgs.pop();
  var correlationId = this.nextId();
  
  requestArgs.unshift(this.clientId, correlationId);
  
  var request = encode.apply(null, requestArgs);
  var broker = this.brokerForLeader(this.coordinatorId);
  
  if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
  ...

function kafka-node.Client.prototype.nextSocketId ()

• description and source-code

  nextSocketId = function () {
    return this._socketId++;
  }

• example usage

  ...
    socket = tls.connect(port, host, self.sslOptions);
  } else {
    socket = net.createConnection(port, host);
  }
  socket.addr = host + ':' + port;
  socket.host = host;
  socket.port = port;
  socket.socketId = this.nextSocketId();
  if (longpolling) socket.longpolling = true;
  
  socket.on('connect', function () {
    var lastError = this.error;
    this.error = null;
    if (lastError) {
      this.waiting = false;
  ...

function kafka-node.Client.prototype.payloadsByLeader (payloads)

• description and source-code

  payloadsByLeader = function (payloads) {
    return payloads.reduce(function (out, p) {
      var leader = this.leaderByPartition(p.topic, p.partition);
      out[leader] = out[leader] || [];
      out[leader].push(p);
      return out;
    }.bind(this), {});
  }

• example usage

  ...
      self.sendToBroker(payloads[1].concat(payloads[0]), encoder, decoder, cb);
    });
  }
  };
  
  Client.prototype.sendToBroker = function (payloads, encoder, decoder, cb) {
  var longpolling = encoder.name === 'encodeFetchRequest';
  payloads = this.payloadsByLeader(payloads);
  if (!longpolling) {
    cb = wrap(payloads, cb);
  }
  for (var leader in payloads) {
    if (!payloads.hasOwnProperty(leader)) {
      continue;
    }
  ...

function kafka-node.Client.prototype.queueCallback (socket, id, data)

• description and source-code

  queueCallback = function (socket, id, data) {
    var socketId = socket.socketId;
    var queue;
  
    if (this.cbqueue.hasOwnProperty(socketId)) {
      queue = this.cbqueue[socketId];
    } else {
      queue = {};
      this.cbqueue[socketId] = queue;
    }
  
    queue[id] = data;
  }

• example usage

  ...
    var request = encode.apply(null, requestArgs);
    var broker = this.brokerForLeader(this.coordinatorId);
  
    if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
      return cb(new errors.BrokerNotAvailableError('Broker not available'));
    }
  
    this.queueCallback(broker.socket, correlationId, [decode, cb]);
    broker.write(request);
  };
  
  Client.prototype.sendGroupCoordinatorRequest = function (groupId, cb) {
    this.sendGroupRequest(protocol.encodeGroupCoordinatorRequest, protocol.decodeGroupCoordinatorResponse, arguments);
  };
  ...

function kafka-node.Client.prototype.reconnectBroker (oldSocket)

• description and source-code

  reconnectBroker = function (oldSocket) {
    oldSocket.retrying = false;
    if (oldSocket.error) {
      oldSocket.destroy();
    }
    var brokers = this.getBrokers(oldSocket.longpolling);
    var newBroker = this.setupBroker(oldSocket.host, oldSocket.port, oldSocket.longpolling, brokers);
    newBroker.socket.error = oldSocket.error;
  }

• example usage

  ...
  socket.setKeepAlive(true, 60000);
  
  function retry (s) {
    if (s.retrying || s.closing) return;
    s.retrying = true;
    s.retryTimer = setTimeout(function () {
      if (s.closing) return;
      self.reconnectBroker(s);
    }, 1000);
  }
  return new BrokerWrapper(socket, this.noAckBatchOptions);
  };
  
  Client.prototype.reconnectBroker = function (oldSocket) {
  oldSocket.retrying = false;
  ...

function kafka-node.Client.prototype.refreshBrokers ()

• description and source-code

  refreshBrokers = function () {
    var self = this;
    var validBrokers = Object.keys(this.brokerProfiles);
  
    function closeDeadBrokers (brokers) {
      var deadBrokerKeys = _.difference(Object.keys(brokers), validBrokers);
      if (deadBrokerKeys.length) {
        self.closeBrokers(deadBrokerKeys.map(function (key) {
          var broker = brokers[key];
          delete brokers[key];
          return broker;
        }));
      }
    }
  
    closeDeadBrokers(this.brokers);
    closeDeadBrokers(this.longpollingBrokers);
  }

• example usage

  ...
    }
  });
  zk.on('brokersChanged', function (brokerMetadata) {
    try {
      self.brokerMetadata = brokerMetadata;
      logger.debug('brokersChanged', brokerMetadata);
      self.setupBrokerProfiles(brokerMetadata);
      self.refreshBrokers();
      // Emit after a 3 seconds
      setTimeout(function () {
        self.emit('brokersChanged');
      }, 3000);
    } catch (error) {
      self.emit('error', error);
    }
  ...

function kafka-node.Client.prototype.refreshMetadata (topicNames, cb)

• description and source-code

  refreshMetadata = function (topicNames, cb) {
    var self = this;
    if (!topicNames.length) return cb();
    attemptRequestMetadata(topicNames, cb);
  
    function attemptRequestMetadata (topics, cb) {
      var operation = retry.operation({ minTimeout: 200, maxTimeout: 1000 });
      operation.attempt(function (currentAttempt) {
        logger.debug('refresh metadata currentAttempt', currentAttempt);
        self.loadMetadataForTopics(topics, function (err, resp) {
          err = err || resp[1].error;
          if (operation.retry(err)) {
            return;
          }
          if (err) {
            logger.debug('refresh metadata error', err.message);
            return cb(err);
          }
          self.updateMetadatas(resp);
          cb();
        });
      });
    }
  }

• example usage

  ...
  
  Error:
  
  ```
  BrokerNotAvailableError: Could not find the leader
  ```
  
  Call `client.refreshMetadata()` before sending the first message. Reference issue [#354
  ](https://github.com/SOHU-Co/kafka-node/issues/354)
  
  
  
  ## How do I debug an issue?
  This module uses the [debug module](https://github.com/visionmedia/debug) so you can just run below before starting your app.
  
  ```bash
  ...

function kafka-node.Client.prototype.removeTopicMetadata (topics, cb)

• description and source-code

  removeTopicMetadata = function (topics, cb) {
    topics.forEach(function (t) {
      if (this.topicMetadata[t]) delete this.topicMetadata[t];
    }.bind(this));
    cb(null, topics.length);
  }

• example usage

  ...
  
  Consumer.prototype.removeTopics = function (topics, cb) {
  topics = typeof topics === 'string' ? [topics] : topics;
  this.payloads = this.payloads.filter(function (p) {
    return !~topics.indexOf(p.topic);
  });
  
  this.client.removeTopicMetadata(topics, cb);
  };
  
  Consumer.prototype.close = function (force, cb) {
  this.ready = false;
  if (typeof force === 'function') {
    cb = force;
    force = false;
  ...

function kafka-node.Client.prototype.send (payloads, encoder, decoder, cb)

• description and source-code

  send = function (payloads, encoder, decoder, cb) {
    var self = this;
    var _payloads = payloads;
    // payloads: [ [metadata exists], [metadata not exists] ]
    payloads = this.checkMetadatas(payloads);
    if (payloads[0].length && !payloads[1].length) {
      this.sendToBroker(_.flatten(payloads), encoder, decoder, cb);
      return;
    }
    if (payloads[1].length) {
      var topicNames = payloads[1].map(function (p) { return p.topic; });
      this.loadMetadataForTopics(topicNames, function (err, resp) {
        if (err) {
          return cb(err);
        }
  
        var error = resp[1].error;
        if (error) {
          return cb(error);
        }
  
        self.updateMetadatas(resp);
        // check payloads again
        payloads = self.checkMetadatas(_payloads);
        if (payloads[1].length) {
          return cb(new errors.BrokerNotAvailableError('Could not find the leader'));
        }
  
        self.sendToBroker(payloads[1].concat(payloads[0]), encoder, decoder, cb);
      });
    }
  }

• example usage

  ...
      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 `messages` attribute otherwise
   you may lose messages!
  ...

function kafka-node.Client.prototype.sendFetchRequest (consumer, payloads, fetchMaxWaitMs, fetchMinBytes, maxTickMessages)

• description and source-code

  sendFetchRequest = function (consumer, payloads, fetchMaxWaitMs, fetchMinBytes, maxTickMessages) {
    var self = this;
    var encoder = protocol.encodeFetchRequest(fetchMaxWaitMs, fetchMinBytes);
    var decoder = protocol.decodeFetchResponse(function (err, type, message) {
      if (err) {
        if (err.message === 'OffsetOutOfRange') {
          return consumer.emit('offsetOutOfRange', err);
        } else if (err.message === 'NotLeaderForPartition' || err.message === 'UnknownTopicOrPartition') {
          return self.emit('brokersChanged');
        }
  
        return consumer.emit('error', err);
      }
  
      var encoding = consumer.options.encoding;
  
      if (type === 'message') {
        if (encoding !== 'buffer' && message.value) {
          message.value = message.value.toString(encoding);
        }
  
        consumer.emit('message', message);
      } else {
        consumer.emit('done', message);
      }
    }, maxTickMessages);
  
    this.send(payloads, encoder, decoder, function (err) {
      if (err) {
        Array.prototype.unshift.call(arguments, 'error');
        consumer.emit.apply(consumer, arguments);
      }
    });
  }

• example usage

  ...
      cb(null, 'Nothing to be committed');
    }
  }
  Consumer.prototype.commit = Consumer.prototype.autoCommit = autoCommit;
  
  Consumer.prototype.fetch = function () {
    if (!this.ready || this.paused) return;
    this.client.sendFetchRequest(this, this.payloads, this.options.fetchMaxWaitMs, this
  .options.fetchMinBytes);
  };
  
  Consumer.prototype.fetchOffset = function (payloads, cb) {
    this.client.sendOffsetFetchRequest(this.options.groupId, payloads, cb);
  };
  
  Consumer.prototype.addTopics = function (topics, cb, fromOffset) {
  ...

function kafka-node.Client.prototype.sendGroupCoordinatorRequest (groupId, cb)

• description and source-code

  sendGroupCoordinatorRequest = function (groupId, cb) {
    this.sendGroupRequest(protocol.encodeGroupCoordinatorRequest, protocol.decodeGroupCoordinatorResponse, arguments);
  }

• example usage

  ...
    this.emit('rebalancing');
  
    async.waterfall([
  function (callback) {
    if (self.client.coordinatorId) {
      return callback(null, null);
    }
    self.client.sendGroupCoordinatorRequest(self.options.groupId, callback);
  },
  
  function (coordinatorInfo, callback) {
    logger.debug('GroupCoordinator Response:', coordinatorInfo);
    if (coordinatorInfo) {
      self.setCoordinatorId(coordinatorInfo.coordinatorId);
    }
  ...

function kafka-node.Client.prototype.sendGroupRequest (encode, decode, requestArgs)

• description and source-code

  sendGroupRequest = function (encode, decode, requestArgs) {
    requestArgs = _.values(requestArgs);
    var cb = requestArgs.pop();
    var correlationId = this.nextId();
  
    requestArgs.unshift(this.clientId, correlationId);
  
    var request = encode.apply(null, requestArgs);
    var broker = this.brokerForLeader(this.coordinatorId);
  
    if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
      return cb(new errors.BrokerNotAvailableError('Broker not available'));
    }
  
    this.queueCallback(broker.socket, correlationId, [decode, cb]);
    broker.write(request);
  }

• example usage

  ...
    var decoder = protocol.decodeOffsetCommitResponse;
    this.send(payloads, encoder, decoder, cb);
  };
  
  Client.prototype.sendOffsetCommitV2Request = function (group, generationId, memberId, payloads, cb) {
    var encoder = protocol.encodeOffsetCommitV2Request;
    var decoder = protocol.decodeOffsetCommitResponse;
    this.sendGroupRequest(encoder, decoder, arguments);
  };
  
  Client.prototype.sendOffsetFetchV1Request = function (group, payloads, cb) {
    var encoder = protocol.encodeOffsetFetchV1Request;
    var decoder = protocol.decodeOffsetFetchV1Response;
    this.sendGroupRequest(encoder, decoder, arguments);
  };
  ...

function kafka-node.Client.prototype.sendHeartbeatRequest (groupId, generationId, memberId, cb)

• description and source-code

  sendHeartbeatRequest = function (groupId, generationId, memberId, cb) {
    this.sendGroupRequest(protocol.encodeGroupHeartbeat, protocol.decodeGroupHeartbeat, arguments);
  }

• example usage

  ...
  constructor (client, handler) {
    this.client = client;
    this.handler = handler;
    this.pending = true;
  }
  
  send (groupId, generationId, memberId) {
    this.client.sendHeartbeatRequest(groupId, generationId, memberId, (error) => {
      if (this.canceled) {
        logger.debug('heartbeat yielded after being canceled', error);
        return;
      }
      this.pending = false;
      this.handler(error);
    });
  ...

function kafka-node.Client.prototype.sendJoinGroupRequest (groupId, memberId, sessionTimeout, groupProtocol, cb)

• description and source-code

  sendJoinGroupRequest = function (groupId, memberId, sessionTimeout, groupProtocol, cb) {
    this.sendGroupRequest(protocol.encodeJoinGroupRequest, protocol.decodeJoinGroupResponse, arguments);
  }

• example usage

  ...
  },
  
  function (coordinatorInfo, callback) {
    logger.debug('GroupCoordinator Response:', coordinatorInfo);
    if (coordinatorInfo) {
      self.setCoordinatorId(coordinatorInfo.coordinatorId);
    }
    self.client.sendJoinGroupRequest(self.options.groupId, emptyStrIfNull(self.memberId
  ), self.options.sessionTimeout, self.protocols, callback);
  },
  
  function (joinGroupResponse, callback) {
    self.handleJoinGroup(joinGroupResponse, callback);
  },
  
  function (groupAssignment, callback) {
  ...

function kafka-node.Client.prototype.sendLeaveGroupRequest (groupId, memberId, cb)

• description and source-code

  sendLeaveGroupRequest = function (groupId, memberId, cb) {
    this.sendGroupRequest(protocol.encodeLeaveGroupRequest, protocol.decodeLeaveGroupResponse, arguments);
  }

• example usage

  ...
  };
  
  ConsumerGroup.prototype.leaveGroup = function (callback) {
    logger.debug('%s leaving group', this.client.clientId);
    var self = this;
    this.stopHeartbeats();
    if (self.generationId != null && self.memberId) {
      this.client.sendLeaveGroupRequest(this.options.groupId, this.memberId, function (
  error) {
        self.generationId = null;
        callback(error);
      });
    } else {
      callback(null);
    }
  };
  ...

function kafka-node.Client.prototype.sendOffsetCommitRequest (group, payloads, cb)

• description and source-code

  sendOffsetCommitRequest = function (group, payloads, cb) {
    var encoder = protocol.encodeOffsetCommitRequest(group);
    var decoder = protocol.decodeOffsetCommitResponse;
    this.send(payloads, encoder, decoder, cb);
  }

• example usage

  ...
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var payloads = this.payloads;
    if (this.pausedPayloads) payloads = payloads.concat(this.pausedPayloads);
  
    var commits = payloads.filter(function (p) { return p.offset !== 0; });
    if (commits.length) {
      this.client.sendOffsetCommitRequest(this.options.groupId, commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }
  Consumer.prototype.commit = Consumer.prototype.autoCommit = autoCommit;
  
  Consumer.prototype.fetch = function () {
  ...

function kafka-node.Client.prototype.sendOffsetCommitV2Request (group, generationId, memberId, payloads, cb)

• description and source-code

  sendOffsetCommitV2Request = function (group, generationId, memberId, payloads, cb) {
    var encoder = protocol.encodeOffsetCommitV2Request;
    var decoder = protocol.decodeOffsetCommitResponse;
    this.sendGroupRequest(encoder, decoder, arguments);
  }

• example usage

  ...
  
  ConsumerGroup.prototype.fetchOffset = function (payloads, cb) {
  this.client.sendOffsetFetchV1Request(this.options.groupId, payloads, cb);
  };
  
  ConsumerGroup.prototype.sendOffsetCommitRequest = function (commits, cb) {
  if (this.generationId && this.memberId) {
    this.client.sendOffsetCommitV2Request(this.options.groupId, this.generationId, this
  .memberId, commits, cb);
  } else {
    cb(null, 'Nothing to be committed');
  }
  };
  
  ConsumerGroup.prototype.close = function (force, cb) {
  var self = this;
  ...

function kafka-node.Client.prototype.sendOffsetFetchRequest (group, payloads, cb)

• description and source-code

  sendOffsetFetchRequest = function (group, payloads, cb) {
    var encoder = protocol.encodeOffsetFetchRequest(group);
    var decoder = protocol.decodeOffsetFetchResponse;
    this.send(payloads, encoder, decoder, cb);
  }

• example usage

  ...
      }
    }
  );
  };
  
  ConsumerGroupMigrator.prototype.saveHighLevelConsumerOffsets = function (topicPartitionList, callback) {
  const self = this;
  this.client.sendOffsetFetchRequest(this.consumerGroup.options.groupId, topicPartitionList
  , function (error, results) {
    logger.debug('sendOffsetFetchRequest response:', results, error);
    if (error) {
      return callback(error);
    }
    self.offsets = results;
    callback(null);
  });
  ...

function kafka-node.Client.prototype.sendOffsetFetchV1Request (group, payloads, cb)

• description and source-code

  sendOffsetFetchV1Request = function (group, payloads, cb) {
    var encoder = protocol.encodeOffsetFetchV1Request;
    var decoder = protocol.decodeOffsetFetchV1Response;
    this.sendGroupRequest(encoder, decoder, arguments);
  }

• example usage

  ...
  const heartbeat = new Heartbeat(this.client, heartbeatCallback);
  heartbeat.send(this.options.groupId, this.generationId, this.memberId);
  
  return heartbeat;
  };
  
  ConsumerGroup.prototype.fetchOffset = function (payloads, cb) {
  this.client.sendOffsetFetchV1Request(this.options.groupId, payloads, cb);
  };
  
  ConsumerGroup.prototype.sendOffsetCommitRequest = function (commits, cb) {
  if (this.generationId && this.memberId) {
    this.client.sendOffsetCommitV2Request(this.options.groupId, this.generationId, this.memberId, commits, cb);
  } else {
    cb(null, 'Nothing to be committed');
  ...

function kafka-node.Client.prototype.sendOffsetRequest (payloads, cb)

• description and source-code

  sendOffsetRequest = function (payloads, cb) {
    var encoder = protocol.encodeOffsetRequest;
    var decoder = protocol.decodeOffsetResponse;
    this.send(payloads, encoder, decoder, cb);
  }

• example usage

  ...
  
  HighLevelConsumer.prototype.fetchOffset = function (payloads, cb) {
   logger.debug('in fetchOffset %s payloads: %j', this.id, payloads);
   this.client.sendOffsetFetchRequest(this.options.groupId, payloads, cb);
  };
  
  HighLevelConsumer.prototype.offsetRequest = function (payloads, cb) {
   this.client.sendOffsetRequest(payloads, cb);
  };
  
  /**
  * Register a consumer against a group
  *
  * @param consumer to register
  *
  ...

function kafka-node.Client.prototype.sendProduceRequest (payloads, requireAcks, ackTimeoutMs, cb)

• description and source-code

  sendProduceRequest = function (payloads, requireAcks, ackTimeoutMs, cb) {
    var encoder = protocol.encodeProduceRequest(requireAcks, ackTimeoutMs);
    var decoder = protocol.decodeProduceResponse;
    var self = this;
  
    decoder.requireAcks = requireAcks;
  
    async.each(payloads, buildRequest, function (err) {
      if (err) return cb(err);
      self.send(payloads, encoder, decoder, function (err, result) {
        if (err) {
          if (err.message === 'NotLeaderForPartition') {
            self.emit('brokersChanged');
          }
          cb(err);
        } else {
          cb(null, result);
        }
      });
    });
  
    function buildRequest (payload, cb) {
      var attributes = payload.attributes;
      var codec = getCodec(attributes);
  
      if (!codec) return cb();
  
      var innerSet = encodeMessageSet(payload.messages);
      codec.encode(innerSet, function (err, message) {
        if (err) return cb(err);
        payload.messages = [ new Message(0, attributes, '', message) ];
        cb();
      });
    }
  }

• example usage

  ...
   * @param {BaseProducer~sendCallback} cb A function to call once the send has completed
   */
  BaseProducer.prototype.send = function (payloads, cb) {
  var client = this.client;
  var requireAcks = this.requireAcks;
  var ackTimeoutMs = this.ackTimeoutMs;
  
  client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks
  , ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
  const topicPartitionRequests = Object.create(null);
  payloads.forEach((p) => {
    p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition(_.map(topicMetadata[p.topic
  ], 'partition'), p.key);
    p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  ...

function kafka-node.Client.prototype.sendSyncGroupRequest (groupId, generationId, memberId, groupAssignment, cb)

• description and source-code

  sendSyncGroupRequest = function (groupId, generationId, memberId, groupAssignment, cb) {
    this.sendGroupRequest(protocol.encodeSyncGroupRequest, protocol.decodeSyncGroupResponse, arguments);
  }

• example usage

  ...
  
    function (joinGroupResponse, callback) {
      self.handleJoinGroup(joinGroupResponse, callback);
    },
  
    function (groupAssignment, callback) {
      logger.debug('SyncGroup Request from %s', self.memberId);
      self.client.sendSyncGroupRequest(self.options.groupId, self.generationId, self.memberId
  , groupAssignment, callback);
    },
  
    function (syncGroupResponse, callback) {
      self.handleSyncGroup(syncGroupResponse, callback);
    }
  ], function (error, startFetch) {
    self.connecting = false;
  ...

function kafka-node.Client.prototype.sendToBroker (payloads, encoder, decoder, cb)

• description and source-code

  sendToBroker = function (payloads, encoder, decoder, cb) {
    var longpolling = encoder.name === 'encodeFetchRequest';
    payloads = this.payloadsByLeader(payloads);
    if (!longpolling) {
      cb = wrap(payloads, cb);
    }
    for (var leader in payloads) {
      if (!payloads.hasOwnProperty(leader)) {
        continue;
      }
      var correlationId = this.nextId();
      var request = encoder(this.clientId, correlationId, payloads[leader]);
      var broker = this.brokerForLeader(leader, longpolling);
      if (!broker || !broker.socket || broker.socket.error || broker.socket.closing || broker.socket.destroyed) {
        return cb(new errors.BrokerNotAvailableError('Could not find the leader'), payloads[leader]);
      }
  
      if (longpolling) {
        if (broker.socket.waiting) continue;
        broker.socket.waiting = true;
      }
  
      if (decoder.requireAcks === 0) {
        broker.writeAsync(request);
        cb(null, { result: 'no ack' });
      } else {
        this.queueCallback(broker.socket, correlationId, [decoder, cb]);
        broker.write(request);
      }
    }
  }

• example usage

  ...
  
  Client.prototype.send = function (payloads, encoder, decoder, cb) {
  var self = this;
  var _payloads = payloads;
  // payloads: [ [metadata exists], [metadata not exists] ]
  payloads = this.checkMetadatas(payloads);
  if (payloads[0].length && !payloads[1].length) {
    this.sendToBroker(_.flatten(payloads), encoder, decoder, cb);
    return;
  }
  if (payloads[1].length) {
    var topicNames = payloads[1].map(function (p) { return p.topic; });
    this.loadMetadataForTopics(topicNames, function (err, resp) {
      if (err) {
        return cb(err);
  ...

function kafka-node.Client.prototype.setupBroker (host, port, longpolling, brokers)

• description and source-code

  setupBroker = function (host, port, longpolling, brokers) {
    var brokerKey = host + ':' + port;
    brokers[brokerKey] = this.createBroker(host, port, longpolling);
    return brokers[brokerKey];
  }

• example usage

  ...
    self.ready = true;
    self.brokerMetadata = brokers;
    self.setupBrokerProfiles(brokers);
    Object
        .keys(self.brokerProfiles)
        .some(function (key, index) {
          var broker = self.brokerProfiles[key];
          self.setupBroker(broker.host, broker.port, false, self.brokers);
          // Only connect one broker
          return !index;
        });
    self.emit('ready');
  } catch (error) {
    self.ready = false;
    self.emit('error', error);
  ...

function kafka-node.Client.prototype.setupBrokerProfiles (brokers)

• description and source-code

  setupBrokerProfiles = function (brokers) {
    this.brokerProfiles = Object.create(null);
    var self = this;
    var protocol = self.ssl ? 'ssl:' : 'plaintext:';
  
    Object.keys(brokers).forEach(function (key) {
      var brokerProfile = brokers[key];
      var addr;
  
      if (brokerProfile.endpoints && brokerProfile.endpoints.length) {
        var endpoint = _.find(brokerProfile.endpoints, function (endpoint) {
          return url.parse(endpoint).protocol === protocol;
        });
  
        if (endpoint == null) {
          throw new Error(['No kafka endpoint found for broker: ', key, ' with protocol ', protocol].join(''));
        }
  
        var endpointUrl = url.parse(endpoint);
  
        addr = endpointUrl.hostname + ':' + endpointUrl.port;
  
        brokerProfile.host = endpointUrl.hostname;
        brokerProfile.port = endpointUrl.port;
      } else {
        addr = brokerProfile.host + ':' + brokerProfile.port;
      }
      assert(brokerProfile.host && brokerProfile.port, 'kafka host or port is empty');
  
      self.brokerProfiles[addr] = brokerProfile;
      self.brokerProfiles[addr].id = key;
    });
  }

• example usage

  ...
  Client.prototype.connect = function () {
  var zk = this.zk = new Zookeeper(this.connectionString, this.zkOptions);
  var self = this;
  zk.once('init', function (brokers) {
    try {
      self.ready = true;
      self.brokerMetadata = brokers;
      self.setupBrokerProfiles(brokers);
      Object
          .keys(self.brokerProfiles)
          .some(function (key, index) {
            var broker = self.brokerProfiles[key];
            self.setupBroker(broker.host, broker.port, false, self.brokers);
            // Only connect one broker
            return !index;
  ...

function kafka-node.Client.prototype.topicExists (topics, cb)

• description and source-code

  topicExists = function (topics, cb) {
    var notExistsTopics = [];
    var self = this;
  
    async.each(topics, checkZK, function (err) {
      if (err) return cb(err);
      if (notExistsTopics.length) return cb(new errors.TopicsNotExistError(notExistsTopics));
      cb();
    });
  
    function checkZK (topic, cb) {
      self.zk.topicExists(topic, function (err, existed, topic) {
        if (err) return cb(err);
        if (!existed) notExistsTopics.push(topic);
        cb();
      });
    }
  }

• example usage

  ...
    async.each(topics, checkZK, function (err) {
      if (err) return cb(err);
      if (notExistsTopics.length) return cb(new errors.TopicsNotExistError(notExistsTopics));
      cb();
    });
  
    function checkZK (topic, cb) {
      self.zk.topicExists(topic, function (err, existed, topic) {
        if (err) return cb(err);
        if (!existed) notExistsTopics.push(topic);
        cb();
      });
    }
  };
  ...

function kafka-node.Client.prototype.unqueueCallback (socket, id)

• description and source-code

  unqueueCallback = function (socket, id) {
    var socketId = socket.socketId;
  
    if (!this.cbqueue.hasOwnProperty(socketId)) {
      return null;
    }
  
    var queue = this.cbqueue[socketId];
    if (!queue.hasOwnProperty(id)) {
      return null;
    }
  
    var result = queue[id];
  
    // cleanup socket queue
    delete queue[id];
    if (!Object.keys(queue).length) {
      delete this.cbqueue[socketId];
    }
  
    return result;
  }

• example usage

  ...
  Client.prototype.handleReceivedData = function (socket) {
    var vars = Binary.parse(socket.buffer).word32bu('size').word32bu('correlationId').vars;
    var size = vars.size + 4;
    var correlationId = vars.correlationId;
  
    if (socket.buffer.length >= size) {
  var resp = socket.buffer.slice(0, size);
  var handlers = this.unqueueCallback(socket, correlationId);
  
  if (!handlers) return;
  var decoder = handlers[0];
  var cb = handlers[1];
  var result = decoder(resp);
  (result instanceof Error)
    ? cb.call(this, result)
  ...

function kafka-node.Client.prototype.updateMetadatas (metadatas)

• description and source-code

  updateMetadatas = function (metadatas) {
    // _.extend(this.brokerMetadata, metadatas[0])
    _.extend(this.topicMetadata, metadatas[1].metadata);
    for (var topic in this.topicMetadata) {
      if (!this.topicMetadata.hasOwnProperty(topic)) {
        continue;
      }
      this.topicPartitions[topic] = Object.keys(this.topicMetadata[topic]).map(function (val) {
        return parseInt(val, 10);
      });
    }
  }

• example usage

  ...
  
  Client.prototype.addTopics = function (topics, cb) {
  var self = this;
  this.topicExists(topics, function (err) {
    if (err) return cb(err);
    self.loadMetadataForTopics(topics, function (err, resp) {
      if (err) return cb(err);
      self.updateMetadatas(resp);
      cb(null, topics);
    });
  });
  };
  
  Client.prototype.nextId = function () {
  if (this.correlationId >= MAX_INT32) {
  ...

module kafka-node.Consumer

function kafka-node.Consumer (client, topics, options)

• description and source-code

  Consumer = function (client, topics, options) {
    if (!topics) {
      throw new Error('Must have payloads');
    }
  
    utils.validateTopics(topics);
  
    this.fetchCount = 0;
    this.client = client;
    this.options = _.defaults((options || {}), DEFAULTS);
    this.ready = false;
    this.paused = this.options.paused;
    this.id = nextId();
    this.payloads = this.buildPayloads(topics);
    this.connect();
    this.encoding = this.options.encoding;
  
    if (this.options.groupId) {
      utils.validateConfig('options.groupId', this.options.groupId);
    }
  }

• example usage

  n/a

function kafka-node.Consumer.super_ ()

• description and source-code

  function EventEmitter() {
    EventEmitter.init.call(this);
  }

• example usage

  n/a

module kafka-node.Consumer.prototype

function kafka-node.Consumer.prototype.addTopics (topics, cb, fromOffset)

• description and source-code

  addTopics = function (topics, cb, fromOffset) {
    fromOffset = !!fromOffset;
    var self = this;
    if (!this.ready) {
      setTimeout(function () {
        self.addTopics(topics, cb, fromOffset);
      }
      , 100);
      return;
    }
  
    // The default is that the topics is a string array of topic names
    var topicNames = topics;
  
    // If the topics is actually an object and not string we assume it is an array of payloads
    if (typeof topics[0] === 'object') {
      topicNames = topics.map(function (p) { return p.topic; });
    }
  
    this.client.addTopics(
      topicNames,
      function (err, added) {
        if (err) return cb && cb(err, added);
  
        var payloads = self.buildPayloads(topics);
        var reFetch = !self.payloads.length;
  
        if (fromOffset) {
          payloads.forEach(function (p) {
            self.payloads.push(p);
          });
          if (reFetch) self.fetch();
          cb && cb(null, added);
          return;
        }
  
        // update offset of topics that will be added
        self.fetchOffset(payloads, function (err, offsets) {
          if (err) return cb(err);
          payloads.forEach(function (p) {
            var offset = offsets[p.topic][p.partition];
            if (offset === -1) offset = 0;
            p.offset = offset;
            self.payloads.push(p);
          });
          if (reFetch) self.fetch();
          cb && cb(null, added);
        });
      }
    );
  }

• example usage

  ...
  * `topics`: **Array**, array of topics to add
  * `cb`: **Function**,the callback
  * `fromOffset`: **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:
  
  ``` js
  consumer.addTopics(['t1', 't2'], function (err, added) {
  });
  
  or
  
  consumer.addTopics([{ topic: 't1', offset: 10 }], function (err, added) {
  }, true);
  ```
  ...

function kafka-node.Consumer.prototype.autoCommit (force, cb)

• description and source-code

  function autoCommit(force, cb) {
    if (arguments.length === 1) {
      cb = force;
      force = false;
    }
  
    if (this.committing && !force) return cb(null, 'Offset committing');
  
    this.committing = true;
    setTimeout(function () {
      this.committing = false;
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var payloads = this.payloads;
    if (this.pausedPayloads) payloads = payloads.concat(this.pausedPayloads);
  
    var commits = payloads.filter(function (p) { return p.offset !== 0; });
    if (commits.length) {
      this.client.sendOffsetCommitRequest(this.options.groupId, commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }

• example usage

  ...
      if (offset === -1) offset = 0;
      if (!initing) p.offset = offset + 1;
      else p.offset = offset;
    }
  });
  
  if (this.options.autoCommit && !initing) {
    this.autoCommit(false, function (err) {
      err && logger.debug('auto commit offset', err);
    });
  }
  };
  
  function autoCommit (force, cb) {
  if (arguments.length === 1) {
  ...

function kafka-node.Consumer.prototype.buildPayloads (payloads)

• description and source-code

  buildPayloads = function (payloads) {
    var self = this;
    return payloads.map(function (p) {
      if (typeof p !== 'object') p = { topic: p };
      p.partition = p.partition || 0;
      p.offset = p.offset || 0;
      p.maxBytes = self.options.fetchMaxBytes;
      p.metadata = 'm'; // metadata can be arbitrary
      return p;
    });
  }

• example usage

  ...
   * @param {BaseProducer~sendCallback} cb A function to call once the send has completed
   */
  BaseProducer.prototype.send = function (payloads, cb) {
  var client = this.client;
  var requireAcks = this.requireAcks;
  var ackTimeoutMs = this.ackTimeoutMs;
  
  client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks
  , ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
  const topicPartitionRequests = Object.create(null);
  payloads.forEach((p) => {
    p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition(_.map(topicMetadata[p.topic
  ], 'partition'), p.key);
    p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  ...

function kafka-node.Consumer.prototype.close (force, cb)

• description and source-code

  close = function (force, cb) {
    this.ready = false;
    if (typeof force === 'function') {
      cb = force;
      force = false;
    }
  
    if (force) {
      this.commit(force, function (err) {
        if (err) {
          return cb(err);
        }
        this.client.close(cb);
      }.bind(this));
    } else {
      this.client.close(cb);
    }
  }

• example usage

  ...
  
  ### close(force, cb)
  * `force`: **Boolean**, if set to true, it forces the consumer to commit the current offset before closing, default `false`
  
  Example
  
  ```js
  consumer.close(true, cb);
  consumer.close(cb); //force is disabled
  ```
  
  ## HighLevelConsumer
  ⚠️ ***This consumer has been deprecated in the latest version of Kafka (0.10.1) and is likely to be removed in the future. Please
   use the ConsumerGroup instead.***
  
  ### HighLevelConsumer(client, payloads, options)
  ...

function kafka-node.Consumer.prototype.commit (force, cb)

• description and source-code

  function autoCommit(force, cb) {
    if (arguments.length === 1) {
      cb = force;
      force = false;
    }
  
    if (this.committing && !force) return cb(null, 'Offset committing');
  
    this.committing = true;
    setTimeout(function () {
      this.committing = false;
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var payloads = this.payloads;
    if (this.pausedPayloads) payloads = payloads.concat(this.pausedPayloads);
  
    var commits = payloads.filter(function (p) { return p.offset !== 0; });
    if (commits.length) {
      this.client.sendOffsetCommitRequest(this.options.groupId, commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }

• example usage

  ...
  Commit offset of the current topics manually, this method should be called when a consumer leaves
  
  * `cb`: **Function**, the callback
  
  Example:
  
  ``` js
  consumer.commit(function(err, data) {
  });
  ```
  
  ### setOffset(topic, partition, offset)
  Set offset of the given topic
  
  * `topic`: **String**
  ...

function kafka-node.Consumer.prototype.connect ()

• description and source-code

  connect = function () {
    var self = this;
    // Client already exists
    this.ready = this.client.ready;
    if (this.ready) this.init();
  
    this.client.on('ready', function () {
      logger.debug('consumer ready');
      if (!self.ready) self.init();
      self.ready = true;
    });
  
    this.client.on('error', function (err) {
      logger.error('client error %s', err.message);
      self.emit('error', err);
    });
  
    this.client.on('close', function () {
      logger.debug('connection closed');
    });
  
    this.client.on('brokersChanged', function () {
      var topicNames = self.payloads.map(function (p) {
        return p.topic;
      });
  
      this.refreshMetadata(topicNames, function (err) {
        if (err) return self.emit('error', err);
        self.fetch();
      });
    });
    // 'done' will be emit when a message fetch request complete
    this.on('done', function (topics) {
      self.updateOffsets(topics);
      setImmediate(function () {
        self.fetch();
      });
    });
  }

• example usage

  ...
  }
  
  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);
  
  this.connect();
  }
  
  util.inherits(BaseProducer, events.EventEmitter);
  
  BaseProducer.prototype.connect = function () {
  // emiter...
  var self = this;
  ...

function kafka-node.Consumer.prototype.fetch ()

• description and source-code

  fetch = function () {
    if (!this.ready || this.paused) return;
    this.client.sendFetchRequest(this, this.payloads, this.options.fetchMaxWaitMs, this.options.fetchMinBytes);
  }

• example usage

  ...
  
  Example
  
  ```js
  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] } }
      });
  ```
  ...

function kafka-node.Consumer.prototype.fetchOffset (payloads, cb)

• description and source-code

  fetchOffset = function (payloads, cb) {
    this.client.sendOffsetFetchRequest(this.options.groupId, payloads, cb);
  }

• example usage

  ...
      logger.debug('%s owns topics: ', self.client.clientId, syncGroupResponse.partitions);
  
      const topicPartitionList = createTopicPartitionList(syncGroupResponse.partitions);
      const useDefaultOffsets = self.options.fromOffset in ACCEPTED_FROM_OFFSET;
  
      async.waterfall([
        function (callback) {
  self.fetchOffset(syncGroupResponse.partitions, callback);
        },
        function (offsets, callback) {
  logger.debug('%s fetchOffset Response: %j', self.client.clientId, offsets);
  
  var noOffset = topicPartitionList.some(function (tp) {
    return offsets[tp.topic][tp.partition] === -1;
  });
  ...

function kafka-node.Consumer.prototype.init ()

• description and source-code

  init = function () {
    if (!this.payloads.length) {
      return;
    }
  
    var self = this;
    var topics = self.payloads.map(function (p) { return p.topic; });
  
    self.client.topicExists(topics, function (err) {
      if (err) {
        return self.emit('error', err);
      }
  
      if (self.options.fromOffset) {
        return self.fetch();
      }
  
      self.fetchOffset(self.payloads, function (err, topics) {
        if (err) {
          return self.emit('error', err);
        }
  
        self.updateOffsets(topics, true);
        self.fetch();
      });
    });
  }

• example usage

  ...
  });
  };
  
  Consumer.prototype.connect = function () {
  var self = this;
  // Client already exists
  this.ready = this.client.ready;
  if (this.ready) this.init();
  
  this.client.on('ready', function () {
    logger.debug('consumer ready');
    if (!self.ready) self.init();
    self.ready = true;
  });
  ...

function kafka-node.Consumer.prototype.pause ()

• description and source-code

  pause = function () {
    this.paused = true;
  }

• example usage

  ...
      } else if (!self.connecting) {
        self.fetch();
      }
    });
  }, 200);
  
  this.client.on('brokersChanged', function () {
    self.pause();
    recoverFromBrokerChange();
  });
  
  this.client.on('reconnect', function (lastError) {
    self.fetch();
  });
  ...

function kafka-node.Consumer.prototype.pauseTopics (topics)

• description and source-code

  pauseTopics = function (topics) {
    if (!this.pausedPayloads) this.pausedPayloads = [];
    pauseOrResume(this.payloads, this.pausedPayloads, topics);
  }

• example usage

  ...
  ### resume()
  Resume the consumer. Resumes the fetch loop.
  
  ### pauseTopics(topics)
  Pause specify topics
  
  ```
  consumer.pauseTopics([
      'topic1',
      { topic: 'topic2', partition: 0 }
  ]);
  ```
  
  ### resumeTopics(topics)
  Resume specify topics
  ...

function kafka-node.Consumer.prototype.removeTopics (topics, cb)

• description and source-code

  removeTopics = function (topics, cb) {
    topics = typeof topics === 'string' ? [topics] : topics;
    this.payloads = this.payloads.filter(function (p) {
      return !~topics.indexOf(p.topic);
    });
  
    this.client.removeTopicMetadata(topics, cb);
  }

• example usage

  ...
  ### removeTopics(topics, cb)
  * `topics`: **Array**, array of topics to remove
  * `cb`: **Function**, the callback
  
  Example:
  
  ``` js
  consumer.removeTopics(['t1', 't2'], function (err, removed) {
  });
  ```
  
  ### commit(cb)
  Commit offset of the current topics manually, this method should be called when a consumer leaves
  
  * `cb`: **Function**, the callback
  ...

function kafka-node.Consumer.prototype.resume ()

• description and source-code

  resume = function () {
    this.paused = false;
    this.fetch();
  }

• example usage

  ...
      }
      const offset = _.head(result[topic.topic][topic.partition]);
      const oldOffset = _.find(this.topicPayloads, {topic: topic.topic, partition: topic.partition}).offset;
  
      logger.debug('replacing %s-%s stale offset of %d with %d', topic.topic, topic.partition, oldOffset, offset);
  
      this.setOffset(topic.topic, topic.partition, offset);
      this.resume();
    });
  });
  
  // 'done' will be emit when a message fetch request complete
  this.on('done', function (topics) {
    self.updateOffsets(topics);
    if (!self.paused) {
  ...

function kafka-node.Consumer.prototype.resumeTopics (topics)

• description and source-code

  resumeTopics = function (topics) {
    if (!this.pausedPayloads) this.pausedPayloads = [];
    var reFetch = !this.payloads.length;
    pauseOrResume(this.pausedPayloads, this.payloads, topics);
    reFetch = reFetch && this.payloads.length;
    if (reFetch) this.fetch();
  }

• example usage

  ...
  ]);
  ```
  
  ### resumeTopics(topics)
  Resume specify topics
  
  ```
  consumer.resumeTopics([
      'topic1',
      { topic: 'topic2', partition: 0 }
  ]);
  ```
  
  ### close(force, cb)
  * `force`: **Boolean**, if set to true, it forces the consumer to commit the current offset before closing, default `false`
  ...

function kafka-node.Consumer.prototype.setOffset (topic, partition, offset)

• description and source-code

  setOffset = function (topic, partition, offset) {
    this.payloads.every(function (p) {
      if (p.topic === topic && p.partition == partition) { // eslint-disable-line eqeqeq
        p.offset = offset;
        return false;
      }
      return true;
    });
  }

• example usage

  ...
  * `partition`: **Number**
  
  * `offset`: **Number**
  
  Example:
  
  ``` js
  consumer.setOffset('topic', 0, 0);
  ```
  
  ### pause()
  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()
  Resume the consumer. Resumes the fetch loop.
  ...

function kafka-node.Consumer.prototype.updateOffsets (topics, initing)

• description and source-code

  updateOffsets = function (topics, initing) {
    this.payloads.forEach(function (p) {
      if (!_.isEmpty(topics[p.topic]) && topics[p.topic][p.partition] !== undefined) {
        var offset = topics[p.topic][p.partition];
        if (offset === -1) offset = 0;
        if (!initing) p.offset = offset + 1;
        else p.offset = offset;
      }
    });
  
    if (this.options.autoCommit && !initing) {
      this.autoCommit(false, function (err) {
        err && logger.debug('auto commit offset', err);
      });
    }
  }

• example usage

  ...
      this.setOffset(topic.topic, topic.partition, offset);
      this.resume();
    });
  });
  
  // 'done' will be emit when a message fetch request complete
  this.on('done', function (topics) {
    self.updateOffsets(topics);
    if (!self.paused) {
      setImmediate(function () {
        self.fetch();
      });
    }
  });
  ...

module kafka-node.ConsumerGroup

function kafka-node.ConsumerGroup (memberOptions, topics)

• description and source-code

  function ConsumerGroup(memberOptions, topics) {
    EventEmitter.call(this);
    const self = this;
    this.options = _.defaults((memberOptions || {}), DEFAULTS);
  
    if (!this.options.heartbeatInterval) {
      this.options.heartbeatInterval = Math.floor(this.options.sessionTimeout / 3);
    }
  
    if (memberOptions.ssl === true) {
      memberOptions.ssl = {};
    }
  
    if (!(this.options.fromOffset in ACCEPTED_FROM_OFFSET)) {
      throw new Error(`fromOffset ${this.options.fromOffset} should be either: ${Object.keys(ACCEPTED_FROM_OFFSET).join(', ')}`);
    }
  
    if (!(this.options.outOfRangeOffset in ACCEPTED_FROM_OFFSET)) {
      throw new Error(`outOfRangeOffset ${this.options.outOfRangeOffset} should be either: ${Object.keys(ACCEPTED_FROM_OFFSET).join
  (', ')}`);
    }
  
    this.client = new Client(memberOptions.host, memberOptions.id, memberOptions.zk,
      memberOptions.batch, memberOptions.ssl);
  
    if (_.isString(topics)) {
      topics = [topics];
    }
  
    assert(Array.isArray(topics), 'Array of topics is required');
  
    this.topics = topics;
  
    this.recovery = new ConsumerGroupRecovery(this);
  
    this.setupProtocols(this.options.protocol);
  
    if (this.options.connectOnReady && !this.options.migrateHLC) {
      this.client.once('ready', this.connect.bind(this));
    }
  
    if (this.options.migrateHLC) {
      const ConsumerGroupMigrator = require('./consumerGroupMigrator');
      this.migrator = new ConsumerGroupMigrator(this);
      this.migrator.on('error', function (error) {
        self.emit('error', error);
      });
    }
  
    this.client.on('error', function (err) {
      logger.error('Error from %s', self.client.clientId, err);
      self.emit('error', err);
    });
  
    const recoverFromBrokerChange = _.debounce(function () {
      logger.debug('brokersChanged refreshing metadata');
      self.client.refreshMetadata(self.topics, function (error) {
        if (error) {
          self.emit(error);
          return;
        }
        self.paused = false;
        if (!self.ready && !self.connecting) {
          if (self.reconnectTimer) {
            // brokers changed so bypass backoff retry and reconnect now
            clearTimeout(self.reconnectTimer);
            self.reconnectTimer = null;
          }
          self.connect();
        } else if (!self.connecting) {
          self.fetch();
        }
      });
    }, 200);
  
    this.client.on('brokersChanged', function () {
      self.pause();
      recoverFromBrokerChange();
    });
  
    this.client.on('reconnect', function (lastError) {
      self.fetch();
    });
  
    this.on('offsetOutOfRange', topic => {
      this.pause();
      if (this.options.outOfRangeOffset === 'none') {
        this.emit('error', new errors.InvalidConsumerOffsetError(`Offset out of range for topic "${topic.topic}" partition ${topic
  .partition}`));
        return;
      }
  
      topic.time = ACCEPTED_FROM_OFFSET[this.options.outOfRangeOffset];
  
      this.getOffset().fetch([topic], (error, result) => {
        if (error) {
          this.emit('error', new errors.InvalidConsumerOffsetError(`Fetching ${this.options.outOfRangeOffset} offset failed`, error
  ));
          return;
        }
        const offset = _.head(result[topic.topic][topic.partition]);
        const oldOffset = _.find(this.topicPayloads, {topic: topic.topic, partition: topic.partition}).offset;
  
        logger.debug('replacing %s-%s stale offset of %d with %d', topic.topic, topic.partition, oldOffset, offset);
  
        this.setOffset(topic.topic, topic.partition, offset);
        this.resume();
      });
    });
  
    // 'done' will be emit when a message fetch request complete
    this.on('done', function (topics) {
      self.updateOffsets(topics);
      if (!self.paused) {
        setImmediate(function () {
          self.fetch();
        });
      }
    });
  
    if (this.options.groupId) {
      validateConfig('options.groupId', this.options.groupId);
    }
  
    this.isLeader = false;
    this.coordinatorId = null;
    this.generationId = null;
    this.ready = false;
    this.topicPayloads = [];
  }

• example usage

  n/a

function kafka-node.ConsumerGroup.super_ (client, topics, options)

• description and source-code

  super_ = function (client, topics, options) {
    if (!topics) {
      throw new Error('Must have payloads');
    }
    this.fetchCount = 0;
    this.client = client;
    this.options = _.defaults((options || {}), DEFAULTS);
    this.initialised = false;
    this.ready = false;
    this.closing = false;
    this.paused = this.options.paused;
    this.rebalancing = false;
    this.pendingRebalances = 0;
    this.committing = false;
    this.needToCommit = false;
    this.id = this.options.id || this.options.groupId + '_' + uuid.v4();
    this.payloads = this.buildPayloads(topics);
    this.topicPayloads = this.buildTopicPayloads(topics);
    this.connect();
  
    if (this.options.groupId) {
      validateConfig('options.groupId', this.options.groupId);
    }
  }

• example usage

  n/a

module kafka-node.ConsumerGroup.prototype

function kafka-node.ConsumerGroup.prototype.assignPartitions (protocol, groupMembers, callback)

• description and source-code

  assignPartitions = function (protocol, groupMembers, callback) {
    logger.debug('Assigning Partitions to members', groupMembers);
    logger.debug('Using group protocol', protocol);
  
    protocol = _.find(this.protocols, {name: protocol});
  
    var self = this;
    var topics = _(groupMembers).map('subscription').flatten().uniq().value();
  
    async.waterfall([
      function (callback) {
        logger.debug('loadingMetadata for topics:', topics);
        self.client.loadMetadataForTopics(topics, callback);
      },
  
      function (metadataResponse, callback) {
        var metadata = mapTopicToPartitions(metadataResponse[1].metadata);
        logger.debug('mapTopicToPartitions', metadata);
        protocol.assign(metadata, groupMembers, callback);
      }
    ], callback);
  }

• example usage

  ...
  this.isLeader = (joinGroupResponse.leaderId === joinGroupResponse.memberId);
  this.generationId = joinGroupResponse.generationId;
  this.memberId = joinGroupResponse.memberId;
  
  var groupAssignment;
  if (this.isLeader) {
    // assign partitions
    return this.assignPartitions(joinGroupResponse.groupProtocol, joinGroupResponse.members
  , callback);
  }
  callback(null, groupAssignment);
  };
  
  ConsumerGroup.prototype.saveDefaultOffsets = function (topicPartitionList, callback) {
  var self = this;
  const offsetPayload = _(topicPartitionList).cloneDeep().map(tp => {
  ...

function kafka-node.ConsumerGroup.prototype.close (force, cb)

• description and source-code

  close = function (force, cb) {
    var self = this;
    this.ready = false;
  
    this.stopHeartbeats();
  
    if (typeof force === 'function') {
      cb = force;
      force = false;
    }
  
    async.series([
      function (callback) {
        if (force) {
          self.commit(true, callback);
          return;
        }
        callback(null);
      },
      function (callback) {
        self.leaveGroup(function (error) {
          if (error) {
            logger.error('Leave group failed with', error);
          }
          callback(null);
        });
      },
      function (callback) {
        self.client.close(callback);
      }
    ], cb);
  }

• example usage

  ...
  
  ### close(force, cb)
  * `force`: **Boolean**, if set to true, it forces the consumer to commit the current offset before closing, default `false`
  
  Example
  
  ```js
  consumer.close(true, cb);
  consumer.close(cb); //force is disabled
  ```
  
  ## HighLevelConsumer
  ⚠️ ***This consumer has been deprecated in the latest version of Kafka (0.10.1) and is likely to be removed in the future. Please
   use the ConsumerGroup instead.***
  
  ### HighLevelConsumer(client, payloads, options)
  ...

function kafka-node.ConsumerGroup.prototype.connect ()

• description and source-code

  connect = function () {
    if (this.connecting) {
      logger.warn('Connect ignored. Currently connecting.');
      return;
    }
  
    logger.debug('Connecting %s', this.client.clientId);
    var self = this;
  
    this.connecting = true;
    this.emit('rebalancing');
  
    async.waterfall([
      function (callback) {
        if (self.client.coordinatorId) {
          return callback(null, null);
        }
        self.client.sendGroupCoordinatorRequest(self.options.groupId, callback);
      },
  
      function (coordinatorInfo, callback) {
        logger.debug('GroupCoordinator Response:', coordinatorInfo);
        if (coordinatorInfo) {
          self.setCoordinatorId(coordinatorInfo.coordinatorId);
        }
        self.client.sendJoinGroupRequest(self.options.groupId, emptyStrIfNull(self.memberId), self.options.sessionTimeout, self.protocols
  , callback);
      },
  
      function (joinGroupResponse, callback) {
        self.handleJoinGroup(joinGroupResponse, callback);
      },
  
      function (groupAssignment, callback) {
        logger.debug('SyncGroup Request from %s', self.memberId);
        self.client.sendSyncGroupRequest(self.options.groupId, self.generationId, self.memberId, groupAssignment, callback);
      },
  
      function (syncGroupResponse, callback) {
        self.handleSyncGroup(syncGroupResponse, callback);
      }
    ], function (error, startFetch) {
      self.connecting = false;
      self.rebalancing = false;
      if (error) {
        return self.recovery.tryToRecoverFrom(error, 'connect');
      }
  
      self.ready = true;
      self.recovery.clearError();
  
      logger.debug('generationId', self.generationId);
  
      if (startFetch) {
        self.fetch();
      }
      self.startHeartbeats();
      self.emit('connect');
      self.emit('rebalanced');
    });
  }

• example usage

  ...
  }
  
  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);
  
  this.connect();
  }
  
  util.inherits(BaseProducer, events.EventEmitter);
  
  BaseProducer.prototype.connect = function () {
  // emiter...
  var self = this;
  ...

function kafka-node.ConsumerGroup.prototype.fetchOffset (payloads, cb)

• description and source-code

  fetchOffset = function (payloads, cb) {
    this.client.sendOffsetFetchV1Request(this.options.groupId, payloads, cb);
  }

• example usage

  ...
      logger.debug('%s owns topics: ', self.client.clientId, syncGroupResponse.partitions);
  
      const topicPartitionList = createTopicPartitionList(syncGroupResponse.partitions);
      const useDefaultOffsets = self.options.fromOffset in ACCEPTED_FROM_OFFSET;
  
      async.waterfall([
        function (callback) {
  self.fetchOffset(syncGroupResponse.partitions, callback);
        },
        function (offsets, callback) {
  logger.debug('%s fetchOffset Response: %j', self.client.clientId, offsets);
  
  var noOffset = topicPartitionList.some(function (tp) {
    return offsets[tp.topic][tp.partition] === -1;
  });
  ...

function kafka-node.ConsumerGroup.prototype.getDefaultOffset (tp, defaultOffset)

• description and source-code

  getDefaultOffset = function (tp, defaultOffset) {
    return _.get(this.defaultOffsets, [tp.topic, tp.partition], defaultOffset);
  }

• example usage

  ...
      callback(null, offsets);
    }
  },
  function (offsets, callback) {
    self.topicPayloads = self.buildPayloads(topicPartitionList).map(function (p) {
      var offset = offsets[p.topic][p.partition];
      if (offset === -1) { // -1 means no offset was saved for this topic/partition combo
        offset = useDefaultOffsets ? self.getDefaultOffset(p, 0) : 0;
        if (self.migrator) {
          offset = self.migrator.getOffset(p, offset);
        }
      }
      p.offset = offset;
      return p;
    });
  ...

function kafka-node.ConsumerGroup.prototype.getOffset ()

• description and source-code

  getOffset = function () {
    if (this.offset) {
      return this.offset;
    }
    this.offset = new Offset(this.client);
    // we can ignore this since we are already forwarding error event emitted from client
    this.offset.on('error', _.noop);
    return this.offset;
  }

• example usage

  ...
  if (this.options.outOfRangeOffset === 'none') {
    this.emit('error', new errors.InvalidConsumerOffsetError(`Offset out of range for topic "${topic.topic}" partition
   ${topic.partition}`));
    return;
  }
  
  topic.time = ACCEPTED_FROM_OFFSET[this.options.outOfRangeOffset];
  
  this.getOffset().fetch([topic], (error, result) => {
    if (error) {
      this.emit('error', new errors.InvalidConsumerOffsetError(`Fetching ${this.options.outOfRangeOffset} offset failed`,
  error));
      return;
    }
    const offset = _.head(result[topic.topic][topic.partition]);
    const oldOffset = _.find(this.topicPayloads, {topic: topic.topic, partition: topic.partition}).offset;
  ...

function kafka-node.ConsumerGroup.prototype.handleJoinGroup (joinGroupResponse, callback)

• description and source-code

  handleJoinGroup = function (joinGroupResponse, callback) {
    logger.debug('joinGroupResponse %j from %s', joinGroupResponse, this.client.clientId);
  
    this.isLeader = (joinGroupResponse.leaderId === joinGroupResponse.memberId);
    this.generationId = joinGroupResponse.generationId;
    this.memberId = joinGroupResponse.memberId;
  
    var groupAssignment;
    if (this.isLeader) {
      // assign partitions
      return this.assignPartitions(joinGroupResponse.groupProtocol, joinGroupResponse.members, callback);
    }
    callback(null, groupAssignment);
  }

• example usage

  ...
    if (coordinatorInfo) {
      self.setCoordinatorId(coordinatorInfo.coordinatorId);
    }
    self.client.sendJoinGroupRequest(self.options.groupId, emptyStrIfNull(self.memberId), self.options.sessionTimeout, self.protocols
  , callback);
  },
  
  function (joinGroupResponse, callback) {
    self.handleJoinGroup(joinGroupResponse, callback);
  },
  
  function (groupAssignment, callback) {
    logger.debug('SyncGroup Request from %s', self.memberId);
    self.client.sendSyncGroupRequest(self.options.groupId, self.generationId, self.memberId, groupAssignment, callback);
  },
  ...

function kafka-node.ConsumerGroup.prototype.handleSyncGroup (syncGroupResponse, callback)

• description and source-code

  handleSyncGroup = function (syncGroupResponse, callback) {
    logger.debug('SyncGroup Response');
    var self = this;
    var ownedTopics = Object.keys(syncGroupResponse.partitions);
    if (ownedTopics.length) {
      logger.debug('%s owns topics: ', self.client.clientId, syncGroupResponse.partitions);
  
      const topicPartitionList = createTopicPartitionList(syncGroupResponse.partitions);
      const useDefaultOffsets = self.options.fromOffset in ACCEPTED_FROM_OFFSET;
  
      async.waterfall([
        function (callback) {
          self.fetchOffset(syncGroupResponse.partitions, callback);
        },
        function (offsets, callback) {
          logger.debug('%s fetchOffset Response: %j', self.client.clientId, offsets);
  
          var noOffset = topicPartitionList.some(function (tp) {
            return offsets[tp.topic][tp.partition] === -1;
          });
  
          if (noOffset) {
            logger.debug('No saved offsets');
  
            if (self.options.fromOffset === 'none') {
              return callback(new Error(`${self.client.clientId} owns topics and partitions which contains no saved offsets for group
   '${self.options.groupId}'`));
            }
  
            async.parallel([
              function (callback) {
                if (self.migrator) {
                  return self.migrator.saveHighLevelConsumerOffsets(topicPartitionList, callback);
                }
                callback(null);
              },
              function (callback) {
                if (useDefaultOffsets) {
                  return self.saveDefaultOffsets(topicPartitionList, callback);
                }
                callback(null);
              }
            ], function (error) {
              if (error) {
                return callback(error);
              }
              logger.debug('%s defaultOffset Response for %s: %j', self.client.clientId, self.options.fromOffset, self.defaultOffsets
  );
              callback(null, offsets);
            });
          } else {
            logger.debug('Has saved offsets');
            callback(null, offsets);
          }
        },
        function (offsets, callback) {
          self.topicPayloads = self.buildPayloads(topicPartitionList).map(function (p) {
            var offset = offsets[p.topic][p.partition];
            if (offset === -1) { // -1 means no offset was saved for this topic/partition combo
              offset = useDefaultOffsets ? self.getDefaultOffset(p, 0) : 0;
              if (self.migrator) {
                offset = self.migrator.getOffset(p, offset);
              }
            }
            p.offset = offset;
            return p;
          });
          callback(null, true);
        }
      ], callback);
    } else { // no partitions assigned
      callback(null, false);
    }
  }

• example usage

  ...
  
    function (groupAssignment, callback) {
      logger.debug('SyncGroup Request from %s', self.memberId);
      self.client.sendSyncGroupRequest(self.options.groupId, self.generationId, self.memberId, groupAssignment, callback);
    },
  
    function (syncGroupResponse, callback) {
      self.handleSyncGroup(syncGroupResponse, callback);
    }
  ], function (error, startFetch) {
    self.connecting = false;
    self.rebalancing = false;
    if (error) {
      return self.recovery.tryToRecoverFrom(error, 'connect');
    }
  ...

function kafka-node.ConsumerGroup.prototype.leaveGroup (callback)

• description and source-code

  leaveGroup = function (callback) {
    logger.debug('%s leaving group', this.client.clientId);
    var self = this;
    this.stopHeartbeats();
    if (self.generationId != null && self.memberId) {
      this.client.sendLeaveGroupRequest(this.options.groupId, this.memberId, function (error) {
        self.generationId = null;
        callback(error);
      });
    } else {
      callback(null);
    }
  }

• example usage

  ...
    if (force) {
      self.commit(true, callback);
      return;
    }
    callback(null);
  },
  function (callback) {
    self.leaveGroup(function (error) {
      if (error) {
        logger.error('Leave group failed with', error);
      }
      callback(null);
    });
  },
  function (callback) {
  ...

function kafka-node.ConsumerGroup.prototype.saveDefaultOffsets (topicPartitionList, callback)

• description and source-code

  saveDefaultOffsets = function (topicPartitionList, callback) {
    var self = this;
    const offsetPayload = _(topicPartitionList).cloneDeep().map(tp => {
      tp.time = ACCEPTED_FROM_OFFSET[this.options.fromOffset];
      return tp;
    });
  
    self.getOffset().fetch(offsetPayload, function (error, result) {
      if (error) {
        return callback(error);
      }
      self.defaultOffsets = _.mapValues(result, function (partitionOffsets) {
        return _.mapValues(partitionOffsets, _.head);
      });
      callback(null);
    });
  }

• example usage

  ...
      if (self.migrator) {
        return self.migrator.saveHighLevelConsumerOffsets(topicPartitionList, callback);
      }
      callback(null);
    },
    function (callback) {
      if (useDefaultOffsets) {
        return self.saveDefaultOffsets(topicPartitionList, callback);
      }
      callback(null);
    }
  ], function (error) {
    if (error) {
      return callback(error);
    }
  ...

function kafka-node.ConsumerGroup.prototype.scheduleReconnect (timeout)

• description and source-code

  scheduleReconnect = function (timeout) {
    assert(timeout);
    this.rebalancing = true;
  
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
    }
  
    var self = this;
    this.reconnectTimer = setTimeout(function () {
      self.reconnectTimer = null;
      self.connect();
    }, timeout);
  }

• example usage

  ...
  
    if (retry) {
      retryTimeout = this.getRetryTimeout(error);
    }
  
    if (retry && retryTimeout) {
      logger.debug('RECOVERY from %s: %s retrying in %s ms', source, this.consumerGroup.client.clientId, retryTimeout, error
  );
      this.consumerGroup.scheduleReconnect(retryTimeout);
    } else {
      this.consumerGroup.emit('error', error);
    }
    this.lastError = error;
  };
  
  ConsumerGroupRecovery.prototype.clearError = function () {
  ...

function kafka-node.ConsumerGroup.prototype.sendHeartbeat ()

• description and source-code

  sendHeartbeat = function () {
    assert(this.memberId, 'invalid memberId');
    assert(this.generationId >= 0, 'invalid generationId');
    // logger.debug('%s ❤️  ->', this.client.clientId);
    var self = this;
  
    function heartbeatCallback (error) {
      if (error) {
        logger.warn('%s Heartbeat error:', self.client.clientId, error);
        self.recovery.tryToRecoverFrom(error, 'heartbeat');
      }
      // logger.debug('%s 💚 <-', self.client.clientId, error);
    }
  
    const heartbeat = new Heartbeat(this.client, heartbeatCallback);
    heartbeat.send(this.options.groupId, this.generationId, this.memberId);
  
    return heartbeat;
  }

• example usage

  ...
  assert(this.ready, 'consumerGroup is not ready');
  
  const heartbeatIntervalMs = this.options.heartbeatInterval || (Math.floor(this.options.sessionTimeout / 3));
  
  logger.debug('%s started heartbeats at every %d ms', this.client.clientId, heartbeatIntervalMs);
  this.stopHeartbeats();
  
  let heartbeat = this.sendHeartbeat();
  
  this.heartbeatInterval = setInterval(() => {
    // only send another heartbeat if we got a response from the last one
    if (heartbeat.verifyResolved()) {
      heartbeat = this.sendHeartbeat();
    }
  }, heartbeatIntervalMs);
  ...

function kafka-node.ConsumerGroup.prototype.sendOffsetCommitRequest (commits, cb)

• description and source-code

  sendOffsetCommitRequest = function (commits, cb) {
    if (this.generationId && this.memberId) {
      this.client.sendOffsetCommitV2Request(this.options.groupId, this.generationId, this.memberId, commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }

• example usage

  ...
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var payloads = this.payloads;
    if (this.pausedPayloads) payloads = payloads.concat(this.pausedPayloads);
  
    var commits = payloads.filter(function (p) { return p.offset !== 0; });
    if (commits.length) {
      this.client.sendOffsetCommitRequest(this.options.groupId, commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }
  Consumer.prototype.commit = Consumer.prototype.autoCommit = autoCommit;
  
  Consumer.prototype.fetch = function () {
  ...

function kafka-node.ConsumerGroup.prototype.setCoordinatorId (coordinatorId)

• description and source-code

  setCoordinatorId = function (coordinatorId) {
    this.client.coordinatorId = String(coordinatorId);
  }

• example usage

  ...
    }
    self.client.sendGroupCoordinatorRequest(self.options.groupId, callback);
  },
  
  function (coordinatorInfo, callback) {
    logger.debug('GroupCoordinator Response:', coordinatorInfo);
    if (coordinatorInfo) {
      self.setCoordinatorId(coordinatorInfo.coordinatorId);
    }
    self.client.sendJoinGroupRequest(self.options.groupId, emptyStrIfNull(self.memberId), self.options.sessionTimeout, self.protocols
  , callback);
  },
  
  function (joinGroupResponse, callback) {
    self.handleJoinGroup(joinGroupResponse, callback);
  },
  ...

function kafka-node.ConsumerGroup.prototype.setupProtocols (protocols)

• description and source-code

  setupProtocols = function (protocols) {
    if (!Array.isArray(protocols)) {
      protocols = [protocols];
    }
  
    this.protocols = protocols.map(function (protocol) {
      if (typeof protocol === 'string') {
        if (!(protocol in builtInProtocols)) {
          throw new Error('Unknown built in assignment protocol ' + protocol);
        }
        protocol = _.assign({}, builtInProtocols[protocol]);
      } else {
        checkProtocol(protocol);
      }
  
      protocol.subscription = this.topics;
      return protocol;
    }, this);
  }

• example usage

  ...
  
  assert(Array.isArray(topics), 'Array of topics is required');
  
  this.topics = topics;
  
  this.recovery = new ConsumerGroupRecovery(this);
  
  this.setupProtocols(this.options.protocol);
  
  if (this.options.connectOnReady && !this.options.migrateHLC) {
    this.client.once('ready', this.connect.bind(this));
  }
  
  if (this.options.migrateHLC) {
    const ConsumerGroupMigrator = require('./consumerGroupMigrator');
  ...

function kafka-node.ConsumerGroup.prototype.startHeartbeats ()

• description and source-code

  startHeartbeats = function () {
    assert(this.options.sessionTimeout > 0);
    assert(this.ready, 'consumerGroup is not ready');
  
    const heartbeatIntervalMs = this.options.heartbeatInterval || (Math.floor(this.options.sessionTimeout / 3));
  
    logger.debug('%s started heartbeats at every %d ms', this.client.clientId, heartbeatIntervalMs);
    this.stopHeartbeats();
  
    let heartbeat = this.sendHeartbeat();
  
    this.heartbeatInterval = setInterval(() => {
      // only send another heartbeat if we got a response from the last one
      if (heartbeat.verifyResolved()) {
        heartbeat = this.sendHeartbeat();
      }
    }, heartbeatIntervalMs);
  }

• example usage

  ...
    self.recovery.clearError();
  
    logger.debug('generationId', self.generationId);
  
    if (startFetch) {
      self.fetch();
    }
    self.startHeartbeats();
    self.emit('connect');
    self.emit('rebalanced');
  });
  };
  
  ConsumerGroup.prototype.scheduleReconnect = function (timeout) {
  assert(timeout);
  ...

function kafka-node.ConsumerGroup.prototype.stopHeartbeats ()

• description and source-code

  stopHeartbeats = function () {
    this.heartbeatInterval && clearInterval(this.heartbeatInterval);
  }

• example usage

  ...
  ConsumerGroup.prototype.startHeartbeats = function () {
  assert(this.options.sessionTimeout > 0);
  assert(this.ready, 'consumerGroup is not ready');
  
  const heartbeatIntervalMs = this.options.heartbeatInterval || (Math.floor(this.options.sessionTimeout / 3));
  
  logger.debug('%s started heartbeats at every %d ms', this.client.clientId, heartbeatIntervalMs);
  this.stopHeartbeats();
  
  let heartbeat = this.sendHeartbeat();
  
  this.heartbeatInterval = setInterval(() => {
    // only send another heartbeat if we got a response from the last one
    if (heartbeat.verifyResolved()) {
      heartbeat = this.sendHeartbeat();
  ...

module kafka-node.CustomPartitioner

function kafka-node.CustomPartitioner (partitioner)

• description and source-code

  CustomPartitioner = function (partitioner) {
    this.getPartition = partitioner;
  }

• example usage

  n/a

function kafka-node.CustomPartitioner.super_ ()

• description and source-code

  super_ = function () {}

• example usage

  n/a

module kafka-node.CyclicPartitioner

function kafka-node.CyclicPartitioner ()

• description and source-code

  CyclicPartitioner = function () {
    this.c = 0;
  }

• example usage

  n/a

function kafka-node.CyclicPartitioner.super_ ()

• description and source-code

  super_ = function () {}

• example usage

  n/a

module kafka-node.CyclicPartitioner.prototype

function kafka-node.CyclicPartitioner.prototype.getPartition (partitions)

• description and source-code

  getPartition = function (partitions) {
    if (_.isEmpty(partitions)) return 0;
    return partitions[ this.c++ % partitions.length ];
  }

• example usage

  ...
  
    client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
    const topicPartitionRequests = Object.create(null);
    payloads.forEach((p) => {
  p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition
  (_.map(topicMetadata[p.topic], 'partition'), p.key);
  p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  let messages = _.isArray(p.messages) ? p.messages : [p.messages];
  
  messages = messages.map(function (message) {
    if (message instanceof KeyedMessage) {
      return message;
    }
  ...

module kafka-node.DefaultPartitioner

function kafka-node.DefaultPartitioner ()

• description and source-code

  DefaultPartitioner = function () {}

• example usage

  n/a

function kafka-node.DefaultPartitioner.super_ ()

• description and source-code

  super_ = function () {}

• example usage

  n/a

module kafka-node.DefaultPartitioner.prototype

function kafka-node.DefaultPartitioner.prototype.getPartition (partitions)

• description and source-code

  getPartition = function (partitions) {
    if (partitions && _.isArray(partitions) && partitions.length > 0) {
      return partitions[0];
    } else {
      return 0;
    }
  }

• example usage

  ...
  
    client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
    const topicPartitionRequests = Object.create(null);
    payloads.forEach((p) => {
  p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition
  (_.map(topicMetadata[p.topic], 'partition'), p.key);
  p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  let messages = _.isArray(p.messages) ? p.messages : [p.messages];
  
  messages = messages.map(function (message) {
    if (message instanceof KeyedMessage) {
      return message;
    }
  ...

module kafka-node.FailedToRebalanceConsumerError

function kafka-node.FailedToRebalanceConsumerError (message)

• description and source-code

  FailedToRebalanceConsumerError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  ...
        operation.attempt(function (currentAttempt) {
  self.rebalanceAttempt(oldTopicPayloads, function (err) {
    if (operation.retry(err)) {
      return;
    }
    if (err) {
      self.rebalancing = false;
      return self.emit('error', new errors.FailedToRebalanceConsumerError(operation
  .mainError().toString()));
    } else {
      var topicNames = self.topicPayloads.map(function (p) {
        return p.topic;
      });
      self.client.refreshMetadata(topicNames, function (err) {
        register();
        if (err) {
  ...

function kafka-node.FailedToRebalanceConsumerError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.FailedToRegisterConsumerError

function kafka-node.FailedToRegisterConsumerError (message, nested)

• description and source-code

  FailedToRegisterConsumerError = function (message, nested) {
    NestedError.call(this, message, nested);
    this.message = message;
  }

• example usage

  ...
            } else {
              callback(new Error(util.format('Consumer %s is not registered in group %s', self.id, self.options.groupId)));
            }
          });
        }
      ], function (error) {
        if (error) {
          self.emit('error', new errors.FailedToRegisterConsumerError(error.toString
  (), error));
        }
      });
    }
  }, 20000);
  
  function fetchAndUpdateOffsets (cb) {
    self.fetchOffset(self.topicPayloads, function (err, topics) {
  ...

function kafka-node.FailedToRegisterConsumerError.super_ (message, nested)

• description and source-code

  super_ = function (message, nested) {
      this.nested = nested;
  
      if (typeof message !== 'undefined') {
          Object.defineProperty(this, 'message', {
              value: message,
              writable: true,
              enumerable: false,
              configurable: true
          });
      }
  
      Error.captureStackTrace(this, this.constructor);
      var oldStackDescriptor = Object.getOwnPropertyDescriptor(this, 'stack');
      var stackDescriptor = buildStackDescriptor(oldStackDescriptor, nested);
      Object.defineProperty(this, 'stack', stackDescriptor);
  }

• example usage

  n/a

module kafka-node.GroupCoordinatorNotAvailableError

function kafka-node.GroupCoordinatorNotAvailableError (message)

• description and source-code

  GroupCoordinatorNotAvailableError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.GroupCoordinatorNotAvailableError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.GroupLoadInProgressError

function kafka-node.GroupLoadInProgressError (message)

• description and source-code

  GroupLoadInProgressError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.GroupLoadInProgressError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.HeartbeatTimeoutError

function kafka-node.HeartbeatTimeoutError (message)

• description and source-code

  HeartbeatTimeoutError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.HeartbeatTimeoutError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.HighLevelConsumer

function kafka-node.HighLevelConsumer (client, topics, options)

• description and source-code

  HighLevelConsumer = function (client, topics, options) {
    if (!topics) {
      throw new Error('Must have payloads');
    }
    this.fetchCount = 0;
    this.client = client;
    this.options = _.defaults((options || {}), DEFAULTS);
    this.initialised = false;
    this.ready = false;
    this.closing = false;
    this.paused = this.options.paused;
    this.rebalancing = false;
    this.pendingRebalances = 0;
    this.committing = false;
    this.needToCommit = false;
    this.id = this.options.id || this.options.groupId + '_' + uuid.v4();
    this.payloads = this.buildPayloads(topics);
    this.topicPayloads = this.buildTopicPayloads(topics);
    this.connect();
  
    if (this.options.groupId) {
      validateConfig('options.groupId', this.options.groupId);
    }
  }

• example usage

  n/a

function kafka-node.HighLevelConsumer.super_ ()

• description and source-code

  function EventEmitter() {
    EventEmitter.init.call(this);
  }

• example usage

  n/a

module kafka-node.HighLevelConsumer.prototype

function kafka-node.HighLevelConsumer.prototype._releasePartitions (topicPayloads, callback)

• description and source-code

  _releasePartitions = function (topicPayloads, callback) {
    var self = this;
    async.each(topicPayloads, function (tp, cbb) {
      if (tp.partition !== undefined) {
        async.series([
          function (delcbb) {
            self.client.zk.checkPartitionOwnership(self.id, self.options.groupId, tp.topic, tp.partition, function (err) {
              if (err) {
                // Partition doesn't exist simply carry on
                cbb();
              } else delcbb();
            });
          },
          function (delcbb) {
            self.client.zk.deletePartitionOwnership(self.options.groupId, tp.topic, tp.partition, delcbb);
          },
          function (delcbb) {
            self.client.zk.checkPartitionOwnership(self.id, self.options.groupId, tp.topic, tp.partition, function (err) {
              if (err) {
                delcbb();
              } else {
                delcbb('Partition should not exist');
              }
            });
          }],
        cbb);
      } else {
        cbb();
      }
    }, callback);
  }

• example usage

  ...
      }
    });
  },
  
  // Release current partitions
  function (callback) {
    logger.debug('HighLevelConsumer %s releasing current partitions during rebalance', self.id);
    self._releasePartitions(oldTopicPayloads, callback);
  },
  
  // Rebalance
  function (callback) {
    logger.debug('HighLevelConsumer %s determining the partitions to own during rebalance', self.id);
    logger.debug('consumerPerTopicMap.consumerTopicMap %j', consumerPerTopicMap.consumerTopicMap);
    for (var topic in consumerPerTopicMap.consumerTopicMap[self.id]) {
  ...

function kafka-node.HighLevelConsumer.prototype.addTopics (topics, cb)

• description and source-code

  addTopics = function (topics, cb) {
    var self = this;
    if (!this.ready) {
      setTimeout(function () {
        self.addTopics(topics, cb);
      }, 100);
      return;
    }
    this.client.addTopics(
      topics,
      function (err, added) {
        if (err) return cb && cb(err, added);
  
        var payloads = self.buildPayloads(topics);
        // update offset of topics that will be added
        self.fetchOffset(payloads, function (err, offsets) {
          if (err) return cb(err);
          payloads.forEach(function (p) {
            var offset = offsets[p.topic][p.partition];
            if (offset === -1) offset = 0;
            p.offset = offset;
            self.topicPayloads.push(p);
          });
          // TODO: rebalance consumer
          cb && cb(null, added);
        });
      }
    );
  }

• example usage

  ...
  * `topics`: **Array**, array of topics to add
  * `cb`: **Function**,the callback
  * `fromOffset`: **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:
  
  ``` js
  consumer.addTopics(['t1', 't2'], function (err, added) {
  });
  
  or
  
  consumer.addTopics([{ topic: 't1', offset: 10 }], function (err, added) {
  }, true);
  ```
  ...

function kafka-node.HighLevelConsumer.prototype.autoCommit (force, cb)

• description and source-code

  function autoCommit(force, cb) {
    if (arguments.length === 1) {
      cb = force;
      force = false;
    }
  
    if (!force) {
      if (this.committing) return cb(null, 'Offset committing');
      if (!this.needToCommit) return cb(null, 'Commit not needed');
    }
  
    this.needToCommit = false;
    this.committing = true;
    setTimeout(function () {
      this.committing = false;
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var commits = this.topicPayloads.filter(function (p) { return p.offset !== -1; });
  
    if (commits.length) {
      this.sendOffsetCommitRequest(commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }

• example usage

  ...
      if (offset === -1) offset = 0;
      if (!initing) p.offset = offset + 1;
      else p.offset = offset;
    }
  });
  
  if (this.options.autoCommit && !initing) {
    this.autoCommit(false, function (err) {
      err && logger.debug('auto commit offset', err);
    });
  }
  };
  
  function autoCommit (force, cb) {
  if (arguments.length === 1) {
  ...

function kafka-node.HighLevelConsumer.prototype.buildPayloads (payloads)

• description and source-code

  buildPayloads = function (payloads) {
    var self = this;
    return payloads.map(function (p) {
      if (typeof p !== 'object') p = { topic: p };
      p.partition = p.partition || 0;
      p.offset = p.offset || 0;
      p.maxBytes = self.options.fetchMaxBytes;
      p.metadata = 'm'; // metadata can be arbitrary
      return p;
    });
  }

• example usage

  ...
   * @param {BaseProducer~sendCallback} cb A function to call once the send has completed
   */
  BaseProducer.prototype.send = function (payloads, cb) {
  var client = this.client;
  var requireAcks = this.requireAcks;
  var ackTimeoutMs = this.ackTimeoutMs;
  
  client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks
  , ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
  const topicPartitionRequests = Object.create(null);
  payloads.forEach((p) => {
    p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition(_.map(topicMetadata[p.topic
  ], 'partition'), p.key);
    p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  ...

function kafka-node.HighLevelConsumer.prototype.buildTopicPayloads (topics)

• description and source-code

  buildTopicPayloads = function (topics) {
    return topics.map(function (j) {
      var k = { topic: j.topic };
      return k;
    });
  }

• example usage

  ...
    this.paused = this.options.paused;
    this.rebalancing = false;
    this.pendingRebalances = 0;
    this.committing = false;
    this.needToCommit = false;
    this.id = this.options.id || this.options.groupId + '_' + uuid.v4();
    this.payloads = this.buildPayloads(topics);
    this.topicPayloads = this.buildTopicPayloads(topics);
    this.connect();
  
    if (this.options.groupId) {
      validateConfig('options.groupId', this.options.groupId);
    }
  };
  util.inherits(HighLevelConsumer, events.EventEmitter);
  ...

function kafka-node.HighLevelConsumer.prototype.close (force, cb)

• description and source-code

  close = function (force, cb) {
    var self = this;
    this.ready = false;
    this.closing = true;
    clearInterval(this.checkPartitionOwnershipInterval);
  
    if (typeof force === 'function') {
      cb = force;
      force = false;
    }
  
    async.series([
      function (callback) {
        self.leaveGroup(callback);
      },
      function (callback) {
        if (force) {
          async.series([
            function (callback) {
              self.commit(true, callback);
            },
            function (callback) {
              self.client.close(callback);
            }
          ], callback);
          return;
        }
        self.client.close(callback);
      }
    ], cb);
  }

• example usage

  ...
  
  ### close(force, cb)
  * `force`: **Boolean**, if set to true, it forces the consumer to commit the current offset before closing, default `false`
  
  Example
  
  ```js
  consumer.close(true, cb);
  consumer.close(cb); //force is disabled
  ```
  
  ## HighLevelConsumer
  ⚠️ ***This consumer has been deprecated in the latest version of Kafka (0.10.1) and is likely to be removed in the future. Please
   use the ConsumerGroup instead.***
  
  ### HighLevelConsumer(client, payloads, options)
  ...

function kafka-node.HighLevelConsumer.prototype.commit (force, cb)

• description and source-code

  function autoCommit(force, cb) {
    if (arguments.length === 1) {
      cb = force;
      force = false;
    }
  
    if (!force) {
      if (this.committing) return cb(null, 'Offset committing');
      if (!this.needToCommit) return cb(null, 'Commit not needed');
    }
  
    this.needToCommit = false;
    this.committing = true;
    setTimeout(function () {
      this.committing = false;
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var commits = this.topicPayloads.filter(function (p) { return p.offset !== -1; });
  
    if (commits.length) {
      this.sendOffsetCommitRequest(commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }

• example usage

  ...
  Commit offset of the current topics manually, this method should be called when a consumer leaves
  
  * `cb`: **Function**, the callback
  
  Example:
  
  ``` js
  consumer.commit(function(err, data) {
  });
  ```
  
  ### setOffset(topic, partition, offset)
  Set offset of the given topic
  
  * `topic`: **String**
  ...

function kafka-node.HighLevelConsumer.prototype.connect ()

• description and source-code

  connect = function () {
    var self = this;
    // Client alreadyexists
    if (this.client.ready) {
      this.init();
    }
  
    this.client.on('ready', function () {
      if (!self.initialised) self.init();
  
      // Check the topics exist and create a watcher on them
      var topics = self.payloads.map(function (p) {
        return p.topic;
      });
  
      self.client.topicExists(topics, function (err) {
        if (err) {
          return self.emit('error', err);
        }
        self.initialised = true;
      });
    });
  
    function checkPartitionOwnership (callback) {
      async.each(self.topicPayloads, function (tp, cbb) {
        if (tp.partition !== undefined) {
          self.client.zk.checkPartitionOwnership(self.id, self.options.groupId, tp.topic, tp.partition, function (err) {
            if (err) {
              cbb(err);
            } else {
              cbb();
            }
          });
        } else {
          cbb();
        }
      }, callback);
    }
  
    // Check partition ownership and registration
    this.checkPartitionOwnershipInterval = setInterval(function () {
      if (!self.rebalancing) {
        async.parallel([
          checkPartitionOwnership,
          function (callback) {
            self.client.zk.isConsumerRegistered(self.options.groupId, self.id, function (error, registered) {
              if (error) {
                return callback(error);
              }
              if (registered) {
                callback();
              } else {
                callback(new Error(util.format('Consumer %s is not registered in group %s', self.id, self.options.groupId)));
              }
            });
          }
        ], function (error) {
          if (error) {
            self.emit('error', new errors.FailedToRegisterConsumerError(error.toString(), error));
          }
        });
      }
    }, 20000);
  
    function fetchAndUpdateOffsets (cb) {
      self.fetchOffset(self.topicPayloads, function (err, topics) {
        if (err) {
          return cb(err);
        }
  
        self.ready = true;
        self.updateOffsets(topics, true);
  
        return cb();
      });
    }
  
    function rebalance () {
      logger.debug('rebalance() %s is rebalancing: %s ready: %s', self.id, self.rebalancing, self.ready);
      if (!self.rebalancing && !self.closing) {
        deregister();
  
        self.emit('rebalancing');
  
        self.rebalancing = true;
        logger.debug('HighLevelConsumer rebalance retry config: %s', JSON.stringify(self.options.rebalanceRetry));
        var oldTopicPayloads = self.topicPayloads;
        var operation = retry.operation(self.options.rebalanceRetry);
  
        operation.attempt(function (currentAttempt) {
          self.rebalanceAttempt(oldTopicPayloads, function (err) {
            if (operation.retry(err)) {
              return;
            }
            if (err) {
              self.rebalancing = false;
              return self.emit('error', new errors.FailedToRebalanceConsumerError(operation.mainError().toString()));
            } else {
              var topicNames = self.topicPayloads.map(function (p) {
                return p.topic;
              });
              self.client.refreshMetadata(topicNames, function (err) {
                register();
                if (err) {
                  self.rebalancing = false;
                  self.emit('error', err);
                  return;
                }
  
                if (self.topicPayloads.length) {
                  fetchAndUpdateOffsets(function (err) {
                    self.rebalancing = false;
                    if (err) {
                      self.emit('error', new errors.FailedToRebalanceConsumerError(err.message));
                      return;
                    }
                    self.fetch();
                    self.emit('rebalanced');
                  });
                } else { // was not assigned any partitions during rebalance
                  self.rebalancing = false;
                  self.emit('rebalanced');
                }
              });
            }
          });
        });
      }
    }
  
    // Wait for the consumer to be ready
    this.on('registered', rebalance);
  
    function register (fn) {
      logger.debug('Registered listeners %s', self.id);
      self.client.zk.on('consumersChanged', fn ...

• example usage

  ...
  }
  
  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);
  
  this.connect();
  }
  
  util.inherits(BaseProducer, events.EventEmitter);
  
  BaseProducer.prototype.connect = function () {
  // emiter...
  var self = this;
  ...

function kafka-node.HighLevelConsumer.prototype.fetch ()

• description and source-code

  fetch = function () {
    if (!this.ready || this.rebalancing || this.paused) {
      return;
    }
  
    this.client.sendFetchRequest(this, this.topicPayloads, this.options.fetchMaxWaitMs, this.options.fetchMinBytes, this.options.maxTickMessages
  );
  }

• example usage

  ...
  
  Example
  
  ```js
  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] } }
      });
  ```
  ...

function kafka-node.HighLevelConsumer.prototype.fetchOffset (payloads, cb)

• description and source-code

  fetchOffset = function (payloads, cb) {
    logger.debug('in fetchOffset %s payloads: %j', this.id, payloads);
    this.client.sendOffsetFetchRequest(this.options.groupId, payloads, cb);
  }

• example usage

  ...
      logger.debug('%s owns topics: ', self.client.clientId, syncGroupResponse.partitions);
  
      const topicPartitionList = createTopicPartitionList(syncGroupResponse.partitions);
      const useDefaultOffsets = self.options.fromOffset in ACCEPTED_FROM_OFFSET;
  
      async.waterfall([
        function (callback) {
  self.fetchOffset(syncGroupResponse.partitions, callback);
        },
        function (offsets, callback) {
  logger.debug('%s fetchOffset Response: %j', self.client.clientId, offsets);
  
  var noOffset = topicPartitionList.some(function (tp) {
    return offsets[tp.topic][tp.partition] === -1;
  });
  ...

function kafka-node.HighLevelConsumer.prototype.getTopicPayloads ()

• description and source-code

  getTopicPayloads = function () {
    if (!this.rebalancing) return this.topicPayloads;
    return null;
  }

• example usage

  n/a

function kafka-node.HighLevelConsumer.prototype.init ()

• description and source-code

  init = function () {
    var self = this;
  
    if (!self.topicPayloads.length) {
      return;
    }
  
    self.registerConsumer(function (err) {
      if (err) {
        return self.emit('error', new errors.FailedToRegisterConsumerError(err.toString()));
      }
  
      // Close the
      return self.emit('registered');
    });
  }

• example usage

  ...
  });
  };
  
  Consumer.prototype.connect = function () {
  var self = this;
  // Client already exists
  this.ready = this.client.ready;
  if (this.ready) this.init();
  
  this.client.on('ready', function () {
    logger.debug('consumer ready');
    if (!self.ready) self.init();
    self.ready = true;
  });
  ...

function kafka-node.HighLevelConsumer.prototype.leaveGroup (cb)

• description and source-code

  leaveGroup = function (cb) {
    var self = this;
    async.parallel([
      function (callback) {
        if (self.topicPayloads.length) {
          self._releasePartitions(self.topicPayloads, callback);
        } else {
          callback(null);
        }
      },
      function (callback) {
        self.client.zk.unregisterConsumer(self.options.groupId, self.id, callback);
      }
    ], cb);
  }

• example usage

  ...
    if (force) {
      self.commit(true, callback);
      return;
    }
    callback(null);
  },
  function (callback) {
    self.leaveGroup(function (error) {
      if (error) {
        logger.error('Leave group failed with', error);
      }
      callback(null);
    });
  },
  function (callback) {
  ...

function kafka-node.HighLevelConsumer.prototype.offsetRequest (payloads, cb)

• description and source-code

  offsetRequest = function (payloads, cb) {
    this.client.sendOffsetRequest(payloads, cb);
  }

• example usage

  ...
  });
  
  this.on('offsetOutOfRange', function (topic) {
    self.pause();
    topic.maxNum = self.options.maxNumSegments;
    topic.metadata = 'm';
    topic.time = Date.now();
    self.offsetRequest([topic], function (err, offsets) {
      if (err) {
        self.emit('error', new errors.InvalidConsumerOffsetError(self));
      } else {
        var min = Math.min.apply(null, offsets[topic.topic][topic.partition]);
        // set minimal offset
        self.setOffset(topic.topic, topic.partition, min);
        self.resume();
  ...

function kafka-node.HighLevelConsumer.prototype.pause ()

• description and source-code

  pause = function () {
    this.paused = true;
  }

• example usage

  ...
      } else if (!self.connecting) {
        self.fetch();
      }
    });
  }, 200);
  
  this.client.on('brokersChanged', function () {
    self.pause();
    recoverFromBrokerChange();
  });
  
  this.client.on('reconnect', function (lastError) {
    self.fetch();
  });
  ...

function kafka-node.HighLevelConsumer.prototype.rebalanceAttempt (oldTopicPayloads, cb)

• description and source-code

  rebalanceAttempt = function (oldTopicPayloads, cb) {
    var self = this;
    // Do the rebalance.....
    var consumerPerTopicMap;
    var newTopicPayloads = [];
    logger.debug('HighLevelConsumer %s is attempting to rebalance', self.id);
    async.series([
  
      // Stop fetching data and commit offsets
      function (callback) {
        logger.debug('HighLevelConsumer %s stopping data read during rebalance', self.id);
        self.stop(function () {
          callback();
        });
      },
  
      // Assemble the data
      function (callback) {
        logger.debug('HighLevelConsumer %s assembling data for rebalance', self.id);
        self.client.zk.getConsumersPerTopic(self.options.groupId, function (err, obj) {
          if (err) {
            callback(err);
          } else {
            consumerPerTopicMap = obj;
            callback();
          }
        });
      },
  
      // Release current partitions
      function (callback) {
        logger.debug('HighLevelConsumer %s releasing current partitions during rebalance', self.id);
        self._releasePartitions(oldTopicPayloads, callback);
      },
  
      // Rebalance
      function (callback) {
        logger.debug('HighLevelConsumer %s determining the partitions to own during rebalance', self.id);
        logger.debug('consumerPerTopicMap.consumerTopicMap %j', consumerPerTopicMap.consumerTopicMap);
        for (var topic in consumerPerTopicMap.consumerTopicMap[self.id]) {
          if (!consumerPerTopicMap.consumerTopicMap[self.id].hasOwnProperty(topic)) {
            continue;
          }
          var topicToAdd = consumerPerTopicMap.consumerTopicMap[self.id][topic];
          var numberOfConsumers = consumerPerTopicMap.topicConsumerMap[topicToAdd].length;
          var numberOfPartition = consumerPerTopicMap.topicPartitionMap[topicToAdd].length;
          var partitionsPerConsumer = Math.floor(numberOfPartition / numberOfConsumers);
          var extraPartitions = numberOfPartition % numberOfConsumers;
          var currentConsumerIndex;
          for (var index in consumerPerTopicMap.topicConsumerMap[topicToAdd]) {
            if (!consumerPerTopicMap.topicConsumerMap[topicToAdd].hasOwnProperty(index)) {
              continue;
            }
            if (consumerPerTopicMap.topicConsumerMap[topicToAdd][index] === self.id) {
              currentConsumerIndex = parseInt(index);
              break;
            }
          }
          var extraBit = currentConsumerIndex;
          if (currentConsumerIndex > extraPartitions) extraBit = extraPartitions;
          var startPart = partitionsPerConsumer * currentConsumerIndex + extraBit;
          var extraNParts = 1;
          if (currentConsumerIndex + 1 > extraPartitions) extraNParts = 0;
          var nParts = partitionsPerConsumer + extraNParts;
  
          for (var i = startPart; i < startPart + nParts; i++) {
            newTopicPayloads.push({
              topic: topicToAdd,
              partition: consumerPerTopicMap.topicPartitionMap[topicToAdd][i],
              offset: 0,
              maxBytes: self.options.fetchMaxBytes,
              metadata: 'm'
            });
          }
        }
        logger.debug('newTopicPayloads %j', newTopicPayloads);
        callback();
      },
  
      // Update ZK with new ownership
      function (callback) {
        if (newTopicPayloads.length) {
          logger.debug('HighLevelConsumer %s gaining ownership of partitions during rebalance', self.id);
          async.eachSeries(newTopicPayloads, function (tp, cbb) {
            if (tp.partition !== undefined) {
              async.series([
                function (addcbb) {
                  self.client.zk.checkPartitionOwnership(self.id, self.options.groupId, tp.topic, tp.partition, function (err) {
                    if (err) {
                      // Partition doesn't exist simply carry on
                      addcbb();
                    } else cbb(); // Partition exists simply carry on
                  });
                },
                function (addcbb) {
                  self.client.zk.addPartitionOwnership(self.id, self.options.groupId, tp.topic, tp.partition, function (err) {
                    if (err) {
                      addcbb(err);
                    } else addcbb(); ...

• example usage

  ...
  
  self.rebalancing = true;
  logger.debug('HighLevelConsumer rebalance retry config: %s', JSON.stringify(self.options.rebalanceRetry));
  var oldTopicPayloads = self.topicPayloads;
  var operation = retry.operation(self.options.rebalanceRetry);
  
  operation.attempt(function (currentAttempt) {
    self.rebalanceAttempt(oldTopicPayloads, function (err) {
      if (operation.retry(err)) {
        return;
      }
      if (err) {
        self.rebalancing = false;
        return self.emit('error', new errors.FailedToRebalanceConsumerError(operation.mainError().toString()));
      } else {
  ...

function kafka-node.HighLevelConsumer.prototype.registerConsumer (cb)

• description and source-code

  registerConsumer = function (cb) {
    var self = this;
    var groupId = this.options.groupId;
    this.client.zk.registerConsumer(groupId, this.id, this.payloads, function (err) {
      if (err) return cb(err);
      self.client.zk.listConsumers(self.options.groupId);
      var topics = self.topicPayloads.reduce(function (ret, topicPayload) {
        if (ret.indexOf(topicPayload.topic) === -1) {
          ret.push(topicPayload.topic);
        }
        return ret;
      }, []);
      topics.forEach(function (topic) {
        self.client.zk.listPartitions(topic);
      });
      cb();
    });
  }

• example usage

  ...
  this.client.on('zkReconnect', function () {
    logger.debug('zookeeper reconnect for %s', self.id);
    attachZookeeperErrorListener();
  
    // clean up what's leftover
    self.leaveGroup(function () {
      // rejoin the group
      self.registerConsumer(function (error) {
        if (error) {
          return self.emit('error', new errors.FailedToRegisterConsumerError('Failed to register consumer on zkReconnect
  ', error));
        }
        self.emit('registered');
      });
    });
  });
  ...

function kafka-node.HighLevelConsumer.prototype.removeTopics (topics, cb)

• description and source-code

  removeTopics = function (topics, cb) {
    topics = typeof topics === 'string' ? [topics] : topics;
    this.payloads = this.payloads.filter(function (p) {
      return !~topics.indexOf(p.topic);
    });
  
    this.client.removeTopicMetadata(topics, cb);
  }

• example usage

  ...
  ### removeTopics(topics, cb)
  * `topics`: **Array**, array of topics to remove
  * `cb`: **Function**, the callback
  
  Example:
  
  ``` js
  consumer.removeTopics(['t1', 't2'], function (err, removed) {
  });
  ```
  
  ### commit(cb)
  Commit offset of the current topics manually, this method should be called when a consumer leaves
  
  * `cb`: **Function**, the callback
  ...

function kafka-node.HighLevelConsumer.prototype.resume ()

• description and source-code

  resume = function () {
    this.paused = false;
    this.fetch();
  }

• example usage

  ...
      }
      const offset = _.head(result[topic.topic][topic.partition]);
      const oldOffset = _.find(this.topicPayloads, {topic: topic.topic, partition: topic.partition}).offset;
  
      logger.debug('replacing %s-%s stale offset of %d with %d', topic.topic, topic.partition, oldOffset, offset);
  
      this.setOffset(topic.topic, topic.partition, offset);
      this.resume();
    });
  });
  
  // 'done' will be emit when a message fetch request complete
  this.on('done', function (topics) {
    self.updateOffsets(topics);
    if (!self.paused) {
  ...

function kafka-node.HighLevelConsumer.prototype.sendOffsetCommitRequest (commits, cb)

• description and source-code

  sendOffsetCommitRequest = function (commits, cb) {
    this.client.sendOffsetCommitRequest(this.options.groupId, commits, cb);
  }

• example usage

  ...
    }.bind(this), this.options.autoCommitIntervalMs);
  
    var payloads = this.payloads;
    if (this.pausedPayloads) payloads = payloads.concat(this.pausedPayloads);
  
    var commits = payloads.filter(function (p) { return p.offset !== 0; });
    if (commits.length) {
      this.client.sendOffsetCommitRequest(this.options.groupId, commits, cb);
    } else {
      cb(null, 'Nothing to be committed');
    }
  }
  Consumer.prototype.commit = Consumer.prototype.autoCommit = autoCommit;
  
  Consumer.prototype.fetch = function () {
  ...

function kafka-node.HighLevelConsumer.prototype.setOffset (topic, partition, offset)

• description and source-code

  setOffset = function (topic, partition, offset) {
    this.topicPayloads.every(function (p) {
      if (p.topic === topic && p.partition == partition) { // eslint-disable-line eqeqeq
        p.offset = offset;
        return false;
      }
      return true;
    });
  }

• example usage

  ...
  * `partition`: **Number**
  
  * `offset`: **Number**
  
  Example:
  
  ``` js
  consumer.setOffset('topic', 0, 0);
  ```
  
  ### pause()
  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()
  Resume the consumer. Resumes the fetch loop.
  ...

function kafka-node.HighLevelConsumer.prototype.stop (cb)

• description and source-code

  stop = function (cb) {
    if (!this.options.autoCommit) return cb && cb();
    this.commit(true, function (err) {
      cb && cb(err);
    });
  }

• example usage

  ...
    var newTopicPayloads = [];
    logger.debug('HighLevelConsumer %s is attempting to rebalance', self.id);
    async.series([
  
  // Stop fetching data and commit offsets
  function (callback) {
    logger.debug('HighLevelConsumer %s stopping data read during rebalance', self.id);
    self.stop(function () {
      callback();
    });
  },
  
  // Assemble the data
  function (callback) {
    logger.debug('HighLevelConsumer %s assembling data for rebalance', self.id);
  ...

function kafka-node.HighLevelConsumer.prototype.updateOffsets (topics, initing)

• description and source-code

  updateOffsets = function (topics, initing) {
    this.topicPayloads.forEach(p => {
      if (!_.isEmpty(topics[p.topic]) && topics[p.topic][p.partition] !== undefined) {
        var offset = topics[p.topic][p.partition];
        if (offset === -1) offset = 0;
        if (!initing) p.offset = offset + 1;
        else p.offset = offset;
        this.needToCommit = true;
      }
    });
  
    if (this.options.autoCommit && !initing) {
      this.autoCommit(false, function (err) {
        err && logger.debug('auto commit offset', err);
      });
    }
  }

• example usage

  ...
      this.setOffset(topic.topic, topic.partition, offset);
      this.resume();
    });
  });
  
  // 'done' will be emit when a message fetch request complete
  this.on('done', function (topics) {
    self.updateOffsets(topics);
    if (!self.paused) {
      setImmediate(function () {
        self.fetch();
      });
    }
  });
  ...

module kafka-node.HighLevelProducer

function kafka-node.HighLevelProducer (client, options, customPartitioner)

• description and source-code

  function HighLevelProducer(client, options, customPartitioner) {
    BaseProducer.call(this, client, options, BaseProducer.PARTITIONER_TYPES.cyclic, customPartitioner);
  }

• example usage

  n/a

function kafka-node.HighLevelProducer.super_ (client, options, defaultPartitionerType, customPartitioner)

• description and source-code

  function BaseProducer(client, options, defaultPartitionerType, customPartitioner) {
    options = options || {};
  
    this.ready = false;
    this.client = client;
  
    this.requireAcks = options.requireAcks === undefined
      ? DEFAULTS.requireAcks
      : options.requireAcks;
    this.ackTimeoutMs = options.ackTimeoutMs === undefined
      ? DEFAULTS.ackTimeoutMs
      : options.ackTimeoutMs;
  
    if (customPartitioner !== undefined && options.partitionerType !== PARTITIONER_TYPES.custom) {
      throw new Error('Partitioner Type must be custom if providing a customPartitioner.');
    } else if (customPartitioner === undefined && options.partitionerType === PARTITIONER_TYPES.custom) {
      throw new Error('No customer partitioner defined');
    }
  
    var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
    // eslint-disable-next-line
    this.partitioner = new partitionerType(customPartitioner);
  
    this.connect();
  }

• example usage

  n/a

module kafka-node.HighLevelProducer.super_

function kafka-node.HighLevelProducer.super_ ()

• description and source-code

  function EventEmitter() {
    EventEmitter.init.call(this);
  }

• example usage

  n/a

module kafka-node.HighLevelProducer.super_.prototype

function kafka-node.HighLevelProducer.super_.prototype.buildPayloads (payloads, topicMetadata)

• description and source-code

  buildPayloads = function (payloads, topicMetadata) {
    const topicPartitionRequests = Object.create(null);
    payloads.forEach((p) => {
      p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition(_.map(topicMetadata[p.topic], 'partition
  '), p.key);
      p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
      let messages = _.isArray(p.messages) ? p.messages : [p.messages];
  
      messages = messages.map(function (message) {
        if (message instanceof KeyedMessage) {
          return message;
        }
        return new Message(0, 0, '', message);
      });
  
      let key = p.topic + p.partition;
      let request = topicPartitionRequests[key];
  
      if (request == null) {
        topicPartitionRequests[key] = new ProduceRequest(p.topic, p.partition, messages, p.attributes);
      } else {
        assert(request.attributes === p.attributes);
        Array.prototype.push.apply(request.messages, messages);
      }
    });
    return _.values(topicPartitionRequests);
  }

• example usage

  ...
   * @param {BaseProducer~sendCallback} cb A function to call once the send has completed
   */
  BaseProducer.prototype.send = function (payloads, cb) {
  var client = this.client;
  var requireAcks = this.requireAcks;
  var ackTimeoutMs = this.ackTimeoutMs;
  
  client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks
  , ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
  const topicPartitionRequests = Object.create(null);
  payloads.forEach((p) => {
    p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition(_.map(topicMetadata[p.topic
  ], 'partition'), p.key);
    p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  ...

function kafka-node.HighLevelProducer.super_.prototype.close (cb)

• description and source-code

  close = function (cb) {
    this.client.close(cb);
  }

• example usage

  ...
  
  ### close(force, cb)
  * `force`: **Boolean**, if set to true, it forces the consumer to commit the current offset before closing, default `false`
  
  Example
  
  ```js
  consumer.close(true, cb);
  consumer.close(cb); //force is disabled
  ```
  
  ## HighLevelConsumer
  ⚠️ ***This consumer has been deprecated in the latest version of Kafka (0.10.1) and is likely to be removed in the future. Please
   use the ConsumerGroup instead.***
  
  ### HighLevelConsumer(client, payloads, options)
  ...

function kafka-node.HighLevelProducer.super_.prototype.connect ()

• description and source-code

  connect = function () {
    // emiter...
    var self = this;
    this.ready = this.client.ready;
    if (this.ready) self.emit('ready');
    this.client.on('ready', function () {
      if (!self.ready) {
        self.ready = true;
        self.emit('ready');
      }
    });
    this.client.on('brokersChanged', function () {
      let topics = Object.keys(this.topicMetadata);
      this.refreshMetadata(topics, function (error) {
        if (error) {
          self.emit('error', error);
        }
      });
    });
    this.client.on('error', function (err) {
      self.emit('error', err);
    });
    this.client.on('close', function () {});
  }

• example usage

  ...
  }
  
  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);
  
  this.connect();
  }
  
  util.inherits(BaseProducer, events.EventEmitter);
  
  BaseProducer.prototype.connect = function () {
  // emiter...
  var self = this;
  ...

function kafka-node.HighLevelProducer.super_.prototype.createTopics (topics, async, cb)

• description and source-code

  createTopics = function (topics, async, cb) {
    if (!this.ready) {
      return cb(new Error('Producer not ready!'));
    }
  
    this.client.createTopics(topics, async, cb);
  }

• example usage

  ...
  
  ``` js
  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
  ```
  ...

function kafka-node.HighLevelProducer.super_.prototype.send (payloads, cb)

• description and source-code

  send = function (payloads, cb) {
    var client = this.client;
    var requireAcks = this.requireAcks;
    var ackTimeoutMs = this.ackTimeoutMs;
  
    client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb);
  }

• example usage

  ...
      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 `messages` attribute otherwise
   you may lose messages!
  ...

module kafka-node.IllegalGenerationError

function kafka-node.IllegalGenerationError (message)

• description and source-code

  IllegalGenerationError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.IllegalGenerationError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.InvalidConfigError

function kafka-node.InvalidConfigError (message)

• description and source-code

  InvalidConfigError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.InvalidConfigError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.InvalidConsumerOffsetError

function kafka-node.InvalidConsumerOffsetError (message, nested)

• description and source-code

  InvalidConsumerOffsetError = function (message, nested) {
    NestedError.apply(this, arguments);
  }

• example usage

  ...
    this.client.on('reconnect', function (lastError) {
  self.fetch();
    });
  
    this.on('offsetOutOfRange', topic => {
  this.pause();
  if (this.options.outOfRangeOffset === 'none') {
    this.emit('error', new errors.InvalidConsumerOffsetError(`Offset out of range
   for topic "${topic.topic}" partition ${topic.partition}`));
    return;
  }
  
  topic.time = ACCEPTED_FROM_OFFSET[this.options.outOfRangeOffset];
  
  this.getOffset().fetch([topic], (error, result) => {
    if (error) {
  ...

function kafka-node.InvalidConsumerOffsetError.super_ (message, nested)

• description and source-code

  super_ = function (message, nested) {
      this.nested = nested;
  
      if (typeof message !== 'undefined') {
          Object.defineProperty(this, 'message', {
              value: message,
              writable: true,
              enumerable: false,
              configurable: true
          });
      }
  
      Error.captureStackTrace(this, this.constructor);
      var oldStackDescriptor = Object.getOwnPropertyDescriptor(this, 'stack');
      var stackDescriptor = buildStackDescriptor(oldStackDescriptor, nested);
      Object.defineProperty(this, 'stack', stackDescriptor);
  }

• example usage

  n/a

module kafka-node.KafkaBuffer

function kafka-node.KafkaBuffer (batchSize, batchAge)

• description and source-code

  KafkaBuffer = function (batchSize, batchAge) {
    this._batch_size = batchSize;
    this._batch_age = batchAge;
    this._batch_age_timer = null;
    this._buffer = null;
  }

• example usage

  n/a

module kafka-node.KafkaBuffer.prototype

function kafka-node.KafkaBuffer.prototype._setupTimer (callback)

• description and source-code

  _setupTimer = function (callback) {
    var self = this;
  
    if (this._batch_age_timer != null) {
      clearTimeout(this._batch_age_timer);
    }
  
    this._batch_age_timer = setTimeout(function () {
      if (self._buffer && (self._buffer.length > 0)) {
        callback();
      }
    }, this._batch_age);
  }

• example usage

  ...
  }
  
  if (typeof callback !== 'undefined' && callback != null) {
    if (this._batch_size == null || this._batch_age == null ||
      (this._buffer && (this._buffer.length > this._batch_size))) {
      callback();
    } else {
      this._setupTimer(callback);
    }
  }
  };
  
  KafkaBuffer.prototype._setupTimer = function (callback) {
  var self = this;
  ...

function kafka-node.KafkaBuffer.prototype.addChunk (buffer, callback)

• description and source-code

  addChunk = function (buffer, callback) {
    if (this._buffer == null) {
      this._buffer = new Buffer(buffer);
    } else {
      this._buffer = Buffer.concat([this._buffer, buffer]);
    }
  
    if (typeof callback !== 'undefined' && callback != null) {
      if (this._batch_size == null || this._batch_age == null ||
        (this._buffer && (this._buffer.length > this._batch_size))) {
        callback();
      } else {
        this._setupTimer(callback);
      }
    }
  }

• example usage

  ...
    this.noAckBatchAge = options ? options.noAckBatchAge : null;
    this._KafkaBuffer = new KafkaBuffer(this.noAckBatchSize, this.noAckBatchAge);
  };
  
  util.inherits(BrokerTransform, Transform);
  
  BrokerTransform.prototype._transform = function (chunk, enc, done) {
    this._KafkaBuffer.addChunk(chunk, this._transformNext.bind(this));
    done();
  };
  
  BrokerTransform.prototype._transformNext = function () {
    this.push(this._KafkaBuffer.getBatch());
    this._KafkaBuffer.truncateBatch();
  };
  ...

function kafka-node.KafkaBuffer.prototype.getBatch ()

• description and source-code

  getBatch = function () {
    return this._buffer;
  }

• example usage

  ...
  
  BrokerTransform.prototype._transform = function (chunk, enc, done) {
    this._KafkaBuffer.addChunk(chunk, this._transformNext.bind(this));
    done();
  };
  
  BrokerTransform.prototype._transformNext = function () {
    this.push(this._KafkaBuffer.getBatch());
    this._KafkaBuffer.truncateBatch();
  };
  
  module.exports = BrokerTransform;
  ...

function kafka-node.KafkaBuffer.prototype.truncateBatch ()

• description and source-code

  truncateBatch = function () {
    this._buffer = null;
  }

• example usage

  ...
  BrokerTransform.prototype._transform = function (chunk, enc, done) {
    this._KafkaBuffer.addChunk(chunk, this._transformNext.bind(this));
    done();
  };
  
  BrokerTransform.prototype._transformNext = function () {
    this.push(this._KafkaBuffer.getBatch());
    this._KafkaBuffer.truncateBatch();
  };
  
  module.exports = BrokerTransform;
  ...

module kafka-node.KeyedPartitioner

function kafka-node.KeyedPartitioner ()

• description and source-code

  KeyedPartitioner = function () {}

• example usage

  n/a

function kafka-node.KeyedPartitioner.super_ ()

• description and source-code

  super_ = function () {}

• example usage

  n/a

module kafka-node.KeyedPartitioner.prototype

function kafka-node.KeyedPartitioner.prototype.getPartition (partitions, key)

• description and source-code

  getPartition = function (partitions, key) {
    key = key || '';
  
    var index = this.hashCode(key) % partitions.length;
    return partitions[index];
  }

• example usage

  ...
  
    client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
    const topicPartitionRequests = Object.create(null);
    payloads.forEach((p) => {
  p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition
  (_.map(topicMetadata[p.topic], 'partition'), p.key);
  p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  let messages = _.isArray(p.messages) ? p.messages : [p.messages];
  
  messages = messages.map(function (message) {
    if (message instanceof KeyedMessage) {
      return message;
    }
  ...

function kafka-node.KeyedPartitioner.prototype.hashCode (string)

• description and source-code

  hashCode = function (string) {
    var hash = 0;
    var length = string.length;
  
    for (var i = 0; i < length; i++) {
      hash = ((hash * 31) + string.charCodeAt(i)) & 0x7fffffff;
    }
  
    return (hash === 0) ? 1 : hash;
  }

• example usage

  ...
  
    return (hash === 0) ? 1 : hash;
  };
  
  KeyedPartitioner.prototype.getPartition = function (partitions, key) {
    key = key || '';
  
    var index = this.hashCode(key) % partitions.length;
    return partitions[index];
  };
  
  var CustomPartitioner = function (partitioner) {
    this.getPartition = partitioner;
  };
  util.inherits(CustomPartitioner, Partitioner);
  ...

module kafka-node.NotCoordinatorForGroupError

function kafka-node.NotCoordinatorForGroupError (message)

• description and source-code

  NotCoordinatorForGroupError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.NotCoordinatorForGroupError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.Offset

function kafka-node.Offset (client)

• description and source-code

  Offset = function (client) {
    var self = this;
    this.client = client;
    this.ready = this.client.ready;
    this.client.on('ready', function () {
      self.ready = true;
      self.emit('ready');
    });
    this.client.once('connect', function () {
      self.emit('connect');
    });
    this.client.on('error', function (err) {
      self.emit('error', err);
    });
  }

• example usage

  ...
  * `cb`: *Function*, the callback
  
  Example
  
  ```js
  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] } }
      });
  ```
  ...

function kafka-node.Offset.super_ ()

• description and source-code

  function EventEmitter() {
    EventEmitter.init.call(this);
  }

• example usage

  n/a

module kafka-node.Offset.prototype

function kafka-node.Offset.prototype.buildPayloads (payloads)

• description and source-code

  buildPayloads = function (payloads) {
    return payloads.map(function (p) {
      p.partition = p.partition || 0;
      p.time = p.time || Date.now();
      p.maxNum = p.maxNum || 1;
      p.metadata = 'm'; // metadata can be arbitrary
      return p;
    });
  }

• example usage

  ...
   * @param {BaseProducer~sendCallback} cb A function to call once the send has completed
   */
  BaseProducer.prototype.send = function (payloads, cb) {
  var client = this.client;
  var requireAcks = this.requireAcks;
  var ackTimeoutMs = this.ackTimeoutMs;
  
  client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks
  , ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
  const topicPartitionRequests = Object.create(null);
  payloads.forEach((p) => {
    p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition(_.map(topicMetadata[p.topic
  ], 'partition'), p.key);
    p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  ...

function kafka-node.Offset.prototype.commit (groupId, payloads, cb)

• description and source-code

  commit = function (groupId, payloads, cb) {
    if (!this.ready) {
      this.once('ready', () => this.commit(groupId, payloads, cb));
      return;
    }
    this.client.sendOffsetCommitRequest(groupId, this.buildPayloads(payloads), cb);
  }

• example usage

  ...
  Commit offset of the current topics manually, this method should be called when a consumer leaves
  
  * `cb`: **Function**, the callback
  
  Example:
  
  ``` js
  consumer.commit(function(err, data) {
  });
  ```
  
  ### setOffset(topic, partition, offset)
  Set offset of the given topic
  
  * `topic`: **String**
  ...

function kafka-node.Offset.prototype.fetch (payloads, cb)

• description and source-code

  fetch = function (payloads, cb) {
    if (!this.ready) {
      this.once('ready', () => this.fetch(payloads, cb));
      return;
    }
    this.client.sendOffsetRequest(this.buildPayloads(payloads), cb);
  }

• example usage

  ...
  
  Example
  
  ```js
  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] } }
      });
  ```
  ...

function kafka-node.Offset.prototype.fetchCommits (groupId, payloads, cb)

• description and source-code

  fetchCommits = function (groupId, payloads, cb) {
    if (!this.ready) {
      this.once('ready', () => this.fetchCommits(groupId, payloads, cb));
      return;
    }
    this.client.sendOffsetFetchRequest(groupId, this.buildPayloads(payloads), cb);
  }

• example usage

  ...
  
  Example
  
  ```js
  var kafka = require('kafka-node'),
      client = new kafka.Client(),
      offset = new kafka.Offset(client);
      offset.fetchCommits('groupId', [
          { topic: 't', partition: 0 }
      ], function (err, data) {
      });
  ```
  
  ### fetchLatestOffsets(topics, cb)
  ...

function kafka-node.Offset.prototype.fetchEarliestOffsets (topics, cb)

• description and source-code

  fetchEarliestOffsets = function (topics, cb) {
    fetchOffsets(this, topics, cb, -2);
  }

• example usage

  ...
  ### fetchEarliestOffsets(topics, cb)
  
  Example
  
  ```js
  	var partition = 0;
  	var topic = 't';
  	offset.fetchEarliestOffsets([topic], function (error, offsets) {
  		if (error)
  			return handleError(error);
  		console.log(offsets[topic][partition]);
  	});
  ```
  
  # Troubleshooting / FAQ
  ...

function kafka-node.Offset.prototype.fetchLatestOffsets (topics, cb)

• description and source-code

  fetchLatestOffsets = function (topics, cb) {
    fetchOffsets(this, topics, cb, -1);
  }

• example usage

  ...
  ### fetchLatestOffsets(topics, cb)
  
  Example
  
  ```js
  	var partition = 0;
  	var topic = 't';
  	offset.fetchLatestOffsets([topic], function (error, offsets) {
  		if (error)
  			return handleError(error);
  		console.log(offsets[topic][partition]);
  	});
  ```
  
  ### fetchEarliestOffsets(topics, cb)
  ...

module kafka-node.Producer

function kafka-node.Producer (client, options, customPartitioner)

• description and source-code

  function Producer(client, options, customPartitioner) {
    BaseProducer.call(this, client, options, BaseProducer.PARTITIONER_TYPES.default, customPartitioner);
  }

• example usage

  n/a

function kafka-node.Producer.super_ (client, options, defaultPartitionerType, customPartitioner)

• description and source-code

  function BaseProducer(client, options, defaultPartitionerType, customPartitioner) {
    options = options || {};
  
    this.ready = false;
    this.client = client;
  
    this.requireAcks = options.requireAcks === undefined
      ? DEFAULTS.requireAcks
      : options.requireAcks;
    this.ackTimeoutMs = options.ackTimeoutMs === undefined
      ? DEFAULTS.ackTimeoutMs
      : options.ackTimeoutMs;
  
    if (customPartitioner !== undefined && options.partitionerType !== PARTITIONER_TYPES.custom) {
      throw new Error('Partitioner Type must be custom if providing a customPartitioner.');
    } else if (customPartitioner === undefined && options.partitionerType === PARTITIONER_TYPES.custom) {
      throw new Error('No customer partitioner defined');
    }
  
    var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];
  
    // eslint-disable-next-line
    this.partitioner = new partitionerType(customPartitioner);
  
    this.connect();
  }

• example usage

  n/a

module kafka-node.RandomPartitioner

function kafka-node.RandomPartitioner ()

• description and source-code

  RandomPartitioner = function () {}

• example usage

  n/a

function kafka-node.RandomPartitioner.super_ ()

• description and source-code

  super_ = function () {}

• example usage

  n/a

module kafka-node.RandomPartitioner.prototype

function kafka-node.RandomPartitioner.prototype.getPartition (partitions)

• description and source-code

  getPartition = function (partitions) {
    return partitions[Math.floor(Math.random() * partitions.length)];
  }

• example usage

  ...
  
    client.sendProduceRequest(this.buildPayloads(payloads, client.topicMetadata), requireAcks, ackTimeoutMs, cb);
  };
  
  BaseProducer.prototype.buildPayloads = function (payloads, topicMetadata) {
    const topicPartitionRequests = Object.create(null);
    payloads.forEach((p) => {
  p.partition = p.hasOwnProperty('partition') ? p.partition : this.partitioner.getPartition
  (_.map(topicMetadata[p.topic], 'partition'), p.key);
  p.attributes = p.hasOwnProperty('attributes') ? p.attributes : 0;
  let messages = _.isArray(p.messages) ? p.messages : [p.messages];
  
  messages = messages.map(function (message) {
    if (message instanceof KeyedMessage) {
      return message;
    }
  ...

module kafka-node.RebalanceInProgressError

function kafka-node.RebalanceInProgressError (message)

• description and source-code

  RebalanceInProgressError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.RebalanceInProgressError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.TopicsNotExistError

function kafka-node.TopicsNotExistError (topics)

• description and source-code

  TopicsNotExistError = function (topics) {
    Error.captureStackTrace(this, this);
    this.topics = topics;
    this.message = 'The topic(s) ' + topics.toString() + ' do not exist';
  }

• example usage

  ...
   */
  Client.prototype.topicExists = function (topics, cb) {
  var notExistsTopics = [];
  var self = this;
  
  async.each(topics, checkZK, function (err) {
    if (err) return cb(err);
    if (notExistsTopics.length) return cb(new errors.TopicsNotExistError(notExistsTopics
  ));
    cb();
  });
  
  function checkZK (topic, cb) {
    self.zk.topicExists(topic, function (err, existed, topic) {
      if (err) return cb(err);
      if (!existed) notExistsTopics.push(topic);
  ...

function kafka-node.TopicsNotExistError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.UnknownMemberIdError

function kafka-node.UnknownMemberIdError (message)

• description and source-code

  UnknownMemberIdError = function (message) {
    Error.captureStackTrace(this, this);
    this.message = message;
  }

• example usage

  n/a

function kafka-node.UnknownMemberIdError.super_ ()

• description and source-code

  function Error() { [native code] }

• example usage

  n/a

module kafka-node.consumerGroupMigrator

function kafka-node.consumerGroupMigrator (consumerGroup)

• description and source-code

  function ConsumerGroupMigrator(consumerGroup) {
    EventEmitter.call(this);
    assert(consumerGroup);
    const self = this;
    this.consumerGroup = consumerGroup;
    this.client = consumerGroup.client;
    var verified = 0;
  
    if (consumerGroup.options.migrateRolling) {
      this.zk = zookeeper.createClient(consumerGroup.client.connectionString, {retries: 10});
      this.zk.on('connected', function () {
        self.filterByExistingZkTopics(function (error, topics) {
          if (error) {
            return self.emit('error', error);
          }
  
          if (topics.length) {
            self.checkForOwnersAndListenForChange(topics);
          } else {
            logger.debug('No HLC topics exist in zookeeper.');
            self.connectConsumerGroup();
          }
        });
      });
  
      this.on('noOwnersForTopics', function (topics) {
        logger.debug('No owners for topics %s reported.', topics);
        if (++verified <= NUMER_OF_TIMES_TO_VERIFY) {
          logger.debug('%s verify %d of %d HLC has given up ownership by checking again in %d', self.client.clientId, verified,
            NUMER_OF_TIMES_TO_VERIFY, VERIFY_WAIT_TIME_MS);
  
          setTimeout(function () {
            self.checkForOwners(topics);
          }, VERIFY_WAIT_TIME_MS);
        } else {
          self.connectConsumerGroup();
        }
      });
  
      this.on('topicOwnerChange', _.debounce(function (topics) {
        verified = 0;
        self.checkForOwnersAndListenForChange(topics);
      }, 250));
  
      this.zk.connect();
    } else {
      this.connectConsumerGroup();
    }
  }

• example usage

  n/a

function kafka-node.consumerGroupMigrator.super_ ()

• description and source-code

  function EventEmitter() {
    EventEmitter.init.call(this);
  }

• example usage

  n/a

module kafka-node.consumerGroupMigrator.prototype

function kafka-node.consumerGroupMigrator.prototype.checkForOwners (topics, listenForChange)

• description and source-code

  checkForOwners = function (topics, listenForChange) {
    const self = this;
    const path = '/consumers/' + this.consumerGroup.options.groupId + '/owners/';
    var ownedPartitions = 0;
  
    function topicWatcher (event) {
      self.emit('topicOwnerChange', topics);
    }
  
    async.each(topics,
      function (topic, callback) {
        const args = [path + topic];
  
        if (listenForChange) {
          logger.debug('%s listening for changes in topic %s', self.client.clientId, topic);
          args.push(topicWatcher);
        }
  
        args.push(function (error, children, stats) {
          if (error) {
            return callback(error);
          }
          ownedPartitions += children.length;
          callback(null);
        });
  
        self.zk.getChildren.apply(self.zk, args);
      },
      function (error) {
        if (error) {
          return self.emit('error', error);
        }
        if (ownedPartitions === 0) {
          self.emit('noOwnersForTopics', topics);
        } else {
          logger.debug('%s %d partitions are owned by old HLC... waiting...', self.client.clientId, ownedPartitions);
        }
      }
    );
  }

• example usage

  ...
  this.on('noOwnersForTopics', function (topics) {
    logger.debug('No owners for topics %s reported.', topics);
    if (++verified <= NUMER_OF_TIMES_TO_VERIFY) {
      logger.debug('%s verify %d of %d HLC has given up ownership by checking again in %d', self.client.clientId, verified
  ,
        NUMER_OF_TIMES_TO_VERIFY, VERIFY_WAIT_TIME_MS);
  
      setTimeout(function () {
        self.checkForOwners(topics);
      }, VERIFY_WAIT_TIME_MS);
    } else {
      self.connectConsumerGroup();
    }
  });
  
  this.on('topicOwnerChange', _.debounce(function (topics) {
  ...

function kafka-node.consumerGroupMigrator.prototype.checkForOwnersAndListenForChange (topics)

• description and source-code

  checkForOwnersAndListenForChange = function (topics) {
    this.checkForOwners(topics, true);
  }

• example usage

  ...
  this.zk.on('connected', function () {
    self.filterByExistingZkTopics(function (error, topics) {
      if (error) {
        return self.emit('error', error);
      }
  
      if (topics.length) {
        self.checkForOwnersAndListenForChange(topics);
      } else {
        logger.debug('No HLC topics exist in zookeeper.');
        self.connectConsumerGroup();
      }
    });
  });
  ...

function kafka-node.consumerGroupMigrator.prototype.connectConsumerGroup ()

• description and source-code

  connectConsumerGroup = function () {
    logger.debug('%s connecting consumer group', this.client.clientId);
    const self = this;
    if (this.client.ready) {
      this.consumerGroup.connect();
    } else {
      this.client.once('ready', function () {
        self.consumerGroup.connect();
      });
    }
    this.zk && this.zk.close();
  }

• example usage

  ...
        return self.emit('error', error);
      }
  
      if (topics.length) {
        self.checkForOwnersAndListenForChange(topics);
      } else {
        logger.debug('No HLC topics exist in zookeeper.');
        self.connectConsumerGroup();
      }
    });
  });
  
  this.on('noOwnersForTopics', function (topics) {
    logger.debug('No owners for topics %s reported.', topics);
    if (++verified <= NUMER_OF_TIMES_TO_VERIFY) {
  ...

function kafka-node.consumerGroupMigrator.prototype.filterByExistingZkTopics (callback)

• description and source-code

  filterByExistingZkTopics = function (callback) {
    const self = this;
    const path = '/consumers/' + this.consumerGroup.options.groupId + '/owners/';
  
    async.filter(this.consumerGroup.topics, function (topic, cb) {
      const topicPath = path + topic;
      logger.debug('%s checking zk path %s', self.client.clientId, topicPath);
      self.zk.exists(topicPath, function (error, stat) {
        if (error) {
          return callback(error);
        }
        cb(stat);
      });
    }, function (result) {
      callback(null, result);
    });
  }

• example usage

  ...
    this.consumerGroup = consumerGroup;
    this.client = consumerGroup.client;
    var verified = 0;
  
    if (consumerGroup.options.migrateRolling) {
      this.zk = zookeeper.createClient(consumerGroup.client.connectionString, {retries: 10});
      this.zk.on('connected', function () {
        self.filterByExistingZkTopics(function (error, topics) {
  if (error) {
    return self.emit('error', error);
  }
  
  if (topics.length) {
    self.checkForOwnersAndListenForChange(topics);
  } else {
  ...

function kafka-node.consumerGroupMigrator.prototype.getOffset (tp, defaultOffset)

• description and source-code

  getOffset = function (tp, defaultOffset) {
    const offset = _.get(this.offsets, [tp.topic, tp.partition], defaultOffset);
    if (offset === -1) {
      return defaultOffset;
    }
    return offset;
  }

• example usage

  ...
  if (this.options.outOfRangeOffset === 'none') {
    this.emit('error', new errors.InvalidConsumerOffsetError(`Offset out of range for topic "${topic.topic}" partition
   ${topic.partition}`));
    return;
  }
  
  topic.time = ACCEPTED_FROM_OFFSET[this.options.outOfRangeOffset];
  
  this.getOffset().fetch([topic], (error, result) => {
    if (error) {
      this.emit('error', new errors.InvalidConsumerOffsetError(`Fetching ${this.options.outOfRangeOffset} offset failed`,
  error));
      return;
    }
    const offset = _.head(result[topic.topic][topic.partition]);
    const oldOffset = _.find(this.topicPayloads, {topic: topic.topic, partition: topic.partition}).offset;
  ...

function kafka-node.consumerGroupMigrator.prototype.saveHighLevelConsumerOffsets (topicPartitionList, callback)

• description and source-code

  saveHighLevelConsumerOffsets = function (topicPartitionList, callback) {
    const self = this;
    this.client.sendOffsetFetchRequest(this.consumerGroup.options.groupId, topicPartitionList, function (error, results) {
      logger.debug('sendOffsetFetchRequest response:', results, error);
      if (error) {
        return callback(error);
      }
      self.offsets = results;
      callback(null);
    });
  }

• example usage

  ...
  if (self.options.fromOffset === 'none') {
    return callback(new Error(`${self.client.clientId} owns topics and partitions which contains no saved offsets for group '${
  self.options.groupId}'`));
  }
  
  async.parallel([
    function (callback) {
      if (self.migrator) {
        return self.migrator.saveHighLevelConsumerOffsets(topicPartitionList, callback);
      }
      callback(null);
    },
    function (callback) {
      if (useDefaultOffsets) {
        return self.saveDefaultOffsets(topicPartitionList, callback);
      }
  ...

module kafka-node.consumerGroupRecovery

function kafka-node.consumerGroupRecovery (consumerGroup)

• description and source-code

  function ConsumerGroupRecovery(consumerGroup) {
    this.consumerGroup = consumerGroup;
    this.options = consumerGroup.options;
  }

• example usage

  n/a

module kafka-node.consumerGroupRecovery.prototype

function kafka-node.consumerGroupRecovery.prototype.clearError ()

• description and source-code

  clearError = function () {
    this.lastError = null;
  }

• example usage

  ...
  self.connecting = false;
  self.rebalancing = false;
  if (error) {
    return self.recovery.tryToRecoverFrom(error, 'connect');
  }
  
  self.ready = true;
  self.recovery.clearError();
  
  logger.debug('generationId', self.generationId);
  
  if (startFetch) {
    self.fetch();
  }
  self.startHeartbeats();
  ...

function kafka-node.consumerGroupRecovery.prototype.getRetryTimeout (error)

• description and source-code

  getRetryTimeout = function (error) {
    assert(error);
    if (!this._timeouts) {
      this._timeouts = retry.timeouts({
        retries: this.options.retries,
        factor: this.options.retryFactor,
        minTimeout: this.options.retryMinTimeout
      });
    }
  
    if (this._retryIndex == null || this.lastError == null ||
        error.errorCode !== this.lastError.errorCode) {
      this._retryIndex = 0;
    }
  
    var index = this._retryIndex++;
    if (index >= this._timeouts.length) {
      return false;
    }
    return this._timeouts[index];
  }

• example usage

  ...
      recoverableItem.handler && recoverableItem.handler.call(this.consumerGroup, error);
      return true;
    }
    return false;
  }, this);
  
  if (retry) {
    retryTimeout = this.getRetryTimeout(error);
  }
  
  if (retry && retryTimeout) {
    logger.debug('RECOVERY from %s: %s retrying in %s ms', source, this.consumerGroup.client.clientId, retryTimeout, error
  );
    this.consumerGroup.scheduleReconnect(retryTimeout);
  } else {
    this.consumerGroup.emit('error', error);
  ...

function kafka-node.consumerGroupRecovery.prototype.tryToRecoverFrom (error, source)

• description and source-code

  tryToRecoverFrom = function (error, source) {
    this.consumerGroup.ready = false;
    this.consumerGroup.stopHeartbeats();
  
    var retryTimeout = false;
    var retry = recoverableErrors.some(function (recoverableItem) {
      if (isErrorInstanceOf(error, recoverableItem.errors)) {
        recoverableItem.handler && recoverableItem.handler.call(this.consumerGroup, error);
        return true;
      }
      return false;
    }, this);
  
    if (retry) {
      retryTimeout = this.getRetryTimeout(error);
    }
  
    if (retry && retryTimeout) {
      logger.debug('RECOVERY from %s: %s retrying in %s ms', source, this.consumerGroup.client.clientId, retryTimeout, error);
      this.consumerGroup.scheduleReconnect(retryTimeout);
    } else {
      this.consumerGroup.emit('error', error);
    }
    this.lastError = error;
  }

• example usage

  ...
  function (syncGroupResponse, callback) {
    self.handleSyncGroup(syncGroupResponse, callback);
  }
    ], function (error, startFetch) {
  self.connecting = false;
  self.rebalancing = false;
  if (error) {
    return self.recovery.tryToRecoverFrom(error, 'connect');
  }
  
  self.ready = true;
  self.recovery.clearError();
  
  logger.debug('generationId', self.generationId);
  ...

module kafka-node.logging

function kafka-node.logging.setLoggerProvider (provider)

• description and source-code

  function setLoggerProvider(provider) {
    loggerProvider = provider;
  }

• example usage

  ...
  ### How do I set a logger provider?
  
  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:
  
  ```javascript
  // 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');
  ```
  
  # Running Tests
  ...

module kafka-node.partitioner

function kafka-node.partitioner.CustomPartitioner (partitioner)

• description and source-code

  CustomPartitioner = function (partitioner) {
    this.getPartition = partitioner;
  }

• example usage

  n/a

function kafka-node.partitioner.CyclicPartitioner ()

• description and source-code

  CyclicPartitioner = function () {
    this.c = 0;
  }

• example usage

  n/a

function kafka-node.partitioner.DefaultPartitioner ()

• description and source-code

  DefaultPartitioner = function () {}

• example usage

  n/a

function kafka-node.partitioner.KeyedPartitioner ()

• description and source-code

  KeyedPartitioner = function () {}

• example usage

  n/a

function kafka-node.partitioner.RandomPartitioner ()

• description and source-code

  RandomPartitioner = function () {}

• example usage

  n/a

module kafka-node.protocol

function kafka-node.protocol.decodeFetchResponse (cb, maxTickMessages)

• description and source-code

  function decodeFetchResponse(cb, maxTickMessages) {
    return function (resp) {
      return _decodeFetchResponse(resp, cb, maxTickMessages);
    };
  }

• example usage

  ...
    broker.socket.end();
  });
  };
  
  Client.prototype.sendFetchRequest = function (consumer, payloads, fetchMaxWaitMs, fetchMinBytes, maxTickMessages) {
  var self = this;
  var encoder = protocol.encodeFetchRequest(fetchMaxWaitMs, fetchMinBytes);
  var decoder = protocol.decodeFetchResponse(function (err, type, message) {
    if (err) {
      if (err.message === 'OffsetOutOfRange') {
        return consumer.emit('offsetOutOfRange', err);
      } else if (err.message === 'NotLeaderForPartition' || err.message === 'UnknownTopicOrPartition') {
        return self.emit('brokersChanged');
      }
  ...

function kafka-node.protocol.decodeGroupCoordinatorResponse (resp)

• description and source-code

  function decodeGroupCoordinatorResponse(resp) {
    var result;
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word16bs('errorCode')
      .word32bs('coordinatorId')
      .word16bs('coordinatorHost')
      .tap(function (vars) {
        this.buffer('coordinatorHost', vars.coordinatorHost);
        vars.coordinatorHost = vars.coordinatorHost.toString();
      })
      .word32bs('coordinatorPort')
      .tap(function (vars) {
        if (vars.errorCode !== 0) {
          result = createGroupError(vars.errorCode);
          return;
        }
  
        result = {
          coordinatorHost: vars.coordinatorHost,
          coordinatorPort: vars.coordinatorPort,
          coordinatorId: vars.coordinatorId
        };
      });
    return result;
  }

• example usage

  n/a

function kafka-node.protocol.decodeGroupHeartbeat (resp)

• description and source-code

  function decodeGroupHeartbeat(resp) {
    var result = null;
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word16bs('errorCode')
      .tap(function (vars) {
        result = createGroupError(vars.errorCode);
      });
    return result;
  }

• example usage

  n/a

function kafka-node.protocol.decodeJoinGroupResponse (resp)

• description and source-code

  function decodeJoinGroupResponse(resp) {
    var result = {
      members: []
    };
  
    var error;
  
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word16bs('errorCode')
      .tap(function (vars) {
        error = createGroupError(vars.errorCode);
      })
      .word32bs('generationId')
      .tap(function (vars) {
        result.generationId = vars.generationId;
      })
      .word16bs('groupProtocol')
      .tap(function (vars) {
        this.buffer('groupProtocol', vars.groupProtocol);
        result.groupProtocol = vars.groupProtocol = vars.groupProtocol.toString();
      })
      .word16bs('leaderId')
      .tap(function (vars) {
        this.buffer('leaderId', vars.leaderId);
        result.leaderId = vars.leaderId = vars.leaderId.toString();
      })
      .word16bs('memberId')
      .tap(function (vars) {
        this.buffer('memberId', vars.memberId);
        result.memberId = vars.memberId = vars.memberId.toString();
      })
      .word32bs('memberNum')
      .loop(function (end, vars) {
        if (error) {
          return end();
        }
  
        if (vars.memberNum-- === 0) return end();
        var memberMetadata;
        this
          .word16bs('groupMemberId').tap(function (vars) {
            this.buffer('groupMemberId', vars.groupMemberId);
            vars.memberId = vars.groupMemberId.toString();
          })
          .word32bs('memberMetadata').tap(function (vars) {
            if (vars.memberMetadata > -1) {
              this.buffer('memberMetadata', vars.memberMetadata);
              memberMetadata = decodeGroupData(this.vars.memberMetadata);
              memberMetadata.id = vars.memberId;
              result.members.push(memberMetadata);
            }
          });
      });
  
    return error || result;
  }

• example usage

  n/a

function kafka-node.protocol.decodeLeaveGroupResponse (resp)

• description and source-code

  function decodeLeaveGroupResponse(resp) {
    var error = null;
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word16bs('errorCode')
      .tap(function (vars) {
        error = createGroupError(vars.errorCode);
      });
    return error;
  }

• example usage

  n/a

function kafka-node.protocol.decodeMetadataResponse (resp)

• description and source-code

  function decodeMetadataResponse(resp) {
    var brokers = {};
    var out = {};
    var topics = {};
    var errors = [];
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word32bs('brokerNum')
      .loop(decodeBrokers)
      .word32bs('topicNum')
      .loop(_decodeTopics);
  
    function decodeBrokers (end, vars) {
      if (vars.brokerNum-- === 0) return end();
      this.word32bs('nodeId')
        .word16bs('host')
        .tap(function (vars) {
          this.buffer('host', vars.host);
          vars.host = vars.host.toString();
        })
        .word32bs('port')
        .tap(function (vars) {
          brokers[vars.nodeId] = { nodeId: vars.nodeId, host: vars.host, port: vars.port };
        });
    }
  
    function _decodeTopics (end, vars) {
      if (vars.topicNum-- === 0) return end();
      this.word16bs('topicError')
        .word16bs('topic')
        .tap(function (vars) {
          this.buffer('topic', vars.topic);
          vars.topic = vars.topic.toString();
        })
        .word32bs('partitionNum')
        .tap(function (vars) {
          if (vars.topicError !== 0) {
            return errors.push(ERROR_CODE[vars.topicError]);
          }
          this.loop(decodePartitions);
        });
    }
  
    function decodePartitions (end, vars) {
      if (vars.partitionNum-- === 0) return end();
      topics[vars.topic] = topics[vars.topic] || {};
      this.word16bs('errorCode')
        .word32bs('partition')
        .word32bs('leader')
        .word32bs('replicasNum')
        .tap(function (vars) {
          var buffer = this.buffer('replicas', vars.replicasNum * 4).vars.replicas;
          this.vars.replicas = bufferToArray(vars.replicasNum, buffer);
        })
        .word32bs('isrNum')
        .tap(function (vars) {
          var buffer = this.buffer('isr', vars.isrNum * 4).vars.isr;
          this.vars.isr = bufferToArray(vars.isrNum, buffer);
          if (vars.errorCode === 0 || vars.errorCode === 9) {
            topics[vars.topic][vars.partition] =
              new PartitionMetadata(vars.topic, vars.partition, vars.leader, vars.replicas, vars.isr);
          } else {
            errors.push(ERROR_CODE[vars.errorCode]);
          }
        });
    }
  
    if (!_.isEmpty(errors)) out.error = errors;
    out.metadata = topics;
    return [brokers, out];
  }

• example usage

  n/a

function kafka-node.protocol.decodeOffsetCommitResponse (resp)

• description and source-code

  function decodeOffsetCommitResponse(resp) {
    var topics = {};
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word32bs('topicNum')
      .loop(decodeTopics(decodePartitions));
  
    function decodePartitions (end, vars) {
      if (--vars.partitionNum === 0) end();
      topics[vars.topic] = topics[vars.topic] || {};
      this.word32bs('partition')
        .word16bs('errorcode')
        .tap(function (vars) {
          topics[vars.topic]['partition'] = vars.partition;
          topics[vars.topic]['errorCode'] = vars.errorcode;
        });
    }
    return topics;
  }

• example usage

  n/a

function kafka-node.protocol.decodeOffsetFetchResponse (resp)

• description and source-code

  function decodeOffsetFetchResponse(resp) {
    var topics = {};
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word32bs('topicNum')
      .loop(decodeTopics(decodePartitions));
  
    function decodePartitions (end, vars) {
      if (--vars.partitionNum === 0) end();
      topics[vars.topic] = topics[vars.topic] || {};
      this.word32bs('partition')
        .word64bs('offset')
        .word16bs('metadata')
        .tap(function (vars) {
          if (vars.metadata === -1) {
            return;
          }
  
          this.buffer('metadata', vars.metadata);
        })
        .word16bs('errorCode')
        .tap(function (vars) {
          topics[vars.topic][vars.partition] = vars.errorCode === 0 ? vars.offset : -1;
        });
    }
    return topics;
  }

• example usage

  n/a

function kafka-node.protocol.decodeOffsetFetchV1Response (resp)

• description and source-code

  function decodeOffsetFetchV1Response(resp) {
    var topics = {};
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word32bs('topicNum')
      .loop(decodeTopics(decodePartitions));
  
    function decodePartitions (end, vars) {
      if (--vars.partitionNum === 0) end();
      topics[vars.topic] = topics[vars.topic] || {};
      this.word32bs('partition')
        .word64bs('offset')
        .word16bs('metadata')
        .tap(function (vars) {
          if (vars.metadata === -1) {
            return;
          }
  
          this.buffer('metadata', vars.metadata);
        })
        .word16bs('errorCode')
        .tap(function (vars) {
          if (vars.metadata.length === 0 && vars.offset === 0) {
            topics[vars.topic][vars.partition] = -1;
          } else {
            topics[vars.topic][vars.partition] = vars.errorCode === 0 ? vars.offset : -1;
          }
        });
    }
    return topics;
  }

• example usage

  n/a

function kafka-node.protocol.decodeOffsetResponse (resp)

• description and source-code

  function decodeOffsetResponse(resp) {
    var topics = {};
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word32bs('topicNum')
      .loop(decodeTopics(decodePartitions));
  
    function decodePartitions (end, vars) {
      if (--vars.partitionNum === 0) end();
      topics[vars.topic] = topics[vars.topic] || {};
      this.word32bs('partition')
        .word16bs('errorCode')
        .word32bs('offsetNum')
        .loop(decodeOffsets);
    }
  
    function decodeOffsets (end, vars) {
      if (--vars.offsetNum <= 0) end();
      topics[vars.topic][vars.partition] = topics[vars.topic][vars.partition] || [];
      this.word64bs('offset')
        .tap(function (vars) {
          if (vars.offset != null) topics[vars.topic][vars.partition].push(vars.offset);
        });
    }
    return topics;
  }

• example usage

  n/a

function kafka-node.protocol.decodeProduceResponse (resp)

• description and source-code

  function decodeProduceResponse(resp) {
    var topics = {};
    var error;
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word32bs('topicNum')
      .loop(decodeTopics(decodePartitions));
  
    function decodePartitions (end, vars) {
      if (--vars.partitionNum === 0) end();
      topics[vars.topic] = topics[vars.topic] || {};
      this.word32bs('partition')
        .word16bs('errorCode')
        .word64bs('offset')
        .tap(function (vars) {
          if (vars.errorCode) {
            error = new Error(ERROR_CODE[vars.errorCode]);
          } else {
            topics[vars.topic][vars.partition] = vars.offset;
          }
        });
    }
    return error || topics;
  }

• example usage

  n/a

function kafka-node.protocol.decodeSyncGroupResponse (resp)

• description and source-code

  function decodeSyncGroupResponse(resp) {
    var result;
    Binary.parse(resp)
      .word32bs('size')
      .word32bs('correlationId')
      .word16bs('errorCode')
      .tap(function (vars) {
        result = createGroupError(vars.errorCode);
      })
      .word32bs('memberAssignment')
      .tap(function (vars) {
        if (result) {
          return;
        }
        this.buffer('memberAssignment', vars.memberAssignment);
        result = decodeMemberAssignment(vars.memberAssignment);
      });
  
    return result;
  }

• example usage

  n/a

function kafka-node.protocol.encodeFetchRequest (maxWaitMs, minBytes)

• description and source-code

  function encodeFetchRequest(maxWaitMs, minBytes) {
    return function encodeFetchRequest (clientId, correlationId, payloads) {
      return _encodeFetchRequest(clientId, correlationId, payloads, maxWaitMs, minBytes);
    };
  }

• example usage

  ...
    broker.socket.closing = true;
    broker.socket.end();
  });
  };
  
  Client.prototype.sendFetchRequest = function (consumer, payloads, fetchMaxWaitMs, fetchMinBytes, maxTickMessages) {
  var self = this;
  var encoder = protocol.encodeFetchRequest(fetchMaxWaitMs, fetchMinBytes);
  var decoder = protocol.decodeFetchResponse(function (err, type, message) {
    if (err) {
      if (err.message === 'OffsetOutOfRange') {
        return consumer.emit('offsetOutOfRange', err);
      } else if (err.message === 'NotLeaderForPartition' || err.message === 'UnknownTopicOrPartition') {
        return self.emit('brokersChanged');
      }
  ...

function kafka-node.protocol.encodeGroupCoordinatorRequest (clientId, correlationId, groupId)

• description and source-code

  function encodeGroupCoordinatorRequest(clientId, correlationId, groupId) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.groupCoordinator);
    request.Int16BE(groupId.length).string(groupId);
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeGroupHeartbeat (clientId, correlationId, groupId, generationId, memberId)

• description and source-code

  function encodeGroupHeartbeat(clientId, correlationId, groupId, generationId, memberId) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.heartbeat);
    request
      .Int16BE(groupId.length).string(groupId)
      .Int32BE(generationId)
      .Int16BE(memberId.length).string(memberId);
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeJoinGroupRequest (clientId, correlationId, groupId, memberId, sessionTimeout, groupProtocols)

• description and source-code

  function encodeJoinGroupRequest(clientId, correlationId, groupId, memberId, sessionTimeout, groupProtocols) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.joinGroup);
    request
      .Int16BE(groupId.length).string(groupId)
      .Int32BE(sessionTimeout)
      .Int16BE(memberId.length).string(memberId)
      .Int16BE(GROUPS_PROTOCOL_TYPE.length).string(GROUPS_PROTOCOL_TYPE)
      .Int32BE(groupProtocols.length);
  
    groupProtocols.forEach(encodeGroupProtocol.bind(request));
  
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeLeaveGroupRequest (clientId, correlationId, groupId, memberId)

• description and source-code

  function encodeLeaveGroupRequest(clientId, correlationId, groupId, memberId) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.leaveGroup);
    request
      .Int16BE(groupId.length).string(groupId)
      .Int16BE(memberId.length).string(memberId);
  
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeMessageSet (messageSet)

• description and source-code

  function encodeMessageSet(messageSet) {
    var buffer = new Buffermaker();
    messageSet.forEach(function (message) {
      var msg = encodeMessage(message);
      buffer.Int64BE(0)
        .Int32BE(msg.length)
        .string(msg);
    });
    return buffer.make();
  }

• example usage

  n/a

function kafka-node.protocol.encodeMetadataRequest (clientId, correlationId, topics)

• description and source-code

  function encodeMetadataRequest(clientId, correlationId, topics) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.metadata);
    request.Int32BE(topics.length);
    topics.forEach(function (topic) {
      request.Int16BE(topic.length)
        .string(topic);
    });
    return encodeRequestWithLength(request.make());
  }

• example usage

  ...
   * containing the topic name, partition, leader number, replica count, and in sync replicas per partition.
   *
   * @param {Array} topics An array of topics to load the metadata for
   * @param {Client~loadMetadataForTopicsCallback} cb Function to call once all metadata is loaded
   */
  Client.prototype.loadMetadataForTopics = function (topics, cb) {
  var correlationId = this.nextId();
  var request = protocol.encodeMetadataRequest(this.clientId, correlationId, topics);
  var broker = this.brokerForLeader();
  
  if (!broker || !broker.socket || broker.socket.error || broker.socket.destroyed) {
    return cb(new errors.BrokerNotAvailableError('Broker not available'));
  }
  
  this.queueCallback(broker.socket, correlationId, [protocol.decodeMetadataResponse, cb]);
  ...

function kafka-node.protocol.encodeOffsetCommitRequest (group)

• description and source-code

  function encodeOffsetCommitRequest(group) {
    return function (clientId, correlationId, payloads) {
      return _encodeOffsetCommitRequest(clientId, correlationId, group, payloads);
    };
  }

• example usage

  ...
      payload.messages = [ new Message(0, attributes, '', message) ];
      cb();
    });
  }
  };
  
  Client.prototype.sendOffsetCommitRequest = function (group, payloads, cb) {
  var encoder = protocol.encodeOffsetCommitRequest(group);
  var decoder = protocol.decodeOffsetCommitResponse;
  this.send(payloads, encoder, decoder, cb);
  };
  
  Client.prototype.sendOffsetCommitV2Request = function (group, generationId, memberId, payloads, cb) {
  var encoder = protocol.encodeOffsetCommitV2Request;
  var decoder = protocol.decodeOffsetCommitResponse;
  ...

function kafka-node.protocol.encodeOffsetCommitV2Request (clientId, correlationId, group, generationId, memberId, payloads)

• description and source-code

  function encodeOffsetCommitV2Request(clientId, correlationId, group, generationId, memberId, payloads) {
    payloads = groupByTopic(payloads);
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.offsetCommit, 2);
    var topics = Object.keys(payloads);
  
    request.Int16BE(group.length).string(group)
      .Int32BE(generationId)
      .Int16BE(memberId.length).string(memberId)
      .Int64BE(-1)
      .Int32BE(topics.length);
  
    topics.forEach(function (topic) {
      request.Int16BE(topic.length)
        .string(topic);
  
      var partitions = _.toPairs(payloads[topic]).map(function (pairs) { return pairs[1]; });
      request.Int32BE(partitions.length);
      partitions.forEach(function (p) {
        request.Int32BE(p.partition)
          .Int64BE(p.offset)
          .Int16BE(p.metadata.length)
          .string(p.metadata);
      });
    });
  
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeOffsetFetchRequest (group)

• description and source-code

  function encodeOffsetFetchRequest(group) {
    return function (clientId, correlationId, payloads) {
      return _encodeOffsetFetchRequest(clientId, correlationId, group, payloads);
    };
  }

• example usage

  ...
  Client.prototype.sendOffsetFetchV1Request = function (group, payloads, cb) {
  var encoder = protocol.encodeOffsetFetchV1Request;
  var decoder = protocol.decodeOffsetFetchV1Response;
  this.sendGroupRequest(encoder, decoder, arguments);
  };
  
  Client.prototype.sendOffsetFetchRequest = function (group, payloads, cb) {
  var encoder = protocol.encodeOffsetFetchRequest(group);
  var decoder = protocol.decodeOffsetFetchResponse;
  this.send(payloads, encoder, decoder, cb);
  };
  
  Client.prototype.sendOffsetRequest = function (payloads, cb) {
  var encoder = protocol.encodeOffsetRequest;
  var decoder = protocol.decodeOffsetResponse;
  ...

function kafka-node.protocol.encodeOffsetFetchV1Request (clientId, correlationId, group, payloads)

• description and source-code

  function encodeOffsetFetchV1Request(clientId, correlationId, group, payloads) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.offsetFetch, 1);
    var topics = Object.keys(payloads);
  
    request.Int16BE(group.length)
      .string(group)
      .Int32BE(topics.length);
  
    topics.forEach(function (topic) {
      request.Int16BE(topic.length).string(topic)
        .Int32BE(payloads[topic].length);
  
      payloads[topic].forEach(function (p) {
        request.Int32BE(p);
      });
    });
  
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeOffsetRequest (clientId, correlationId, payloads)

• description and source-code

  function encodeOffsetRequest(clientId, correlationId, payloads) {
    payloads = groupByTopic(payloads);
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.offset);
    var topics = Object.keys(payloads);
  
    request.Int32BE(REPLICA_ID)
      .Int32BE(topics.length);
  
    topics.forEach(function (topic) {
      request.Int16BE(topic.length)
        .string(topic);
  
      var partitions = _.toPairs(payloads[topic]).map(function (pairs) { return pairs[1]; });
      request.Int32BE(partitions.length);
      partitions.forEach(function (p) {
        request.Int32BE(p.partition)
          .Int64BE(p.time)
          .Int32BE(p.maxNum);
      });
    });
  
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

function kafka-node.protocol.encodeProduceRequest (requireAcks, ackTimeoutMs)

• description and source-code

  function encodeProduceRequest(requireAcks, ackTimeoutMs) {
    return function (clientId, correlationId, payloads) {
      return _encodeProduceRequest(clientId, correlationId, payloads, requireAcks, ackTimeoutMs);
    };
  }

• example usage

  ...
      Array.prototype.unshift.call(arguments, 'error');
      consumer.emit.apply(consumer, arguments);
    }
  });
  };
  
  Client.prototype.sendProduceRequest = function (payloads, requireAcks, ackTimeoutMs, cb) {
  var encoder = protocol.encodeProduceRequest(requireAcks, ackTimeoutMs);
  var decoder = protocol.decodeProduceResponse;
  var self = this;
  
  decoder.requireAcks = requireAcks;
  
  async.each(payloads, buildRequest, function (err) {
    if (err) return cb(err);
  ...

function kafka-node.protocol.encodeSyncGroupRequest (clientId, correlationId, groupId, generationId, memberId, groupAssignment)

• description and source-code

  function encodeSyncGroupRequest(clientId, correlationId, groupId, generationId, memberId, groupAssignment) {
    var request = encodeRequestHeader(clientId, correlationId, REQUEST_TYPE.syncGroup);
    request
      .Int16BE(groupId.length).string(groupId)
      .Int32BE(generationId)
      .Int16BE(memberId.length).string(memberId);
  
    if (groupAssignment && groupAssignment.length) {
      request.Int32BE(groupAssignment.length);
      groupAssignment.forEach(function (assignment) {
        request.Int16BE(assignment.memberId.length).string(assignment.memberId)
          .string(_encodeMemberAssignment(assignment));
      });
    } else {
      request.Int32BE(0);
    }
  
    return encodeRequestWithLength(request.make());
  }

• example usage

  n/a

module kafka-node.protocol_struct

function kafka-node.protocol_struct.FetchRequest ()

• description and source-code

  FetchRequest = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.FetchResponse ()

• description and source-code

  FetchResponse = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.KeyedMessage (key, value)

• description and source-code

  function KeyedMessage(key, value) {
    exports.Message.call(this, 0, 0, key, value);
  }

• example usage

  n/a

function kafka-node.protocol_struct.Message ()

• description and source-code

  Message = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.OffsetCommitRequest ()

• description and source-code

  OffsetCommitRequest = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.OffsetCommitResponse ()

• description and source-code

  OffsetCommitResponse = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.PartitionMetadata ()

• description and source-code

  PartitionMetadata = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.ProduceRequest ()

• description and source-code

  ProduceRequest = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.Request ()

• description and source-code

  Request = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

function kafka-node.protocol_struct.TopicAndPartition ()

• description and source-code

  TopicAndPartition = function () {
    for (var i = 0; i < args.length; i++) {
      this[args[i]] = arguments[i];
    }
  }

• example usage

  n/a

module kafka-node.range

function kafka-node.range.assign (topicPartition, groupMembers, callback)

• description and source-code

  function assignRange(topicPartition, groupMembers, callback) {
    logger.debug('topicPartition: %j', topicPartition);
    var assignment = _(groupMembers).map('id').reduce(function (obj, id) {
      obj[id] = [];
      return obj;
    }, {});
  
    const topicMemberMap = topicToMemberMap(groupMembers);
    for (var topic in topicMemberMap) {
      if (!topicMemberMap.hasOwnProperty(topic)) {
        continue;
      }
      logger.debug('For topic %s', topic);
      topicMemberMap[topic].sort();
      logger.debug('   members: ', topicMemberMap[topic]);
  
      var numberOfPartitionsForTopic = topicPartition[topic].length;
      logger.debug('   numberOfPartitionsForTopic', numberOfPartitionsForTopic);
  
      var numberOfMembersForTopic = topicMemberMap[topic].length;
      logger.debug('   numberOfMembersForTopic', numberOfMembersForTopic);
  
      var numberPartitionsPerMember = Math.floor(numberOfPartitionsForTopic / numberOfMembersForTopic);
      logger.debug('   numberPartitionsPerMember', numberPartitionsPerMember);
  
      var membersWithExtraPartition = numberOfPartitionsForTopic % numberOfMembersForTopic;
      var topicPartitionList = createTopicPartitionArray(topic, numberOfPartitionsForTopic);
  
      for (var i = 0, n = numberOfMembersForTopic; i < n; i++) {
        var start = numberPartitionsPerMember * i + Math.min(i, membersWithExtraPartition);
        var length = numberPartitionsPerMember + (i + 1 > membersWithExtraPartition ? 0 : 1);
        var assignedTopicPartitions = assignment[topicMemberMap[topic][i]];
        assignedTopicPartitions.push.apply(assignedTopicPartitions, topicPartitionList.slice(start, start + length));
      }
    }
  
    logger.debug(assignment);
  
    callback(null, convertToAssignmentList(assignment, VERSION));
  }

• example usage

  ...
  }
  
  this.protocols = protocols.map(function (protocol) {
    if (typeof protocol === 'string') {
      if (!(protocol in builtInProtocols)) {
        throw new Error('Unknown built in assignment protocol ' + protocol);
      }
      protocol = _.assign({}, builtInProtocols[protocol]);
    } else {
      checkProtocol(protocol);
    }
  
    protocol.subscription = this.topics;
    return protocol;
  }, this);
  ...

module kafka-node.roundrobin

function kafka-node.roundrobin.assign (topicPartition, groupMembers, callback)

• description and source-code

  function assignRoundRobin(topicPartition, groupMembers, callback) {
    logger.debug('topicPartition: %j', topicPartition);
    logger.debug('groupMembers: %j', groupMembers);
    var _members = _(groupMembers).map('id');
    var members = _members.value().sort();
    logger.debug('members', members);
    var assignment = _members.reduce(function (obj, id) {
      obj[id] = [];
      return obj;
    }, {});
  
    var subscriberMap = groupMembers.reduce(function (subscribers, member) {
      subscribers[member.id] = member.subscription;
      return subscribers;
    }, {});
  
    logger.debug('subscribers', subscriberMap);
  
    // layout topic/partitions pairs into a list
    var topicPartitionList = _(topicPartition).map(function (partitions, topic) {
      return partitions.map(function (partition) {
        return {
          topic: topic,
          partition: partition
        };
      });
    }).flatten().value();
    logger.debug('round robin on topic partition pairs: ', topicPartitionList);
  
    var assigner = cycle(members);
  
    topicPartitionList.forEach(function (tp) {
      var topic = tp.topic;
      while (!_.includes(subscriberMap[assigner.peek()], topic)) {
        assigner.next();
      }
      assignment[assigner.next()].push(tp);
    });
  
    var ret = _.map(assignment, function (value, key) {
      var ret = {};
      ret.memberId = key;
      ret.topicPartitions = groupPartitionsByTopic(value);
      ret.version = VERSION;
      return ret;
    });
  
    callback(null, ret);
  }

• example usage

  ...
  }
  
  this.protocols = protocols.map(function (protocol) {
    if (typeof protocol === 'string') {
      if (!(protocol in builtInProtocols)) {
        throw new Error('Unknown built in assignment protocol ' + protocol);
      }
      protocol = _.assign({}, builtInProtocols[protocol]);
    } else {
      checkProtocol(protocol);
    }
  
    protocol.subscription = this.topics;
    return protocol;
  }, this);
  ...

module kafka-node.snappy

function kafka-node.snappy.decode (buffer, cb)

• description and source-code

  function decodeSnappy(buffer, cb) {
    if (isChunked(buffer)) {
      var pos = 16;
      var max = buffer.length;
      var encoded = [];
      var size;
  
      while (pos < max) {
        size = buffer.readUInt32BE(pos);
        pos += 4;
        encoded.push(buffer.slice(pos, pos + size));
        pos += size;
      }
      return async.mapSeries(encoded, snappy.uncompress,
        function (err, decodedChunks) {
          if (err) return cb(err);
          return cb(null, Buffer.concat(decodedChunks));
        }
      );
    }
    return snappy.uncompress(buffer, cb);
  }

• example usage

  ...
            value: vars.value,
            offset: vars.offset,
            partition: partition,
            highWaterOffset: highWaterOffset,
            key: vars.key
          });
        }
        codec.decode(vars.value, function (error, inlineMessageSet) {
          if (error) return; // Not sure where to report this
          decodeMessageSet(topic, partition, inlineMessageSet, cb, maxTickMessages);
        });
      }
    });
  // Defensive code around potential denial of service
  if (maxTickMessages && messageCount > maxTickMessages) break;
  ...

function kafka-node.snappy.encode (input, callback)

• description and source-code

  encode = function (input, callback) {
    if (!(typeof (input) === 'string' || Buffer.isBuffer(input))) {
      return callback(new Error('input must be a String or a Buffer'));
    }
  
    binding.compress(input, callback);
  }

• example usage

  ...
    function buildRequest (payload, cb) {
      var attributes = payload.attributes;
      var codec = getCodec(attributes);
  
      if (!codec) return cb();
  
      var innerSet = encodeMessageSet(payload.messages);
      codec.encode(innerSet, function (err, message) {
        if (err) return cb(err);
        payload.messages = [ new Message(0, attributes, '', message) ];
        cb();
      });
    }
  };
  ...

module kafka-node.utils

function kafka-node.utils.createTopicPartitionList (topicPartitions)

• description and source-code

  function createTopicPartitionList(topicPartitions) {
    var tpList = [];
    for (var topic in topicPartitions) {
      if (!topicPartitions.hasOwnProperty(topic)) {
        continue;
      }
      topicPartitions[topic].forEach(function (partition) {
        tpList.push({
          topic: topic,
          partition: partition
        });
      });
    }
    return tpList;
  }

• example usage

  n/a

function kafka-node.utils.groupPartitionsByTopic (topicPartitions)

• description and source-code

  function groupPartitionsByTopic(topicPartitions) {
    assert(Array.isArray(topicPartitions));
    return topicPartitions.reduce(function (result, tp) {
      if (!(tp.topic in result)) {
        result[tp.topic] = [tp.partition];
      } else {
        result[tp.topic].push(tp.partition);
      }
      return result;
    }, {});
  }

• example usage

  n/a

function kafka-node.utils.validateConfig (property, value)

• description and source-code

  function validateConfig(property, value) {
    if (!legalChars.test(value)) {
      throw new InvalidConfigError([property, value, "is illegal, contains a character other than ASCII alphanumerics, '.', '_' and
   '-'"].join(' '));
    }
  }

• example usage

  ...
  this.paused = this.options.paused;
  this.id = nextId();
  this.payloads = this.buildPayloads(topics);
  this.connect();
  this.encoding = this.options.encoding;
  
  if (this.options.groupId) {
    utils.validateConfig('options.groupId', this.options.groupId);
  }
  };
  util.inherits(Consumer, events.EventEmitter);
  
  Consumer.prototype.buildPayloads = function (payloads) {
  var self = this;
  return payloads.map(function (p) {
  ...

function kafka-node.utils.validateTopicNames (topics)

• description and source-code

  function validateTopicNames(topics) {
    // Rewriting same validations done by Apache Kafka team for topics
    // iterating over topics
    topics.some(function (topic) {
      if (topic.length <= 0) {
        throw new InvalidConfigError('topic name is illegal, cannot be empty');
      }
      if (topic === '.' || topic === '..') {
        throw new InvalidConfigError('topic name cannot be . or ..');
      }
      if (topic.length > allowedTopicLength) {
        throw new InvalidConfigError(`topic name is illegal, cannot be longer than ${allowedTopicLength} characters`);
      }
      if (!legalChars.test(topic)) {
        throw new InvalidConfigError(`topic name ${topic} is illegal, contains a character other than ASCII alphanumerics .,_ and -`);
      }
    });
    return true;
  }

• example usage

  n/a

function kafka-node.utils.validateTopics (topics)

• description and source-code

  function validateTopics(topics) {
    if (topics.some(function (topic) {
      if ('partition' in topic) {
        return typeof topic.partition !== 'number';
      }
      return false;
    })) {
      throw new InvalidConfigError('Offset must be a number and can not contain characters');
    }
  }

• example usage

  ...
  })();
  
  var Consumer = function (client, topics, options) {
  if (!topics) {
    throw new Error('Must have payloads');
  }
  
  utils.validateTopics(topics);
  
  this.fetchCount = 0;
  this.client = client;
  this.options = _.defaults((options || {}), DEFAULTS);
  this.ready = false;
  this.paused = this.options.paused;
  this.id = nextId();
  ...

module kafka-node.zookeeper

function kafka-node.zookeeper.Zookeeper (connectionString, options)

• description and source-code

  Zookeeper = function (connectionString, options) {
    this.client = zookeeper.createClient(connectionString, options);
  
    var that = this;
    this.client.on('connected', function () {
      that.listBrokers();
    });
    this.client.on('disconnected', function () {
      that.emit('disconnected');
    });
    this.client.connect();
  }

• example usage

  n/a

function kafka-node.zookeeper.ZookeeperConsumerMappings ()

• description and source-code

  ZookeeperConsumerMappings = function () {
    this.consumerTopicMap = {};
    this.topicConsumerMap = {};
    this.topicPartitionMap = {};
  }

• example usage

  n/a