Ensure VerifiableConsumer halts when --max-messages is reached
Author: Tom Bentley
Reviewers: Jason Gustafson
Closes from tombentley/KAFKA-6130-verifiable-consumer-max-messages
@@ -219,7 +219,7 @@ public class VerifiableConsumer implements Closeable, OffsetCommitCallback, Cons
printJson(new StartupComplete());
consumer.subscribe(Collections.singletonList(topic), this);
- while (true) {
+ while (!isFinished()) {
ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
Map<TopicPartition, OffsetAndMetadata> offsets = onRecordsReceived(records);
