Kafka使用脚本根据时间重置消费位移,格式你写对了么?
Posted 不识君的荒漠
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka使用脚本根据时间重置消费位移,格式你写对了么?相关的知识,希望对你有一定的参考价值。
前言
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)
可视化重置
如果觉得重置命令太麻烦,推荐一款可视化控制台:kafka-console-ui,github地址:https://github.com/xxd763795151/kafka-console-ui
新手用这个学习还是比较友好的:
以上是关于Kafka使用脚本根据时间重置消费位移,格式你写对了么?的主要内容,如果未能解决你的问题,请参考以下文章
Kafka核心技术与实战——18 | Kafka中位移提交那些事儿
Kafka 消费者之消费方式工作流程消费者案例(订阅主题订阅分区)消费者组案例分区的分配以及再平衡offset 位移消费者事务数据积压(消费者如何提高吞吐量)