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 ; }
Prepares a worker to consume messages from a Kafka topic .