private void _fetch ( BlockingQueue < ConsumerRecord < String , byte [ ] > > buffer , String topic , long waitTime , TimeUnit waitTimeUnit ) { KafkaConsumer < String , byte [ ] > consumer = _getConsumer ( topic ) ; synchronized ( consumer ) { _checkAndSubscribe ( consumer , topic ) ; Set < String > subscription = consumer . subscription ( ) ; ConsumerRecords < String , byte [ ] > crList = subscription != null && subscription . contains ( topic ) ? consumer . poll ( waitTimeUnit . toMillis ( waitTime ) ) : null ; if ( crList != null ) { for ( ConsumerRecord < String , byte [ ] > cr : crList ) { buffer . offer ( cr ) ; } } } }
Fetches messages from Kafka and puts into buffer .