lib/kafka/Consumer.js
/**
   * Handles the kafka's rebalance event.
   *
   * Once a rebalance occurs, it's no longer needed to handle the remaining queued
   * messages, as all uncommitted messages will be redelivered to the new assignee
   * that is part of the consumer group. Consequently, the processing queue is drained
   * and the uncommitted offsets are discarded.
   *
   * ATTENTION: A message might be processed more than once if a rebalance occurs
   * and some processed messages has not be committed.
   * @access private
   * @param {*} error the error code
   * @param {*} assignments the assignments
   */
Handles the kafka's rebalance event.
Once a rebalance occurs, it's no longer needed to handle the remaining queued
messages, as all uncommitted messages will be redelivered to the new assignee
that is part of the consumer group. Consequently, the processing queue is drained
and the uncommitted offsets are discarded.

A message might be processed more than once if a rebalance occurs
and some processed messages has not be committed.
@access private
@param {*} error the error code
@param {*} assignments the assignments
Handles the kafka ' s rebalance event . Once a rebalance occurs it ' s no longer needed to handle the remaining queued messages as all uncommitted messages will be redelivered to the new assignee that is part of the consumer group . Consequently the processing queue is drained and the uncommitted offsets are discarded . A message might be processed more than once if a rebalance occurs and some processed messages has not be committed . @access private @param { * } error the error code @param { * } assignments the assignments
async onRebalance(error, assignments) {
    if (error.code === Kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) {
      this.consumer.assign(assignments);
    } else if (error.code === Kafka.CODES.ERRORS.ERR__REVOKE_PARTITIONS) {
      this.resumeConsumer();
      this.msgQueue.remove(() => true);
      this.unackMsgBytes = 0;
      this.epoch = (this.epoch + 1) % Number.MAX_SAFE_INTEGER;
      this.consumer.unassign();
      this.commitManager.onRebalance();
    } else {
      logger.warn(`Rebalance error : ${error}`);
    }
  }
async onRebalance ( error , assignments ) { if ( error . code === Kafka . CODES . ERRORS . ERR__ASSIGN_PARTITIONS ) { this . consumer . assign ( assignments ) ; } else if ( error . code === Kafka . CODES . ERRORS . ERR__REVOKE_PARTITIONS ) { this . resumeConsumer ( ) ; this . msgQueue . remove ( ( ) => true ) ; this . unackMsgBytes = 0 ; this . epoch = ( this . epoch + 1 ) % Number . MAX_SAFE_INTEGER ; this . consumer . unassign ( ) ; this . commitManager . onRebalance ( ) ; } else { logger . warn ( ` ${ error } ` ) ; } }
Handles the kafka's rebalance event.
Handles the kafka ' s rebalance event .
// when partition are revoked we just abort queued tasks and do not // commit any processed task that is waiting for be committed into Kafka.