lib/kafka/Consumer.js
/**
   * Given a kafka message this method invokes all interested callbacks based on
   * the message's topic and mark it to be committed in an asynchronous way.
   *
   * If a processing callback throws an exception, the callback will be recalled
   * by n-times according its configuration. If the callback
   * has not succeeded after all retries, an 'error.processing' event is emitted
   * and the consumer is finished up.
   *
   * It worths to say that in the asynchronous mode the messages are marked to be
   * committed after receiving the acknowledgments.
   *
   * @access private
   * @param {*} data the kafka message with the following attributes:
   * {
   *   value: Buffer.from('konnichiwa'), // message contents as a Buffer
   *   size: 10, // size of the message, in bytes
   *   topic: 'greetings', // topic the message comes from
   *   offset: 1337, // offset the message was read from
   *   partition: 1, // partition the message was on
   *   key: 'someKey', // key of the message if present
   *   timestamp: 1510325354780, // timestamp of message creation
   * }
   */
Given a kafka message this method invokes all interested callbacks based on
the message's topic and mark it to be committed in an asynchronous way.

If a processing callback throws an exception, the callback will be recalled
by n-times according its configuration. If the callback
has not succeeded after all retries, an 'error.processing' event is emitted
and the consumer is finished up.

It worths to say that in the asynchronous mode the messages are marked to be
committed after receiving the acknowledgments.


Given a kafka message this method invokes all interested callbacks based on the message ' s topic and mark it to be committed in an asynchronous way . If a processing callback throws an exception the callback will be recalled by n - times according its configuration . If the callback has not succeeded after all retries an ' error . processing ' event is emitted and the consumer is finished up . It worths to say that in the asynchronous mode the messages are marked to be committed after receiving the acknowledgments .
async invokeInterestedCallbacksAsyncCommit(data) {
    const { epoch } = this;
     Notice that multiple callbacks might be
     called to the same message, and when the counter
     becomes 0 is safe to mark the message to
     be committed on Kafka (all acks have been
     received). Before calling the callbacks
     the counter is set to 1 and is incremented
     for each callback and decremented for each
     corresponding ack. It is also decremented after
     calling all callbacks.
     Incrementing before calling the callbacks and
     decrementing after calling all callbacks guarantees
     that the counter doesn't become 0 before all
     callbacks are called.
    let unackCounter = 1;
     acknowledge callback
     It is supposed that the ack will be called if
     and only if its corresponding callback succeeded.
    const ack = () => {
      if (epoch !== this.epoch) {
        return false;
      }
      unackCounter -= 1;
      if (unackCounter === 0) {
        this.commitManager.notifyFinishedProcessing(data);
        this.unackMsgBytes -= data.size;
      }
      return true;
    };
     set backpressure control
    this.unackMsgBytes += data.size;
      true if any callback has failed
    let anyProcessingFailure = false;
    try {
       verifies if the topic does not matches with a regular expression
      this.topicRegExpArray.forEach(async (entry) => {
        if (entry.regExp.test(data.topic)) {
          let retry = 0;
          let succeeded = false;
          unackCounter += 1;
           called is a guard to guarantee that ack
           is called once by each callback
          let called = false;
          const ackOnce = () => {
            if (!called) {
              called = true;
              return ack();
            }
            return false;
          };
          do {
            logger.debug(
              `Message on topic: ${data.topic}. `
                + `Calling callback: ${entry.id}. `
                + `Attempt: ${retry + 1}/${
                  this.config['max.retries.processing.callbacks']
                }`,
            );
            try {
              await entry.callback(data, ackOnce);
              succeeded = true;
            } catch (error) {
              logger.warn(
                `Error on user's callback ${entry.id} topic: ${
                  data.topic
                }: ${error}. Attempt: ${retry + 1}`,
              );
              retry += 1;
            }
          } while (
            !succeeded
            && retry <= this.config['max.retries.processing.callbacks']
          );
          if (!succeeded) {
            logger.error(
              `Callback ${entry.id} failed in all attempts (${retry})`,
            );
            unackCounter -= 1;
            anyProcessingFailure = true;
            this.eventEmitter.emit('error.processing', entry.id, data);
          }
        }
      });
       checks if there isn't a handler to the topic explicitly
      if (!this.topicMap[data.topic]) {
         just skip
        return;
      }
       iterates by the callbacks
      this.topicMap[data.topic].forEach(async (entry) => {
        let retry = 0;
        let succeeded = false;
        unackCounter += 1;
         called is guard to guarantee that ack
         is called once by each callback
        let called = false;
        const ackOnce = () => {
          if (!called) {
            called = true;
            return ack();
          }
          return false;
        };
        do {
          logger.debug(
            `Message on topic: ${data.topic}. `
              + `Calling callback: ${entry.id}. `
              + `Attempt: ${retry + 1}/${
                this.config['max.retries.processing.callbacks']
              }`,
          );
          try {
            await entry.callback(data, ackOnce);
            succeeded = true;
          } catch (error) {
            logger.warn(
              `Error on user's callback ${entry.id} topic: ${
                data.topic
              }: ${error}. Attempt: ${retry + 1}`,
            );
            retry += 1;
          }
        } while (
          !succeeded
          && retry <= this.config['max.retries.processing.callbacks']
        );
        if (!succeeded) {
          logger.error(
            `Callback ${entry.id} failed in all attempts (${retry})`,
          );
          unackCounter -= 1;
          anyProcessingFailure = true;
          this.eventEmitter.emit('error.processing', entry.id, data);
        }
      });
    } catch (error) {
      logger.warn(`Internal error during processing message: ${error}`);
      this.eventEmitter.emit('error.processing', null, data);
    } finally {
      unackCounter -= 1;
      if (anyProcessingFailure) {
        this.unackMsgBytes -= data.size;
        logger.error(
          'Consumer cannot keep working properly. Finishing consumer.',
        );
        this.finish();
      } else if (unackCounter === 0) {
        this.unackMsgBytes -= data.size;
        this.commitManager.notifyFinishedProcessing(data);
      }
    }
  }
