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