TruckPark implements ApplicationRunner { @Override public void run(ApplicationArguments args) throws Exception { Map<Integer, TopicPartition> running = new HashMap<>(); offsets.forEach((tp, offset) -> { if (offset.getEnd() > offset.getStart()) { running.put(tp.partition(), tp); } }); consumer.assign(new HashSet<>(running.values())); running.values().forEach(partition -> consumer.seek(partition, offsets.get(partition).getStart())); log.info("Offsets {}", offsets); while (running.size() > 0) { log.debug("Running {}", running.values()); for (ConsumerRecord<Void, Record> record : consumer.poll(pollTimeout)) { TopicPartition topicPartition = running.get(record.partition()); if (topicPartition != null) { long endOffset = offsets.get(topicPartition).getEnd(); if (record.offset() < endOffset) { writer.write(record); } if (record.offset() >= endOffset - 1) { log.info("Pausing partition {}. Reached offset {}.", record.partition(), record.offset()); consumer.pause(singleton(topicPartition)); running.remove(topicPartition.partition()); } } } } log.info("Closing writer"); writer.close(); context.close(); } TruckPark(
      KafkaConsumer<Void, Record> consumer,
      ConsumerRecordWriter writer,
      Map<TopicPartition, Offsets> offsets,
      @Value("${kafka.pollTimeout:100}") long pollTimeout,
      ConfigurableApplicationContext context); @Override void run(ApplicationArguments args);  }
@Test public void test() throws Exception { Schema schema = SchemaBuilder.record("r").fields().name("f").type().stringType().noDefault().endRecord(); Record value = new Record(schema); ConsumerRecord<Void, Record> record1 = new ConsumerRecord<>(topic, 0, 1, null, value); ConsumerRecords<Void, Record> records1 = new ConsumerRecords<>( ImmutableMap.of(partition, ImmutableList.of(record1))); when(consumer.poll(pollTimeout)).thenReturn(records1); underTest.run(null); InOrder inOrder = inOrder(consumer, writer, context); inOrder.verify(consumer).assign(partitions); inOrder.verify(consumer).seek(partition, 1L); inOrder.verify(consumer).poll(pollTimeout); inOrder.verify(writer).write(record1); inOrder.verify(consumer).pause(partitions); inOrder.verify(writer).close(); inOrder.verify(context).close(); }