kafka提供了消费组命令工具管理消费组&＃xff1a;kafka-consumer-groups.sh&＃xff0c;在0.11版本之后引入位移重置功能&＃xff0c;重置策略如下&＃xff08;引用自官方文档&＃xff09;&＃xff1a;

--reset-offsets also has following scenarios to choose from (at least one scenario must be selected):

• --to-datetime : Reset offsets to offsets from datetime. Format: &＃39;YYYY-MM-DDTHH:mm:SS.sss&＃39;
• --to-earliest : Reset offsets to earliest offset.
• --to-latest : Reset offsets to latest offset.
• --shift-by : Reset offsets shifting current offset by &＃39;n&＃39;, where &＃39;n&＃39; can be positive or negative.
• --from-file : Reset offsets to values defined in CSV file.
• --to-current : Resets offsets to current offset.
• --by-duration : Reset offsets to offset by duration from current timestamp. Format: &＃39;PnDTnHnMnS&＃39;
• --to-offset : Reset offsets to a specific offset.

就是可以根据时间重置、重置到最小位移、最大位移...等场景。

本文主要聊一下根据时间重置消费位点时候&＃xff0c;这个时间格式的问题。

示例命令&＃xff1a;

sh bin/kafka-consumer-groups.sh \--bootstrap-server &＃39;localhost:9092&＃39; --group test_topic_consumer \--topic test_topic \--reset-offsets --to-datetime 2021-11-29T12:00:00.000 \--execute

为了看起来直观点&＃xff0c;我加了"\"换行展示。

示例命令重置消费组为test_topic_consumer订阅test_topic的消费偏移为2021年11月29号中午12点整的时候消息位点。

实际上&＃xff0c;结果不一定是这样&＃xff0c;和时区有关。

协调世界时(UTC)

关于utc&＃xff0c;查看维基百科

kafka脚本采用的是utc时间标准&＃xff0c;与北京时间换算如下&＃xff1a;

点击查看&＃xff1a;图片来源

北京为东8时区&＃xff0c;采用的是utc&＃43;08:00&＃xff0c;这一段可以看维基百科&＃xff0c;下面我复制出一部分&＃xff0c;可以了解一下&＃xff1a;

UTC&＃43;08:00是比世界协调时间快8小时的时区&＃xff0c;理论上的位置在东经112度30分至127度30分之间&＃xff0c;是东盟标准时间的候选时区之一&＃xff0c;居住在本时区的人数约有17亿人&＃xff0c;占全世界人口的24%&＃xff0c;是全世界人口最多的时区。

该时区亦为包括台湾、新加坡、马来西亚、中国、文莱、印尼中部及澳大利亚西部在内的绝大多数汉语使用者所居住的时区。所以互联网上的不少中文网站会使用该时区标记时间&＃xff0c;而不论该网站所在地的官方时区为何。

所以&＃xff0c;如果你现在使用的是北京时间&＃xff0c;如果按示例重置位点&＃xff0c;则实际上不是重置到12:00&＃xff0c;而且是重置到了20:00的消息位点。

北京时间重置写法

如果是北京时间&＃xff0c;则命令应该如下&＃xff1a;

sh bin/kafka-consumer-groups.sh \--bootstrap-server &＃39;localhost:9092&＃39; --group test_topic_consumer \--topic test_topic \--reset-offsets --to-datetime 2021-11-29T12:00:00.000&＃43;08:00 \--execute

即时间格式为&＃xff1a;2021-11-29T12:00:00.000&＃43;08:00 &＃xff0c;表示重置到11月29号的12点。当然这里还有其它写法&＃xff0c;下面是源码注释中支持的写法&＃xff1a;

(1) yyyy-MM-dd&＃39;T&＃39;HH:mm:ss.SSS, ex: 2020-11-10T16:51:38.198
(2) yyyy-MM-dd&＃39;T&＃39;HH:mm:ss.SSSZ, ex: 2020-11-10T16:51:38.198&＃43;0800
(3) yyyy-MM-dd&＃39;T&＃39;HH:mm:ss.SSSX, ex: 2020-11-10T16:51:38.198&＃43;08
(4) yyyy-MM-dd&＃39;T&＃39;HH:mm:ss.SSSXX, ex: 2020-11-10T16:51:38.198&＃43;0800
(5) yyyy-MM-dd&＃39;T&＃39;HH:mm:ss.SSSXXX, ex: 2020-11-10T16:51:38.198&＃43;08:00
重置流程

kafka根据时间重置消费位点这一块逻辑也是相当简单&＃xff1a;

1. 获取指定topic的分区&＃xff08;也可以是所有topic&＃xff09;
2. 将时间转换为对应的时间戳&＃xff0c;此时转换的时候就是上面提到的时区问题
3. 根据时间戳获取对应的消息位点
4. 修改消费位点为对应的消息位点

下面这段代码是根据时间戳查询位点的逻辑&＃xff1a;

private def getLogTimestampOffsets(groupId: String, topicPartitions: Seq[TopicPartition], timestamp: java.lang.Long): Map[TopicPartition, LogOffsetResult] &＃61; {val timestampOffsets &＃61; topicPartitions.map { topicPartition &＃61;>// 指定根据时间戳类型查询位点&＃xff0c;除此之外还有最小和最大日志位点等topicPartition -> OffsetSpec.forTimestamp(timestamp)}.toMap// 查询消息位点val offsets &＃61; adminClient.listOffsets(timestampOffsets.asJava,withTimeoutMs(new ListOffsetsOptions)).all.get//如果时间戳超过当前最新的消息时间了&＃xff0c;就是查不到了&＃xff0c;就是未知&＃xff0c;下面会把未知这种转换为最新的消费位点val (successfulOffsetsForTimes, unsuccessfulOffsetsForTimes) &＃61;offsets.asScala.partition(_._2.offset !&＃61; ListOffsetsResponse.UNKNOWN_OFFSET)val successfulLogTimestampOffsets &＃61; successfulOffsetsForTimes.map {case (topicPartition, listOffsetsResultInfo) &＃61;> topicPartition -> LogOffsetResult.LogOffset(listOffsetsResultInfo.offset)}.toMapunsuccessfulOffsetsForTimes.foreach { entry &＃61;>println(s"\nWarn: Partition " &＃43; entry._1.partition() &＃43; " from topic " &＃43; entry._1.topic() &＃43;" is empty. Falling back to latest known offset.")}// 将查询到的和未知这种转换为最新的日志位点一起返回&＃xff0c;准备重置successfulLogTimestampOffsets &＃43;&＃43; getLogEndOffsets(groupId, unsuccessfulOffsetsForTimes.keySet.toSeq)}
可视化重置

如果觉得重置命令太麻烦&＃xff0c;推荐一款可视化控制台&＃xff1a;kafka-console-ui&＃xff0c;github地址&＃xff1a;https://github.com/xxd763795151/kafka-console-ui

新手用这个学习还是比较友好的&＃xff1a;