lib/kafka/Consumer.js
/**
   * finish the Consumer
   *
   * This method unsubscribe the consumer from all topics and
   * disconnect it from kafka
   *
   * If it was not connected it won't do anything.
   */
finish the Consumer
This method unsubscribe the consumer from all topics and
disconnect it from kafka

If it was not connected it won't do anything.
finish the Consumer This method unsubscribe the consumer from all topics and disconnect it from kafka If it was not connected it won ' t do anything .
finish() {
    logger.debug('Disconnecting from kafka...');
    if (this.isReady === false) {
      logger.debug('... not initialized. Aborting operation.');
      return Promise.resolve();
    }
    return new Promise((resolve, reject) => {
      logger.debug('Unsubscribing from all topics and disconnecting ...');
      try {
        const disconnectTimeout = setTimeout(() => {
          logger.error('Unable to disconnect the consumer due to timeout.');
          return reject(new Error('disconnection timeout'));
        }, this.config['disconnect.interval.ms']);
        this.consumer.unsubscribe();
        this.commitManager
          .finish()
          .then(() => {
            logger.debug('Committed processed offsets!');
          })
          .catch((err) => reject(err));
        return this.consumer.disconnect((err) => {
          clearTimeout(disconnectTimeout);
          if (err) {
            logger.error(`Error while disconnecting consumer: ${err}`);
            return reject(err);
          }
          logger.debug('Successfully disconnected producer from Kafka.');
          this.isReady = false;
          return resolve();
        });
      } catch (error) {
        return reject(error);
      }
    });
  }
finish ( ) { logger . debug ( 'Disconnecting from kafka...' ) ; if ( this . isReady === false ) { logger . debug ( '... not initialized. Aborting operation.' ) ; return Promise . resolve ( ) ; } return new Promise ( ( resolve , reject ) => { logger . debug ( 'Unsubscribing from all topics and disconnecting ...' ) ; try { const disconnectTimeout = setTimeout ( ( ) => { logger . error ( 'Unable to disconnect the consumer due to timeout.' ) ; return reject ( new Error ( 'disconnection timeout' ) ) ; } , this . config [ 'disconnect.interval.ms' ] ) ; this . consumer . unsubscribe ( ) ; this . commitManager . finish ( ) . then ( ( ) => { logger . debug ( 'Committed processed offsets!' ) ; } ) . catch ( ( err ) => reject ( err ) ) ; return this . consumer . disconnect ( ( err ) => { clearTimeout ( disconnectTimeout ) ; if ( err ) { logger . error ( ` ${ err } ` ) ; return reject ( err ) ; } logger . debug ( 'Successfully disconnected producer from Kafka.' ) ; this . isReady = false ; return resolve ( ) ; } ) ; } catch ( error ) { return reject ( error ) ; } } ) ; }
finish the Consumer
 This method unsubscribe the consumer from all topics and
 disconnect it from kafka
finish the Consumer This method unsubscribe the consumer from all topics and disconnect it from kafka