kafka重复消费的问题

Posted 2023-05-02

参考技术A Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured session.timeout.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

造成的问题：假如consumer.properties配置中max.poll.records=40 (一次最多拉取40条数据) session.timeout.ms=30000 (会话时间)

假设kafka此时一次拉取了40条数据，但在处理第31条的时候抛出了如上的异常，就会导致，本次offset不会提交，完了这40条消息都会在接下来的某刻被再次消费，这其中就包含了其实已经消费了的30条数据

原因：the poll loop is spending too much time message processing， the time between subsequent calls to poll() was longer than the configured session.timeout.ms，好吧其实是一个意思！

意思就是说poll下来数据后，处理这些数据的时间比 session.timeout.ms配置的时间要长，从而导致the group has already rebalanced

解决办法是最后一句话：You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

即要不增大 session.timeout.ms，要不减小max.poll.records ，至于具体配置为多少，得看你处理一条消息花费多长时间 x，需要满足 x乘以max.poll.records < session.timeout.ms

另一种解决思路：

解决此类重复消费的方式：将能够唯一标识消息的信息存储在其他系统，比如redis，什么能够唯一标识消息呢？就是consumergroup+topic+partition+offset，更准确的应该是consumergroup+" "+topic+" "+partition+"_"+offset组成的key,value可以是处理时间存放在redis中，每次处理kafka消息时先从redis中根据key获取value，如果value为空，则表明该消息是第一次被消费，不为空则表示时已经被消费过的消息；

参考： https://www.cnblogs.com/chinano1/p/9357725.html

kafka 重复消费问题

使用kafka版本 0.10.0.0 进行消息消费时发现每隔一天会出现重复消费，经查阅与offsets.retention.minutes配置有关

在kafka 新版本中官方文档对该配置的解释为：

After a consumer group loses all its consumers (i.e. becomes empty) its offsets will be kept for this retention period before getting discarded. For standalone consumers (using manual assignment), offsets will be expired after the time of last commit plus this retention period.

意思是消费组所有消费者都下线kafka会对消费位移记录保持 offsets.retention.minutes 配置的值的时间，此时间后会删除消费位移记录，此时消费者重新上线进行消费会从最开始的记录开始消费出现重复消费

但在0.10.0.0版本该值的官方解释为：

Log retention window in minutes for offsets topic 

经测试该版本的该值的表现与新版本存在差异

测试发现就算消费组中消费者在线，但是持续 offsets.retention.minutes没有新数据消费，kafka定期（offsets.retention.check.interval.ms）会去删除过期位移，此时会删除最新的消费位移记录，若此时发生分区balance 会导致数据重新消费