KafkaRoadConsumer implements RoadConsumer { @Override public void init(long initialRequest, RebalanceListener rebalanceListener) { long maxPollRecords = min(max(initialRequest, minMaxPollRecords), maxMaxPollRecords); properties.setProperty("max.poll.records", Long.toString(maxPollRecords)); consumer = createConsumer(); try { consumer.subscribe(singletonList(topic), new KafkaRebalanceListener(rebalanceListener)); } catch (UnknownTopicOrPartitionException e) { consumer.close(); throw new RuntimeException("Unknown topic: " + topic, e); } } 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 init() throws Exception { underTest.init(1L, rebalanceListener); ArgumentCaptor<KafkaRebalanceListener> captor = ArgumentCaptor.forClass(KafkaRebalanceListener.class); verify(consumer).subscribe(eq(singletonList(topicName)), captor.capture()); assertThat(captor.getValue().getRebalanceListener(), is(rebalanceListener)); }
@Test public void initMaxPollRecords_Mid() throws Exception { underTest.init(7L, rebalanceListener); assertThat(underTest.getProperties().get("max.poll.records"), is("7")); }
@Test public void initMaxPollRecords_Min() throws Exception { underTest.init(4L, rebalanceListener); assertThat(underTest.getProperties().get("max.poll.records"), is("5")); }
@Test public void initMaxPollRecords_Max() throws Exception { underTest.init(11L, rebalanceListener); assertThat(underTest.getProperties().get("max.poll.records"), is("10")); }