kafka offset的存储

Posted 2023-02-21

参考技术A

offset即消费消息的偏移值，记录了kafka每个consumergroup的下一个需要读取消费位置，保障其消息的消费可靠性。

kafka0.8.1.1以前，offset保存在zk中，存放在/consumers节点下。但是由于频繁访问zk，zk需要一个一个节点更新offset，不能批量或分组更新，导致offset更新成了瓶颈。后续两个过渡版本增加了参数“offsets.storage”，该参数可配置为“zookeeper”或“kafka”分别表示offset的保持位置在zk或是broker，默认保留在zk，0.9版本以后offset就默认保存在broker下。若配置的“kafka”，当设置了“dual.commit.enabled”参数时，offset仍然可以提交到zk。
zk中保存offset结构为：

注意：由于kafka对客户端client向下兼容，低版本的client仍然能够通过链接zk消费数据，并提交offset数据，即使broker版本高于0.9，提交的offset仍然保存在zk；此时仍然存在offset更新瓶颈问题，所以建议尽量使用高版本client，通过链接broker方式消费数据。

例如：kafka broker版本2.6.0，consumer版本0.8.2.1：

构建consumer：

启动消费者消费全部10条历史消息，查看zk下/consumer节点的消费者信息：

可以看到group “test_group1”对topic “test1”的3个partition消费情况，offset分别为6，2，2。
这里kafka只记录了每个group的消费情况，没有对某一个consumer做单独记录。早期版本/ids节点记录consumer id信息，owner节点记录各个partition所属consumer信息

如上所述，新版本中offset由broker维护，offset信息由一个特殊的topic “ __consumer_offsets”来保存，offset以消息形式发送到该topic并保存在broker中。这样consumer提交offset时，只需连接到broker，不用访问zk，避免了zk节点更新瓶颈。
broker消息保存目录在配置文件server.properties中：

该目录下默认包含50个以__consumer_offsets开头的目录，用于存放offset：

offset的存放位置决定于groupid的hash值，其获取方式：

其中numPartitions由offsets.topic.num.partitions参数决定，默认值即50。以groupid “test-group”为例，计数其存储位置为：__consumer_offsets-12，当其消费全部10条数据后，使用命令查看该目录下消息记录:kafka-console-consumer --bootstrap-server localhost:9092 --topic __consumer_offsets --partition 12 --from-beginning --formatter \'kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter\'

该数据结构为以groupid-topic-partition作为key，value为OffsetAndMetadata，其中包含了offset信息。可以看到group“test-group”在topic“test1”的三个partition下offset值分别为6，2，2。同保存在zk数据一样，offset只记录groupid的消费情况，对于具体consumer是透明的。

那么offset具体被发送给哪个broker保存呢？
由上文可知，offset的存储分区是通过groupid的hash值取得的，那么offset发送的broker就是该分区的leader broker，这也符合kafka普通消息的发生逻辑。所以，每个group的offset都将发生到一个broker，broker中存在一个offset manager 实例负责接收处理offset提交请求，并返回提交操作结果。

参考：
https://wanwenli.com/kafka/2016/11/04/Kafka-Group-Coordinator.html
https://cwiki.apache.org/confluence/display/KAFKA/Offset+Management