code: @VisibleForTesting
    synchronized void consumerSubscriptionRebalance() {
      log.debug("Subscribed Topics " + this.consumer.subscription());
      if (!this.subscribedTopics.isEmpty()) {
        final Iterator<String> iter = this.subscribedTopics.iterator();
        final List<String> topics = new ArrayList<>();
        while (iter.hasNext()) {
          topics.add(iter.next());
        }
        this.subscribedTopics.clear();
        //re-subscribe topics that are needed
        this.consumer.subscribe(topics);
      }
    }
docs: Dynamically tune subscription only for the topic that dependencies need.