KafkaMRecordSource implements MRecordSource { @Override public UntypedValues get() { getMRecordsIfNoneAvailable(); ConsumerRecord<String, UntypedValues> next = recordIterator.next(); if (logger.isTraceEnabled() && next.offset() % LoggingUtil.getTraceLogReportCheckpoint() == 0) { logger.trace("Offset {} @ {}", next.offset(), Instant.ofEpochMilli(next.value().getTimestamp())); } return next.value(); } KafkaMRecordSource(KafkaConsumer<String, UntypedValues> consumer); @SuppressWarnings("squid:S2095") // KafkaConsumer is responsible for the closing of the Deserializers  KafkaMRecordSource(String kafkaUrl, String kafkaGroup, Collection<String> topics); @SuppressWarnings("squid:S2095") // KafkaConsumer is responsible for the closing of the Deserializers  KafkaMRecordSource(String kafkaUrl, String kafkaGroup, Collection<String> topics, boolean fromBeginning);  KafkaMRecordSource(String kafkaUrl, String kafkaGroup, Collection<String> topics,
                              Map<String, Object> additionalProperties); @Override UntypedValues get(); @Override boolean hasRemaining(); @Override void init(); @Override void close(); @Override Kind getKind();  }
@Test public void get() { for (int i = 0; i < 12; i++) { UntypedValues mRecord = source.get(); assertNotNull(mRecord); } Mockito.verify(consumer, Mockito.times(2)).poll(POLL_TIMEOUT_MS); }