code: private KafkaMsgConsumer getKafkaConsumer(String consumerGroupId,
              boolean consumeFromBeginning) {
          KafkaMsgConsumer kafkaConsumer = cacheConsumers.get(consumerGroupId);
          if (kafkaConsumer == null) {
              kafkaConsumer = _newKafkaConsumer(consumerGroupId, consumeFromBeginning);
              KafkaMsgConsumer temp = cacheConsumers.putIfAbsent(consumerGroupId, kafkaConsumer);
              if (temp != null) {
                  kafkaConsumer.destroy();
                  kafkaConsumer = temp;
              }
          }
          return kafkaConsumer;
      }
docs: Obtains a {@link KafkaMsgConsumer} instance.
  
  <p>
  Note: The existing {@link KafkaMsgConsumer} will be returned if such
  exists; otherwise a new {@link KafkaMsgConsumer} instance will be
  created.
  </p>
  
  @param consumerGroupId
  @param consumeFromBeginning
  @return