文档首页/ 分布式消息服务Kafka版/ 用户指南/ 管理消费组/ 查看和重置Kafka消费进度

更新时间：2024-10-10 GMT+08:00

查看PDF

查看和重置Kafka消费进度

消费进度表示消费者的消费位置，本章节介绍如何查看和重置消费进度，重置消费进度即修改消费者的消费位置。

重置消费进度可能会导致重复消费，请谨慎操作。

前提条件

Kafka实例不支持在线重置消费进度，请先将待重置消费进度的消费组停止消费，然后重置消费进度。

停止待重置消费进度的消费者客户端后，需要等待一段时间（即ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG配置的时间，默认为1000毫秒），服务端才会认为此消费者已下线。

查看消费进度（控制台）

登录管理控制台。
在管理控制台左上角单击，选择区域。

请选择Kafka实例所在的区域。
在管理控制台左上角单击，选择“应用中间件 > 分布式消息服务Kafka版”，进入分布式消息服务Kafka专享版页面。
单击Kafka实例的名称，进入实例详情页面。
在左侧导航栏选择“消费组管理”，进入消费组列表页面。
单击待查看消费进度的消费组名称，进入消费组详情页。
在“消费进度”页签，查看此消费组订阅的Topic列表、Topic中消息的总堆积数、Topic每个分区的消费进度（包括堆积数、偏移量、最大消息位置、消费者ID、消费者地址和客户端ID）。

图1 消费进度
（可选）如果需要查询某个Topic的消费进度，在搜索框中，输入Topic名称，按“Enter”。

查看消费进度（Kafka客户端）

未开启密文接入的Kafka实例，在Kafka客户端的“/bin”目录下，通过以下命令查询消费进度。

./kafka-consumer-groups.sh --bootstrap-server ${connection-address} --offsets --describe --all-groups

参数说明如下：connection-address表示Kafka实例的连接地址，在Kafka控制台的“基本信息 > 连接信息”中获取。

示例如下：

[root@ecs-kafka bin]# ./kafka-consumer-groups.sh --bootstrap-server 192.168.xx.xx:9092,192.168.xx.xx:9092,192.168.xx.xx:9092 --offsets --describe --all-groups

Consumer group '__consumer-group-dial-test' has no active members.

GROUP                      TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
__consumer-group-dial-test __dms_dial_test 0          350             350             0               -               -               -
__consumer-group-dial-test __dms_dial_test 1          350             350             0               -               -               -
__consumer-group-dial-test __dms_dial_test 2          350             350             0               -               -               -

Consumer group 'test' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
test            topic-01        0          5               5               0               -               -               -
test            topic-01        1          3               3               0               -               -               -
test            topic-01        2          10              10              0               -               -               -
[root@ecs-kafka bin]#

已开启密文接入的Kafka实例，通过以下步骤查询消费进度。

（可选）修改客户端配置文件。
在Kafka控制台的“基本信息 > 连接信息”中查看Kafka安全协议，两种安全协议对应的配置文件设置有所不同，请根据实际情况配置。
- SASL_PLAINTEXT：如果已经设置了用户名和密码，请跳过此步骤，执行2。否则在Kafka客户端的“/config”目录中创建“ssl-user-config.properties”文件，在文件中增加如下内容。
```
security.protocol=SASL_PLAINTEXT
#SASL认证机制为“SCRAM-SHA-512”时，配置信息如下。
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
username="**********" \
password="**********";        
sasl.mechanism=SCRAM-SHA-512
#SASL认证机制为“PLAIN”时，配置信息如下。
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="**********" \
password="**********";        
sasl.mechanism=PLAIN
```
  参数说明如下：username和password为首次开启密文接入时填入的用户名和密码，或者创建用户时设置的用户名和密码。
