BlockingQueueRecordReader implements RecordReader<P> { @Override public Record<P> readRecord() throws Exception { return queue.poll(timeout, TimeUnit.MILLISECONDS); } BlockingQueueRecordReader(final BlockingQueue<Record<P>> queue);  BlockingQueueRecordReader(final BlockingQueue<Record<P>> queue, final long timeout); @Override Record<P> readRecord(); static final long DEFAULT_TIMEOUT; }
@Test public void testReadRecord() throws Exception { assertThat(blockingQueueRecordReader.readRecord()).isEqualTo(record); assertThat(blockingQueueRecordReader.readRecord()).isNull(); assertThat(queue).isEmpty(); }