code: public KafkaMessage consume(String topic, long waitTime, TimeUnit waitTimeUnit) {
          BlockingQueue<ConsumerRecord<String, byte[]>> buffer = _getBuffer(topic);
          ConsumerRecord<String, byte[]> cr = buffer.poll();
          if (cr == null) {
              _fetch(buffer, topic, waitTime, waitTimeUnit);
              cr = buffer.poll();
          }
          return cr != null ? new KafkaMessage(cr).consumerGroupId(consumerGroupId) : null;
      }
docs: Consumes one message from a topic, wait up to specified wait-time.
  
  @param topic
  @param waitTime
  @param waitTimeUnit
  @return the consumed message or {@code null} if no message available