KafkaRoadConsumer implements RoadConsumer { @Override public Iterable<Record> poll() { return FluentIterable.from(consumer.poll(pollTimeoutMillis)).transform(r -> { Payload<byte[]> p = r.value(); Schema schema = schemaProvider.schema(roadName, p.getSchemaVersion()); JsonNode message = payloadDecoder.decode(schema, p.getMessage()); Payload<JsonNode> payload = new Payload<>(p.getFormatVersion(), p.getSchemaVersion(), message); return new Record(r.partition(), r.offset(), r.timestamp(), payload); }).toList(); } KafkaRoadConsumer(
      Properties properties,
      String topic,
      String roadName,
      SchemaProvider schemaProvider,
      long pollTimeoutMillis,
      int minMaxPollRecords,
      int maxMaxPollRecords); @Override void init(long initialRequest, RebalanceListener rebalanceListener); @Override Iterable<Record> poll(); @Override boolean commit(Map<Integer, Long> offsets); @Override void close();  }
@Test public void poll() throws Exception { Payload<byte[]> payload = new Payload<>((byte) 0, 1, "{}".getBytes(UTF_8)); ConsumerRecord<Void, Payload<byte[]>> consumerRecord = new ConsumerRecord<>(topicName, 0, 1L, 2L, TimestampType.CREATE_TIME, ConsumerRecord.NULL_CHECKSUM, ConsumerRecord.NULL_SIZE, ConsumerRecord.NULL_SIZE, null, payload); Map<TopicPartition, List<ConsumerRecord<Void, Payload<byte[]>>>> recordsMaps = singletonMap(topicPartition, singletonList(consumerRecord)); ConsumerRecords<Void, Payload<byte[]>> records = new ConsumerRecords<>(recordsMaps); when(consumer.poll(100)).thenReturn(records); when(payloadDecoder.decode(any(), any())).thenReturn(mapper.createObjectNode()); Record record = new Record(0, 1L, 2L, new Payload<JsonNode>((byte) 0, 1, mapper.createObjectNode())); underTest.init(1L, rebalanceListener); Iterable<Record> result = underTest.poll(); assertThat(Iterables.size(result), is(1)); assertThat(Iterables.get(result, 0), is(record)); }