code: private void _checkAndSubscribe(KafkaConsumer<?, ?> consumer, String topic) {
          synchronized (consumer) {
              Set<String> subscription = consumer.subscription();
              if (subscription == null || !subscription.contains(topic)) {
                  consumer.subscribe(Arrays.asList(topic));
              }
          }
      }
docs: Subscribes to a topic.
  
  @param consumer
  @param topic
  @since 1.3.2