重置消费位点是指改变订阅者当前的消费位置。当消费者出现故障或者消费错误数据时,您可通过重置消费位点将消费位置回滚到之前的某个位点或者指定分区位点,重新开始消费。您也可以将消费位置移动至最新位点,暂时不处理堆积的消息。
前提条件
已停止所有Consumer客户端(云消息队列 Kafka 版不支持在线重置消费位点)。
在停止Consumer客户端后,需要经过ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG配置的时间(默认10000 ms),服务端才认为Consumer真正下线。
背景信息
云消息队列 Kafka 版支持以下重置消费位点方式:
从最新位点开始消费:不再消费Broker上堆积的消息,将消费位点重置到最新的位置。
指定时间点开始消费:将消费位点重置到过去的某个时间点(该时间点以Topic的消息存储时间为准)。只要消息仍在 Kafka 的消息保留周期内(默认 3 天),选择此方式即可重新消费该时间点之后的所有未消费消息,不会遗漏。
说明执行重置前,需先停止所有 Consumer 客户端,并等待会话超时时间(默认 10 秒)后再执行重置,以确保重置生效。详见前提条件。
按分区消费位点进行重置:如果只有少量分区产生消息堆积,可以仅重置指定分区的消费位点,避免重复消费其他分区中已正确处理的消息。
堆积的消息本身并不会因此被删除,改变的只是消费位点。
操作步骤
在概览页面的资源分布区域,选择地域。
在实例列表页面,单击目标实例名称。
在左侧导航栏,单击Group 管理。
在Group 管理页面,单击目标Group ID。
在重置Group的消费位点面板,了解其前提条件,设置重置策略。
设置重置所有 Topic。
单击是,重置所有Topic的消费位点。
单击否,在Topic文本框输入需要重置Topic的名称。
设置重置方式。
单击从最新位点开始消费,将消费位点指定到最新的位置,单击确定。
单击从指定时间点的位点开始消费,在时间点文本框,单击
,从指定时间点的位点开启消费功能,单击确定。单击按分区消费位点进行重置,在目标分区所在行, 消费位点文本框输入开始消费位点值,单击确定。
在提示对话框,确认提示信息,单击确定。
常见问题
重置消费位点能否解决部分 Topic 消费失败问题?
重置消费位点可能缓解部分 Topic 消费失败的情况,但建议先排查根本原因。
若消费失败由频繁 Rebalance 或分区分配异常导致,可按以下步骤排查:
检查客户端版本,确认客户端与服务端版本兼容。
调整客户端的
session.timeout.ms和max.poll.interval.ms参数,避免因心跳超时或消息拉取超时频繁触发 Rebalance。重置消费位点后,确认新消费组已正确订阅目标 Topic,再重新启动消费。
更换 Kafka 消费组 ID 后是否能继续消费?
更换消费组 ID 后可以继续消费,但起始消费位置取决于是否首次启动及 Topic 状态。建议先排查导致更换消费组 ID 的根本原因(如网络异常或心跳超时),避免新消费组后续也出现类似问题。
删除 Kafka 消费组是否会同时删除其订阅的 Topic?
删除 Kafka 消费组不会删除其订阅的 Topic。消费组和 Topic 是独立的资源,删除消费组仅会移除该消费组及其消费位点,Topic 中的数据和配置不受影响。
Spark 或 Lindorm-Spark 类型的 ConsumerGroup 如何提交消费位点?
根据 ConsumerGroup 类型,提交方式如下:
spark-kafka-source 类型:支持向 Kafka 提交消费位点。可通过
enable.auto.commit参数控制提交方式:设为
true时自动提交消费位点。设为
false时,需在消费逻辑完成后手动调用commit(offsets)函数提交位点。
Lindorm-Spark 类型:建议手动提交消费位点,以确保消息处理完成后再提交,防止因自动提交导致位点跳跃或监控误报:
将
enable.auto.commit设为false。在消费逻辑完成后调用
commit(offsets)函数手动提交位点。
相关文档
如果您希望通过API来重置消费位点,请参见重置消费者组的消费位点。
重置完成后,您可以通过查看消费状态来获取最新的消费位点信息。