+ Set<ConsumerRecord<String, Long>> received = new HashSet<>();
+ testHandler = record -> received.add(record);
+
+ endlessConsumer.start();
+
+ await("Consumer failed")
+ .atMost(Duration.ofSeconds(30))
+ .until(() -> !endlessConsumer.running());
+
+ checkSeenOffsetsForProgress();
+ compareToCommitedOffsets(newOffsets);
+
+ endlessConsumer.start();