code: 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);
                  }
              }
          }
      }
docs: Fetches messages from Kafka and puts into buffer.
  
  @param buffer
  @param topic
  @param waitTime
  @param waitTimeUnit