Kafka无法从头开始阅读--Java

Question

我是kafka的新手,并尝试使用kafka构建一个生产者 - 消费者应用程序.在这里,我能够向kalka发送消息,但是当我尝试使用消费者消费它时,它返回0条记录.

我检查了我的消费者组的偏移量,我可以看到偏移量等于日志长度是相同的(在我的情况下为1M - 与记录数相同).

如果我在创建我的消费者时使用此配置属性,则从头开始阅读.

configProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

但我的要求是,如果我重启消费者,它应该从前一个终点开始,如AMQ.

这里有什么我想念的吗？我认为抵消只有在消费者民意调查后才能改变.为什么它在开始时设置为最大记录长度？

Answer 1

如链接所述,您需要考虑几种情况:

1. 启动一个新的消费者(新group.id):对于这种情况,将没有提交的偏移量,因此消费者开始根据参数设置读取auto.offset.reset

2. 重新启动消费者(重用group.id):对于这种情况,消费者将从中断的地方恢复.参数设置auto.offset.reset被忽略.

因此,对于方案(1),您可以"配置"您的起始位置.对于场景(2),您的起始位置是"固定的"(即,始终是最后提交的偏移量),并且这不能通过配置进行更改.但是,您可以在第一次打电话之前.seekToBeginning()或.seekToEnd()之前完成poll(),或者阅读整个主题或从主题末尾开始.调用.seekXX()将"覆盖"最后提交的偏移量,并允许您开始使用您喜欢的任何偏移量.请注意,还有seek()"偏移参数",因此您可以指定要开始使用的任何偏移量.