kafka partition offset的保存时间及重置

╰半橙微兮° 2023-02-26 12:26 12阅读 0赞

![在这里插入图片描述][20200715173539650.gif_pic_center]

# kafka partition offset的保存时间及重置 #

## 1. 问题背景 ##

> 后端业务流程设计上有两个进程会以生产者和消费者角色操作kafka，每次操作会指定kafka topic下的指定partition，一段时间没用这个功能后，再次使用是发现消费进程从指定partition中取出的数据不是生产者新写入的数据；

> 即存在重复消费问题

## 2. 问题排查 ##

*  查看被消费的topic的数据情况  
    ![topic][]
 *  查看使用的消费组在当前数据的消费情况  
    ![consumer][]

此时消费进程还在开着，可以看到 `CURRENT-OFFSET` 还在增长，但 `CURRENT-OFFSET` 的值小于生产者进程写入的总量(写入了2千万+数据)，且数据总量 `LOG-END-OFFSET` 大小在2亿左右，所以判断消费者消费数据时的 `offset` 被重置到最小值了

## 3. 问题原因 ##

> 已经排查出了是消费数据时的历史 `offset` 被重置为最小值

*  首先需要了解consumer从kafka读取数据的流程：

> consumer初始化时会从broker取commit offset作为初始fetch offset来取消息，之后会继续在fetch offset上按顺序正确的往后取消息。  
> fetch offset在初始化之后就不需要用户理会，由consumer自行管理维护。 consumer的commit作为松散的支线可以在任意时间点执行，commit的意义在于尽可能及时的把消费处理的结果刷回broker去，以备consumer重启初始化或通过adminClient读取使用，所以习惯上成功消费一条就commit一次。一般来说commit offset会落后fetch offset一些，另外即使一次commit失败了也没关系，只要后序commit成功就能掩盖。

*  查看kafka官方文档关于log和offset的保存时间配置项：
    
     *  log.retention  
        ![log\_retention][log_retention]
     *  offset.retention  
        ![offset\_retention][offset_retention]  
        可以看到，配置文件中的log保存时间和offset保存时间都是默认7天
 *  offset.retention.minutes  
    `offset.retention.minutes` 表示超过配置的时间(10080minutes=7days)没有进行 `commit` 的更新，服务端就会删除 `consumer`的 `commit offset` , 当 `consumer` 从 `broker` 获取 `commit offset` 时，还没有提交过offset或是已经被删除，就返回0。
 *  consumer `auto.offset.reset`
    
     *  创建`consumer`时可在`default.topic.config`配置项中配置`auto.offset.reset`策略，支持 `smallest` 和 `latest` 两个选项(可查看官方文档说明)
     *  `consumer`初始化时发现`commit offset`已经被删除，取到了0去fetch消息，这时会超出`broker`的留存消息范围，触发`consumer`的`offset reset`。如果`auto.offset.reset=smallest` 就会从留存的7天内的最小位消息开始消费(重复消费了大量log)。如果`auto.offset.reset=lastest` 就会从最新消息开始消费(消息丢失)。

## 4. comsumer 重置 commit offset ##

> 业务设计中有时需要调整 `commit offset` 的位置

from confluent_kafka import Consumer, TopicPartition
    
    def reset_offset(bootstrap_server:str, group_id:str,
                     topic:str, partition:int, new_offset:int=None):
        """ 重置commit offset, 不指定offset时重置到HW(high watermark) @param bootstrap_server: kafka server @param group_id: @param topic: @param partition: @param new_offset: @return: """
        if not new_offset:
            consumer = Consumer({ 
                "bootstrap.server": bootstrap_server,
                "group.id": group_id,
                "enable.auto.commit": False,    # 是否自动提交offset
                "default.topic.config": { 
                    "auto.offset.reset": "smallest"     # 有smallest和latest可选
                }
            })
            consumer.subscribe([topic])
            tp = TopicPartition(topic=topic, partition=partition)
            committed = consumer.committed([tp])
            print("committed: {}".format(committed[0].offset))
            watermark_offsets = consumer.get_watermark_offsets(tp)
            print("watermark_offsets: {}, {}".format(watermark_offsets[0], watermark_offsets[1]))
            new_offset = int(watermark_offsets[1]) - 1
        print("new_offset: {}".format(new_offset))
        tp_new = TopicPartition(topic, partition, new_offset)
        consumer = Consumer({ 
            "bootstrap.server": bootstrap_server,
            "group.id": group_id,
            "enable.auto.commit": True,    # 这里打开自动提交,让consumer提交offset到HW
            "default.topic.config": { 
                "auto.offset.reset": "smallest"     # 有smallest和latest可选
            }
        })
        consumer.assign([tp_new])   # assign为从指定的offset开始消费
        consumer.poll()
        print("reset commit offset finish, bootstrap.server:{}, topic:{}, partition:{}, group.id:{}, new offset: {}"\
              .format(bootstrap_server, topic, partition, group_id, new_offset))

[20200715173539650.gif_pic_center]: /images/20230209/db2bf364dce34d75ace6618bd6171175.png
[topic]: /images/20230209/9f73da7cd00448678bf3667cc6c05754.png
[consumer]: /images/20230209/d0129ed65a7e4fa7a113cd51cb42abe8.png
[log_retention]: /images/20230209/716c385b741248d0b324a15ec83a5af2.png
[offset_retention]: /images/20230209/1018b49a37fa496e8feae5136071f321.png