Kafka无法从头开始阅读–Java

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

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

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

configProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,“earliest”);

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

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

解决方法:

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

>启动一个新的使用者(new group.id)：对于这种情况,将没有提交的偏移量,因此消费者开始根据参数设置auto.offset.reset读取
>重新启动使用者(重复使用group.id)：对于这种情况,消费者将从中断的地方恢复.参数设置auto.offset.reset被忽略.

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

标签：kafka-producer-api,kafka-consumer-api,java,apache-kafka
来源： https://codeday.me/bug/20190828/1753007.html