code: private BlockingQueue<ConsumerRecord<String, byte[]>> _getBuffer(String topic) {
          BlockingQueue<ConsumerRecord<String, byte[]>> buffer = topicBuffers.get(topic);
          if (buffer == null) {
              buffer = new LinkedBlockingQueue<ConsumerRecord<String, byte[]>>();
              BlockingQueue<ConsumerRecord<String, byte[]>> existingBuffer = topicBuffers
                      .putIfAbsent(topic, buffer);
              if (existingBuffer != null) {
                  buffer = existingBuffer;
              }
          }
          return buffer;
      }
docs: Gets a buffer to store consumed messages from a Kafka topic.
  
  @param topic
  @return
  @since 1.2.0