code: private KafkaConsumer<String, byte[]> _getConsumer(String topic, boolean autoCommitOffsets) {
          KafkaConsumer<String, byte[]> consumer = topicConsumers.get(topic);
          if (consumer == null) {
              consumer = KafkaHelper.createKafkaConsumer(bootstrapServers, consumerGroupId,
                      consumeFromBeginning, autoCommitOffsets, consumerProperties);
              KafkaConsumer<String, byte[]> existingConsumer = topicConsumers.putIfAbsent(topic,
                      consumer);
              if (existingConsumer != null) {
                  consumer.close();
                  consumer = existingConsumer;
              } else {
                  _checkAndSubscribe(consumer, topic);
              }
          }
          return consumer;
      }
docs: Prepares a consumer to consume messages from a Kafka topic.
  
  @param topic
  @param autoCommitOffsets
  @since 1.2.0