code: private KafkaMsgConsumerWorker _getWorker(String topic, boolean autoCommitOffsets) {
          KafkaMsgConsumerWorker worker = topicWorkers.get(topic);
          if (worker == null) {
              Collection<IKafkaMessageListener> msgListeners = topicMsgListeners.get(topic);
              worker = new KafkaMsgConsumerWorker(this, topic, msgListeners, executorService);
              KafkaMsgConsumerWorker existingWorker = topicWorkers.putIfAbsent(topic, worker);
              if (existingWorker != null) {
                  worker = existingWorker;
              } else {
                  worker.start();
              }
          }
          return worker;
      }
docs: Prepares a worker to consume messages from a Kafka topic.
  
  @param topic
  @param autoCommitOffsets
  @return