KafkaRoadConsumer implements RoadConsumer { @Override public void close() { try { consumer.unsubscribe(); consumer.close(); log.info("Closed KafkaConsumer"); } catch (org.apache.kafka.common.errors.InterruptException e) { log.error("Interrupted while closing KafkaConsumer"); Thread.currentThread().interrupt(); } } 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 close() throws Exception { underTest.init(1L, rebalanceListener); underTest.close(); verify(consumer).close(); }