async invokeInterestedCallbacksAsyncCommit ( data ) { const { epoch } = this ; let unackCounter = 1 ; const ack = ( ) => { if ( epoch !== this . epoch ) { return false ; } unackCounter -= 1 ; if ( unackCounter === 0 ) { this . commitManager . notifyFinishedProcessing ( data ) ; this . unackMsgBytes -= data . size ; } return true ; } ; this . unackMsgBytes += data . size ; let anyProcessingFailure = false ; try { this . topicRegExpArray . forEach ( async ( entry ) => { if ( entry . regExp . test ( data . topic ) ) { let retry = 0 ; let succeeded = false ; unackCounter += 1 ; let called = false ; const ackOnce = ( ) => { if ( ! called ) { called = true ; return ack ( ) ; } return false ; } ; do { logger . debug ( ` ${ data . topic } ` + ` ${ entry . id } ` + ` ${ retry + 1 } ${ this . config [ 'max.retries.processing.callbacks' ] } ` , ) ; try { await entry . callback ( data , ackOnce ) ; succeeded = true ; } catch ( error ) { logger . warn ( ` ${ entry . id } ${ data . topic } ${ error } ${ retry + 1 } ` , ) ; retry += 1 ; } } while ( ! succeeded && retry <= this . config [ 'max.retries.processing.callbacks' ] ) ; if ( ! succeeded ) { logger . error ( ` ${ entry . id } ${ retry } ` , ) ; unackCounter -= 1 ; anyProcessingFailure = true ; this . eventEmitter . emit ( 'error.processing' , entry . id , data ) ; } } } ) ; if ( ! this . topicMap [ data . topic ] ) { return ; } this . topicMap [ data . topic ] . forEach ( async ( entry ) => { let retry = 0 ; let succeeded = false ; unackCounter += 1 ; let called = false ; const ackOnce = ( ) => { if ( ! called ) { called = true ; return ack ( ) ; } return false ; } ; do { logger . debug ( ` ${ data . topic } ` + ` ${ entry . id } ` + ` ${ retry + 1 } ${ this . config [ 'max.retries.processing.callbacks' ] } ` , ) ; try { await entry . callback ( data , ackOnce ) ; succeeded = true ; } catch ( error ) { logger . warn ( ` ${ entry . id } ${ data . topic } ${ error } ${ retry + 1 } ` , ) ; retry += 1 ; } } while ( ! succeeded && retry <= this . config [ 'max.retries.processing.callbacks' ] ) ; if ( ! succeeded ) { logger . error ( ` ${ entry . id } ${ retry } ` , ) ; unackCounter -= 1 ; anyProcessingFailure = true ; this . eventEmitter . emit ( 'error.processing' , entry . id , data ) ; } } ) ; } catch ( error ) { logger . warn ( ` ${ error } ` ) ; this . eventEmitter . emit ( 'error.processing' , null , data ) ; } finally { unackCounter -= 1 ; if ( anyProcessingFailure ) { this . unackMsgBytes -= data . size ; logger . error ( 'Consumer cannot keep working properly. Finishing consumer.' , ) ; this . finish ( ) ; } else if ( unackCounter === 0 ) { this . unackMsgBytes -= data . size ; this . commitManager . notifyFinishedProcessing ( data ) ; } } }
Given a kafka message this method invokes all interested callbacks based on
 the message's topic and mark it to be committed in an asynchronous way.
Given a kafka message this method invokes all interested callbacks based on the message ' s topic and mark it to be committed in an asynchronous way .
// the current epoch // counter to the unacknowledged callbacks // // Notice that multiple callbacks might be // called to the same message, and when the counter // becomes 0 is safe to mark the message to // be committed on Kafka (all acks have been // received). Before calling the callbacks // the counter is set to 1 and is incremented // for each callback and decremented for each // corresponding ack. It is also decremented after // calling all callbacks. // Incrementing before calling the callbacks and // decrementing after calling all callbacks guarantees // that the counter doesn't become 0 before all // callbacks are called. // acknowledge callback // // It is supposed that the ack will be called if // and only if its corresponding callback succeeded. // set backpressure control //  true if any callback has failed // verifies if the topic does not matches with a regular expression // called is a guard to guarantee that ack // is called once by each callback /* eslint-disable no-await-in-loop */ // checks if there isn't a handler to the topic explicitly // just skip // iterates by the callbacks // called is guard to guarantee that ack // is called once by each callback /* eslint-disable no-await-in-loop */