- SASL_SSL：如果已经设置了用户名和密码，以及SSL证书配置，请跳过此步骤，执行2。否则在Kafka客户端的“/config”目录中创建“ssl-user-config.properties”文件，在文件中增加如下内容。
```
security.protocol=SASL_SSL
ssl.truststore.location={ssl_truststore_path}
ssl.truststore.password=dms@kafka
ssl.endpoint.identification.algorithm=
#SASL认证机制为“SCRAM-SHA-512”时，配置信息如下。
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
username="**********" \
password="**********";        
sasl.mechanism=SCRAM-SHA-512
#SASL认证机制为“PLAIN”时，配置信息如下。
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="**********" \
password="**********";        
sasl.mechanism=PLAIN
```
  参数说明如下：
  - ssl.truststore.location配置为client.jks证书的存放路径。注意，Windows系统下证书路径中也必须使用“/”，不能使用Windows系统中复制路径时的“\”，否则客户端获取证书失败。
  - ssl.truststore.password为服务器证书密码，不可更改，需要保持为dms@kafka。
  - ssl.endpoint.identification.algorithm为证书域名校验开关，为空则表示关闭。这里需要保持关闭状态，必须设置为空。
  - username和password为首次开启密文接入时填入的用户名和密码，或者创建用户时设置的用户名和密码。

在Kafka客户端的“/bin”目录下，通过以下命令查询消费进度。

./kafka-consumer-groups.sh --bootstrap-server ${connection-address} --offsets --describe --all-groups --command-config ../config/ssl-user-config.properties

参数说明如下：connection-address表示Kafka实例的连接地址，在Kafka控制台的“基本信息 > 连接信息”中获取。

示例如下：

[root@ecs-kafka bin]# ./kafka-consumer-groups.sh --bootstrap-server 192.168.xx.xx:9093,192.168.xx.xx:9093,192.168.xx.xx:9093 --offsets --describe --all-groups --command-config ../config/ssl-user-config.properties

Consumer group '__consumer-group-dial-test' has no active members.

GROUP                      TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
__consumer-group-dial-test __dms_dial_test 0          347             347             0               -               -               -
__consumer-group-dial-test __dms_dial_test 1          347             347             0               -               -               -
__consumer-group-dial-test __dms_dial_test 2          347             347             0               -               -               -

Consumer group 'test' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
test            topic-01        0          5               5               0               -               -               -
test            topic-01        1          3               3               0               -               -               -
test            topic-01        2          10              10              0               -               -               -
[root@ecs-kafka bin]#

重置消费进度

登录管理控制台。
在管理控制台左上角单击，选择区域。

请选择Kafka实例所在的区域。
在管理控制台左上角单击，选择“应用中间件 > 分布式消息服务Kafka版”，进入分布式消息服务Kafka专享版页面。
单击Kafka实例的名称，进入实例详情页面。
在左侧导航栏选择“消费组管理”，进入消费组列表页面。
单击待重置消费进度的消费组名称，进入消费组详情页。
在“消费进度”页签，通过以下方法，重置消费进度。
- 重置单个Topic所有分区的消费进度：在待重置消费进度的Topic后，单击“重置消费进度”。
- 重置单个Topic中单个分区的消费进度：在待重置消费进度的Topic分区后，单击“重置消费进度”。
- 重置所有Topic所有分区的消费进度：单击“一键重置消费进度”。

在弹出的“重置消费进度”对话框中，参考表1，设置重置消费进度参数。

表1 重置消费进度参数说明
参数	说明
重置类型	选择重置类型：时间：重置消费进度到指定的时间。偏移量：重置消费进度到指定的偏移量。一键重置消费进度只支持重置消费进度到指定时间。
时间	当“重置类型”为“时间”时，需要设置此参数。选择重置消费进度的时间点，重置完成后，将从此时间点开始消费。最早：最早偏移量自定义：自定义时间点最晚：最晚偏移量
偏移量	当“重置类型”为“偏移量”时，需要设置此参数。选择重置消费进度的偏移量，此偏移量不能小于0，重置完成后，将从此偏移量开始消费。

单击“确定”，弹出确认对话框。
单击“是”，完成消费进度的重置。

父主题： 管理消费组

上一篇：查看Kafka消费者信息

下一篇：查看Kafka重平衡日志

意见反馈

文档内容是否对您有帮助？

有帮助没帮助

提供反馈

提交成功！非常感谢您的反馈，我们会继续努力做到更好！

系统繁忙，请稍后重试

在使用文档中是否遇到以下问题

内容与产品页面不一致

内容不易理解

缺失示例代码

步骤不可操作

搜不到想要的内容

缺少最佳实践

意见反馈（选填）

0/500

请至少选择一项反馈信息并填写问题反馈

字符长度不能超过500

直接提交取消