@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); }
@Override public ApiaryListenerEvent next() { readRecordsIfNeeded(); ConsumerRecord<Long, byte[]> next = records.next(); return eventSerDe.unmarshal(next.value()); }
KafkaMessageReader implements Iterator<ApiaryListenerEvent>, Closeable { @Override public ApiaryListenerEvent next() { readRecordsIfNeeded(); ConsumerRecord<Long, byte[]> next = records.next(); return eventSerDe.unmarshal(next.value()); } }
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); }
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(); }
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();  }