前言

kafka提供了消费组命令工具管理消费组：kafka-consumer-groups.sh，在0.11版本之后引入位移重置功能，重置策略如下（引用自官方文档）：

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

--to-datetime <String: datetime> : Reset offsets to offsets from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'
--to-earliest : Reset offsets to earliest offset.
--to-latest : Reset offsets to latest offset.
--shift-by <Long: number-of-offsets> : Reset offsets shifting current offset by 'n', where 'n' can be positive or negative.
--from-file : Reset offsets to values defined in CSV file.
--to-current : Resets offsets to current offset.
--by-duration <String: duration> : Reset offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'
--to-offset : Reset offsets to a specific offset.

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

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

根据时间重置消费位移

示例命令：

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

为了看起来直观点，我加了"\"换行展示。

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

实际上，结果不一定是这样，和时区有关。

协调世界时(UTC)

关于utc，查看维基百科

kafka脚本采用的是utc时间标准，与北京时间换算如下：

点击查看：图片来源

北京为东8时区，采用的是utc+08:00，这一段可以看维基百科，下面我复制出一部分，可以了解一下：

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

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

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

北京时间重置写法

如果是北京时间，则命令应该如下：

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

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

(1) yyyy-MM-dd'T'HH:mm:ss.SSS, ex: 2020-11-10T16:51:38.198
(2) yyyy-MM-dd'T'HH:mm:ss.SSSZ, ex: 2020-11-10T16:51:38.198+0800
(3) yyyy-MM-dd'T'HH:mm:ss.SSSX, ex: 2020-11-10T16:51:38.198+08
(4) yyyy-MM-dd'T'HH:mm:ss.SSSXX, ex: 2020-11-10T16:51:38.198+0800
(5) yyyy-MM-dd'T'HH:mm:ss.SSSXXX, ex: 2020-11-10T16:51:38.198+08:00

重置流程

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

获取指定topic的分区（也可以是所有topic）
将时间转换为对应的时间戳，此时转换的时候就是上面提到的时区问题
根据时间戳获取对应的消息位点
修改消费位点为对应的消息位点

下面这段代码是根据时间戳查询位点的逻辑：

    private def getLogTimestampOffsets(groupId: String, topicPartitions: Seq[TopicPartition], timestamp: java.lang.Long): Map[TopicPartition, LogOffsetResult] = {
      val timestampOffsets = topicPartitions.map { topicPartition =>
          // 指定根据时间戳类型查询位点，除此之外还有最小和最大日志位点等
        topicPartition -> OffsetSpec.forTimestamp(timestamp)
      }.toMap
      // 查询消息位点
      val offsets = adminClient.listOffsets(
        timestampOffsets.asJava,
        withTimeoutMs(new ListOffsetsOptions)
      ).all.get
      //如果时间戳超过当前最新的消息时间了，就是查不到了，就是未知，下面会把未知这种转换为最新的消费位点
      val (successfulOffsetsForTimes, unsuccessfulOffsetsForTimes) =
        offsets.asScala.partition(_._2.offset != ListOffsetsResponse.UNKNOWN_OFFSET)

      val successfulLogTimestampOffsets = successfulOffsetsForTimes.map {
        case (topicPartition, listOffsetsResultInfo) => topicPartition -> LogOffsetResult.LogOffset(listOffsetsResultInfo.offset)
      }.toMap

      unsuccessfulOffsetsForTimes.foreach { entry =>
        println(s"\nWarn: Partition " + entry._1.partition() + " from topic " + entry._1.topic() +
          " is empty. Falling back to latest known offset.")
      }

      // 将查询到的和未知这种转换为最新的日志位点一起返回，准备重置
      successfulLogTimestampOffsets ++ getLogEndOffsets(groupId, unsuccessfulOffsetsForTimes.keySet.toSeq)
    }