lib/kafka-consumer.js
function KafkaConsumer(conf, topicConf) {
  if (!(this instanceof KafkaConsumer)) {
    return new KafkaConsumer(conf, topicConf);
  }

  conf = shallowCopy(conf);
  topicConf = shallowCopy(topicConf);

  var onRebalance = conf.rebalance_cb;

  var self = this;

  // If rebalance is undefined we don't want any part of this
  if (onRebalance && typeof onRebalance === 'boolean') {
    conf.rebalance_cb = function(err, assignment) {
      // Create the librdkafka error
      err = LibrdKafkaError.create(err);
      // Emit the event
      self.emit('rebalance', err, assignment);

      // That's it
      try {
        if (err.code === -175 /*ERR__ASSIGN_PARTITIONS*/) {
          self.assign(assignment);
        } else if (err.code === -174 /*ERR__REVOKE_PARTITIONS*/) {
          self.unassign();
        }
      } catch (e) {
        // Ignore exceptions if we are not connected
        if (self.isConnected()) {
          self.emit('rebalance.error', e);
        }
      }
    };
  } else if (onRebalance && typeof onRebalance === 'function') {
    /*
     * Once this is opted in to, that's it. It's going to manually rebalance
     * forever. There is no way to unset config values in librdkafka, just
     * a way to override them.
     */

     conf.rebalance_cb = function(err, assignment) {
       // Create the librdkafka error
       err = err ? LibrdKafkaError.create(err) : undefined;

       self.emit('rebalance', err, assignment);
       onRebalance.call(self, err, assignment);
     };
  }

  // Same treatment for offset_commit_cb
  var onOffsetCommit = conf.offset_commit_cb;

  if (onOffsetCommit && typeof onOffsetCommit === 'boolean') {
    conf.offset_commit_cb = function(err, offsets) {
      // Emit the event
      self.emit('offset.commit', offsets);
    };
  } else if (onOffsetCommit && typeof onOffsetCommit === 'function') {
    conf.offset_commit_cb = function(err, offsets) {
      // Emit the event
      self.emit('offset.commit', offsets);
      onOffsetCommit.call(self, err, offsets);
    };
  }

  /**
   * KafkaConsumer message.
   *
   * This is the representation of a message read from Kafka.
   *
   * @typedef {object} KafkaConsumer~Message
   * @property {buffer} value - the message buffer from Kafka.
   * @property {string} topic - the topic name
   * @property {number} partition - the partition on the topic the
   * message was on
   * @property {number} offset - the offset of the message
   * @property {string} key - the message key
   * @property {number} size - message size, in bytes.
   * @property {number} timestamp - message timestamp
   */

  Client.call(this, conf, Kafka.KafkaConsumer, topicConf);

  this.globalConfig = conf;
  this.topicConfig = topicConf;

  this._consumeTimeout = 1000;
}
function KafkaConsumer(conf, topicConf) {
  if (!(this instanceof KafkaConsumer)) {
    return new KafkaConsumer(conf, topicConf);
  }

  conf = shallowCopy(conf);
  topicConf = shallowCopy(topicConf);

  var onRebalance = conf.rebalance_cb;

  var self = this;

  // If rebalance is undefined we don't want any part of this
  if (onRebalance && typeof onRebalance === 'boolean') {
    conf.rebalance_cb = function(err, assignment) {
      // Create the librdkafka error
      err = LibrdKafkaError.create(err);
      // Emit the event
      self.emit('rebalance', err, assignment);

      // That's it
      try {
        if (err.code === -175 /*ERR__ASSIGN_PARTITIONS*/) {
          self.assign(assignment);
        } else if (err.code === -174 /*ERR__REVOKE_PARTITIONS*/) {
          self.unassign();
        }
      } catch (e) {
        // Ignore exceptions if we are not connected
        if (self.isConnected()) {
          self.emit('rebalance.error', e);
        }
      }
    };
  } else if (onRebalance && typeof onRebalance === 'function') {
    /*
     * Once this is opted in to, that's it. It's going to manually rebalance
     * forever. There is no way to unset config values in librdkafka, just
     * a way to override them.
     */

     conf.rebalance_cb = function(err, assignment) {
       // Create the librdkafka error
       err = err ? LibrdKafkaError.create(err) : undefined;

       self.emit('rebalance', err, assignment);
       onRebalance.call(self, err, assignment);
     };
  }

  // Same treatment for offset_commit_cb
  var onOffsetCommit = conf.offset_commit_cb;

  if (onOffsetCommit && typeof onOffsetCommit === 'boolean') {
    conf.offset_commit_cb = function(err, offsets) {
      // Emit the event
      self.emit('offset.commit', offsets);
    };
  } else if (onOffsetCommit && typeof onOffsetCommit === 'function') {
    conf.offset_commit_cb = function(err, offsets) {
      // Emit the event
      self.emit('offset.commit', offsets);
      onOffsetCommit.call(self, err, offsets);
    };
  }

  /**
   * KafkaConsumer message.
   *
   * This is the representation of a message read from Kafka.
   *
   * @typedef {object} KafkaConsumer~Message
   * @property {buffer} value - the message buffer from Kafka.
   * @property {string} topic - the topic name
   * @property {number} partition - the partition on the topic the
   * message was on
   * @property {number} offset - the offset of the message
   * @property {string} key - the message key
   * @property {number} size - message size, in bytes.
   * @property {number} timestamp - message timestamp
   */

  Client.call(this, conf, Kafka.KafkaConsumer, topicConf);

  this.globalConfig = conf;
  this.topicConfig = topicConf;

  this._consumeTimeout = 1000;
}
function KafkaConsumer ( conf , topicConf ) { if ( ! ( this instanceof KafkaConsumer ) ) { return new KafkaConsumer ( conf , topicConf ) ; } conf = shallowCopy ( conf ) ; topicConf = shallowCopy ( topicConf ) ; var onRebalance = conf . rebalance_cb ; var self = this ; // If rebalance is undefined we don't want any part of this if ( onRebalance && typeof onRebalance === 'boolean' ) { conf . rebalance_cb = function ( err , assignment ) { // Create the librdkafka error err = LibrdKafkaError . create ( err ) ; // Emit the event self . emit ( 'rebalance' , err , assignment ) ; // That's it try { if ( err . code === - 175 /*ERR__ASSIGN_PARTITIONS*/ ) { self . assign ( assignment ) ; } else if ( err . code === - 174 /*ERR__REVOKE_PARTITIONS*/ ) { self . unassign ( ) ; } } catch ( e ) { // Ignore exceptions if we are not connected if ( self . isConnected ( ) ) { self . emit ( 'rebalance.error' , e ) ; } } } ; } else if ( onRebalance && typeof onRebalance === 'function' ) { /*
     * Once this is opted in to, that's it. It's going to manually rebalance
     * forever. There is no way to unset config values in librdkafka, just
     * a way to override them.
     */ conf . rebalance_cb = function ( err , assignment ) { // Create the librdkafka error err = err ? LibrdKafkaError . create ( err ) : undefined ; self . emit ( 'rebalance' , err , assignment ) ; onRebalance . call ( self , err , assignment ) ; } ; } // Same treatment for offset_commit_cb var onOffsetCommit = conf . offset_commit_cb ; if ( onOffsetCommit && typeof onOffsetCommit === 'boolean' ) { conf . offset_commit_cb = function ( err , offsets ) { // Emit the event self . emit ( 'offset.commit' , offsets ) ; } ; } else if ( onOffsetCommit && typeof onOffsetCommit === 'function' ) { conf . offset_commit_cb = function ( err , offsets ) { // Emit the event self . emit ( 'offset.commit' , offsets ) ; onOffsetCommit . call ( self , err , offsets ) ; } ; } /**
   * KafkaConsumer message.
   *
   * This is the representation of a message read from Kafka.
   *
   * @typedef {object} KafkaConsumer~Message
   * @property {buffer} value - the message buffer from Kafka.
   * @property {string} topic - the topic name
   * @property {number} partition - the partition on the topic the
   * message was on
   * @property {number} offset - the offset of the message
   * @property {string} key - the message key
   * @property {number} size - message size, in bytes.
   * @property {number} timestamp - message timestamp
   */ Client . call ( this , conf , Kafka . KafkaConsumer , topicConf ) ; this . globalConfig = conf ; this . topicConfig = topicConf ; this . _consumeTimeout = 1000 ; }
KafkaConsumer class for reading messages from Kafka

This is the main entry point for reading data from Kafka. You
configure this like you do any other client, with a global
configuration and default topic configuration.

Once you instantiate this object, connecting will open a socket.
Data will not be read until you tell the consumer what topics
you want to read from.

@param {object} conf - Key value pairs to configure the consumer
@param {object} topicConf - Key value pairs to create a default
topic configuration
@extends Client
@constructor
KafkaConsumer class for reading messages from Kafka