lib/kafka/Consumer.js
/**
   * Refreshes the subscription on kafka.
   * To ensure that the consumer doesn't generate excessive load trying
   * to refresh subscriptions immediately after a failure, a truncated
   * exponential backoff with random delta is implemented.
   *
   * @access private
   */
Refreshes the subscription on kafka.
To ensure that the consumer doesn't generate excessive load trying
to refresh subscriptions immediately after a failure, a truncated
exponential backoff with random delta is implemented.

@access private
Refreshes the subscription on kafka . To ensure that the consumer doesn ' t generate excessive load trying to refresh subscriptions immediately after a failure a truncated exponential backoff with random delta is implemented . @access private
refreshSubscriptions() {
    if (!this.isWaitingForRefreshSubscriptions && this.isReady) {
      this.isWaitingForRefreshSubscriptions = true;
      const subscriptionProcedure = (retries = 0) => {
        logger.debug('Refreshing subscriptions');
        try {
          this.consumer.unsubscribe();
          const topics = Array.prototype.concat(
            Object.keys(this.topicMap),
            this.topicRegExpArray.map((entry) => entry.regExp),
          );
          logger.debug(
            `subscribing in the following topics (${
              topics.length
            }): ${JSON.stringify(topics)}`,
          );
          if (topics.length > 0) {
            this.consumer.subscribe(topics);
          }
          this.isWaitingForRefreshSubscriptions = false;
        } catch (error) {
          logger.warn(`Error while subscribing: ${error}`);
          const timeout = this.backoffWithRandomDelta(retries);
          setTimeout(subscriptionProcedure, timeout, retries + 1);
        }
      };
      subscriptionProcedure();
    }
  }
refreshSubscriptions ( ) { if ( ! this . isWaitingForRefreshSubscriptions && this . isReady ) { this . isWaitingForRefreshSubscriptions = true ; const subscriptionProcedure = ( retries = 0 ) => { logger . debug ( 'Refreshing subscriptions' ) ; try { this . consumer . unsubscribe ( ) ; const topics = Array . prototype . concat ( Object . keys ( this . topicMap ) , this . topicRegExpArray . map ( ( entry ) => entry . regExp ) , ) ; logger . debug ( ` ${ topics . length } ${ JSON . stringify ( topics ) } ` , ) ; if ( topics . length > 0 ) { this . consumer . subscribe ( topics ) ; } this . isWaitingForRefreshSubscriptions = false ; } catch ( error ) { logger . warn ( ` ${ error } ` ) ; const timeout = this . backoffWithRandomDelta ( retries ) ; setTimeout ( subscriptionProcedure , timeout , retries + 1 ) ; } } ; subscriptionProcedure ( ) ; } }
Refreshes the subscription on kafka.
Refreshes the subscription on kafka .
// just schedule if does not have a previous timer scheduled and if // the consumer is ready // According to the node-rdkafka documentation we need to call // the unsubscribe method before call the subscribe with new topics // concatenates the topics explicits with the regular expressions // schedules the next retry // run immediately!