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 a synchronous 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 the n-retries, an 'error.processing' event is emitted.
   *
   * Depending on the configuration, the message that couldn't be processed is
   * marked to be committed on Kafka, and the consumer continues working as nothing
   * has happened (message discarded). But, if the message can't be discarded, it
   * won't be marked to be committed on Kafka and  the consumer will be finished.
   *
   * It worths to say that in the synchronous mode the messages are marked to be
   * committed after the callbacks return.
   *
   * @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 a synchronous 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 the n-retries, an 'error.processing' event is emitted.

Depending on the configuration, the message that couldn't be processed is
marked to be committed on Kafka, and the consumer continues working as nothing
has happened (message discarded). But, if the message can't be discarded, it
won't be marked to be committed on Kafka and  the consumer will be finished.

It worths to say that in the synchronous mode the messages are marked to be
committed after the callbacks return.


Given a kafka message this method invokes all interested callbacks based on the message ' s topic and mark it to be committed in a synchronous 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 the n - retries an ' error . processing ' event is emitted . Depending on the configuration the message that couldn ' t be processed is marked to be committed on Kafka and the consumer continues working as nothing has happened ( message discarded ) . But if the message can ' t be discarded it won ' t be marked to be committed on Kafka and the consumer will be finished . It worths to say that in the synchronous mode the messages are marked to be committed after the callbacks return .
async invokeInterestedCallbacksSyncCommit(data) {
    let anyProcessingFailure = false;
    try {
      this.topicRegExpArray.forEach(async (entry) => {
        if (entry.regExp.test(data.topic)) {
          let retry = 0;
          let succeeded = 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);
              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})`,
            );
            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;
        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);
            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})`,
          );
          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 {
      if (this.config['commit.on.failure'] || !anyProcessingFailure) {
        this.commitManager.notifyFinishedProcessing(data);
      } else {
        logger.error(
          'Consumer cannot keep working properly. Finishing consumer.',
        );
        this.finish();
      }
    }
  }
async invokeInterestedCallbacksSyncCommit ( data ) { let anyProcessingFailure = false ; try { this . topicRegExpArray . forEach ( async ( entry ) => { if ( entry . regExp . test ( data . topic ) ) { let retry = 0 ; let succeeded = false ; do { logger . debug ( ` ${ data . topic } ` + ` ${ entry . id } ` + ` ${ retry + 1 } ${ this . config [ 'max.retries.processing.callbacks' ] } ` , ) ; try { await entry . callback ( data ) ; 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 } ` , ) ; 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 ; do { logger . debug ( ` ${ data . topic } ` + ` ${ entry . id } ` + ` ${ retry + 1 } ${ this . config [ 'max.retries.processing.callbacks' ] } ` , ) ; try { await entry . callback ( data ) ; 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 } ` , ) ; anyProcessingFailure = true ; this . eventEmitter . emit ( 'error.processing' , entry . id , data ) ; } } ) ; } catch ( error ) { logger . warn ( ` ${ error } ` ) ; this . eventEmitter . emit ( 'error.processing' , null , data ) ; } finally { if ( this . config [ 'commit.on.failure' ] || ! anyProcessingFailure ) { this . commitManager . notifyFinishedProcessing ( data ) ; } else { logger . error ( 'Consumer cannot keep working properly. Finishing consumer.' , ) ; this . finish ( ) ; } } }
Given a kafka message this method invokes all interested callbacks based on
 the message's topic and mark it to be committed in a synchronous way.
Given a kafka message this method invokes all interested callbacks based on the message ' s topic and mark it to be committed in a synchronous way .
// verifies if the topic does not matches with a regular expression /* eslint-disable no-await-in-loop */ // checks if there isn't a handler to the topic explicitly // just skip // iterates by the callbacks /* eslint-disable no-await-in-loop */