@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)); }
@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 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 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); }
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(); }
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();  }