Hi,
I run a verifiable consumer on HDF-3.1.1 but it never exits.
echo -e "max.poll.records=1\nenable.auto.commit=true\nauto.commit.interval.ms=1" > /tmp/consumer.config && /usr/hdf/3.1.1.0-35/kafka/bin/kafka-verifiable-consumer.sh --broker-list rjk-hdf-m:6667,rjk-hdf-s-01:6667,rjk-hdf-s-02:6667 --topic truck_speed_events_only_avro_keyed_non_transactional --reset-policy earliest --consumer.config /tmp/consumer.config --group-id test_group --verbose --max-messages 10
The output is according to expectations (at first):
{"timestamp":1529393105586,"name":"startup_complete"}
{"timestamp":1529393105824,"name":"partitions_revoked","partitions":[]}
{"timestamp":1529393108933,"name":"partitions_assigned","partitions":[{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":2},{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1},{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":0}]}
{"timestamp":1529393109005,"name":"record_data","key":"95","value":"\u0001\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0004\u0000\u0000\u0000\u0001.2018-05-28 19:43:05.689�����X\"truck_speed_event�\u00010\u001ANadeem Asghar\u0006:Saint Louis to Chicago Route2�\u0001","topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1,"offset":0}
{"timestamp":1529393109008,"name":"records_consumed","count":1,"partitions":[{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1,"count":1,"minOffset":0,"maxOffset":0}]}
{"timestamp":1529393109024,"name":"offsets_committed","offsets":[{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1,"offset":1}],"success":true}
{"timestamp":1529393109032,"name":"record_data","key":"65","value":"\u0001\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0004\u0000\u0000\u0000\u0001.2018-05-28 19:43:05.734�����X\"truck_speed_event�\u00014\u0016Don Hilborn\u00006Saint Louis to Tulsa Route2�\u0001","topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1,"offset":1}
{"timestamp":1529393109032,"name":"records_consumed","count":1,"partitions":[{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1,"count":1,"minOffset":1,"maxOffset":1}]}
{"timestamp":1529393109039,"name":"offsets_committed","offsets":[{"topic":"truck_speed_events_only_avro_keyed_non_transactional","partition":1,"offset":2}],"success":true}
ETC. ETC. ETC.
but after the parameterized 10 messages have been consumed the util just keeps running forever with the following screen output:
{"timestamp":1529393109144,"name":"records_consumed","count":1,"partitions":[]}
{"timestamp":1529393109144,"name":"offsets_committed","offsets":[],"success":true}
{"timestamp":1529393109144,"name":"records_consumed","count":1,"partitions":[]}
{"timestamp":1529393109144,"name":"offsets_committed","offsets":[],"success":true}
{"timestamp":1529393109144,"name":"records_consumed","count":1,"partitions":[]}
{"timestamp":1529393109144,"name":"offsets_committed","offsets":[],"success":true}
{"timestamp":1529393109145,"name":"records_consumed","count":1,"partitions":[]}
Why wouldn't the verifiable consumer just stop after --max-messages is reached?
On the source code of the verifiable consumer (source) there is a method that should break the while loop:
private boolean isFinished() { return hasMessageLimit() && consumedMessages >= maxMessages; }Thanks
