### Question:
KafkaMessageReader implements Iterator<ApiaryListenerEvent>, Closeable { @Override public ApiaryListenerEvent next() { readRecordsIfNeeded(); ConsumerRecord<Long, byte[]> next = records.next(); return eventSerDe.unmarshal(next.value()); } private  KafkaMessageReader(String topicName, MetaStoreEventSerDe eventSerDe, Properties consumerProperties); @VisibleForTesting  KafkaMessageReader(String topicName, MetaStoreEventSerDe eventSerDe, KafkaConsumer<Long, byte[]> consumer); @Override ApiaryListenerEvent next(); @Override boolean hasNext(); @Override void remove(); @Override void close();  }### Answer:
@Test public void nextReadsRecordsFromQueue() { assertThat(reader.next()).isSameAs(event); verify(consumer).poll(any(Duration.class)); verify(serDe).unmarshal(MESSAGE_CONTENT); }
@Test public void nextReadsNoRecordsFromQueue() { when(consumer.poll(any(Duration.class))).thenReturn(ConsumerRecords.empty()).thenReturn(messages); reader.next(); verify(consumer, times(2)).poll(any(Duration.class)); verify(serDe).unmarshal(MESSAGE_CONTENT); }
@Test(expected = SerDeException.class) public void unmarhsallThrowsException() { when(serDe.unmarshal(any(byte[].class))).thenThrow(SerDeException.class); reader.next(); }