Kafka重置消费的OffsetKafka源码分析-汇总

  • Kafka消费后都会提交保存当前的消费位置offset, 可以选择保存在zk, 本地文件或其他存储系统;
  • Kafka 0.8以后提供了Coordinator的角色,.Coordinator除了可以来协调消费的group作balance外, 还接受 OffsetCommit Request, 用来存储消费的offset到Kafka本身中.具体可参考Kafka的消息是如何被消费的?;

Kafka 0.8以前的版本
  • 绝大部分的offset应该都是写到zookeeper上, 类似/consumers/[consumer group]/offsets/[topic]/[partition]
  • 如果不想重启消费进程就能reset, 可以在zk上创建一个新节点,专门用来记录需要reset的offset位軒,然后代码里watch这个节点, 获取到需要重置到的offset值,然后在发送Fetch Request时使用这个新的offset值即可;
Kafka 0.10以后的版本
Kafka 0.9.0.1版本
  • 这个版本你当然还是可以将offset保存在zk中, 然后使用上面提到的方法重置;
  • 我们现在重点来讨论下将offset保存到kafka系统本身的,其实就是存到一个内部的叫__consumer_offsets中,具体可参考Kafka的消息是如何被消费的?;
  • Kafka提供自动reset的配置
    1. auto.offset.reset 1.1 smallest : 自动重置到最小的offset, 这个最小的offset不一定是0, 因为msg可能会被过期删除掉; 1.2 largest : 自动重置到最大的offset;
    2. 这个配置只有在当前无法获取到有效的offset时才生效; 2.1 全新的group; 2.2 已存在的group, 但很久没有提交过offset, 其保存在__consumer_offsets里的信息将被compact并最终清除掉;
  • 需要手动reset时, 并没有像Kafka 0.10(11)版本那样提供给我们工具, 那怎么办? 只能自已搞, 下面提供一个思路:
    1. 确定需要重置到的offset: 1.1 如果想重置到最新或最旧的offset, 可能通过kafka的命令行工具获取: kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list [broker list] --topic [topic name] --time [-1:获取最新offset, -2:获取最旧offset] 1.2 也可以通过代码来获取, 可以使用librdkafkard_kafka_query_watermark_offsets函数来获取;
    2. 重置offset, 以使用librdkafka提供的接口为例: 2.0 需要先停掉需重置的group的所有消费进程,可以使用rd_kafka_list_groups来获取当前消费 gropu的详情; 2.1 使用rd_kafka_topic_partition_list_set_offset来设置需要重置的partiton的offset; 2.2 调用rd_kafka_subscriberd_kafka_consumer_poll来等待group完成balance; 2.3 调用rd_kafka_commit来完成重置的offset的提交;
    3. 当然librdkafka和kafka api都提供了seek接口,也可以用来设置offset;
  • 如果不是想重置到最新或最旧的offset, 而是想重置到某一时间点的offset, 该怎么办?
    1. 这个版本不支持timestamp, 如果不想对kafka源码作改动的话, 可以定时获到group的消费offset, 然后写入到外部存储系统, 比如redis;
    2. 需要重置时,从外部存储系统根据时间点来获到到当时的offset, 由于是定时采样,不一定能完全匹配上指定的时间点,但可以取与其最接近的时间点.

Kafka源码分析-汇总

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏JAVA同学会

Kafka 简介

在Kafka中,客户端和服务器之间的通信是通过一种简单的,高性能的,语言不可知的TCP协议完成的。

42620
来自专栏叁金大数据

不讲CRUSH的Ceph教程是不完整的

前面我们提到了Ceph是一个支持统一存储架构的分布式存储服务。简单介绍了Ceph的基本概念和基础架构包含的组件,其中最重要的就是底层的RADOS和它的两类守护进...

33920
来自专栏叁金大数据

漫谈未来的HDFS

前面我们提到的HDFS,了解了HDFS的特性和架构。HDFS能够存储TB甚至PB规模的数据是有前提的,首先数据要以大文件为主,其次NameNode的内存要足够大...

18330
来自专栏Code_iOS

Objective-C 内存管理(上)学习笔记

这里的“计数”表明必然会有一个东西(变量)来记录引用的变化,而在OC里这个变量就是retainCount;那么还有一个问题就是通过什么方式来操作这个变量,OC里...

8320
来自专栏移动开发的那些事儿

内存泄露的一些坑

如上,在Activity内部如果声明一个这样的Handler,那么myHandler就默认持有Activity引用,假设Activity退出了,但是可能这时候才...

39020
来自专栏叁金大数据

Ceph在手,天下我有

有人问我,你是如何做到统一存储的?我微微一笑,大声告诉他:Ceph在手,天下我有。

18420
来自专栏安富莱嵌入式技术分享

【安富莱二代示波器教程】第5章 示波器设计—波形快速刷新方案

波形快速刷新有很多方案需要测试,由于我们的GUI是采用的emWin,所以下面的这些测试都是基于emWin实现的。

21320
来自专栏叁金大数据

存储是怎样炼成的?

什么FAT,NTFS,NFS,DAS,SAN,NAS,OSD这些名词我一个都不认识。

20030
来自专栏日常分享

数组、List和ArrayList的区别

 有些知识点可能平时一直在使用,不过实际开发中我们可能只是知其然不知其所以然,所以经常的总结会对我们的提高和进步有很大的帮助,这里记录自己在工作之余的问题,持续...

29520
来自专栏最新技术

如何创建私有Python包存储库

Python包的基本脚手架是一个包含与用户交互的代码的__init__.py文件。

83030

扫码关注云+社区

领取腾讯云代金券

年度创作总结 领取年终奖励