图解Kafka消费组偏移量_consumer_offset的数据结构

• 1. 前言

• 2. 存储结构

• 2.1 GROUP_METADATA 消费组元信息

• 2.2 OFFSET_COMMIT 消费偏移量信息

• 3. 写入时机

• 3.1 Group元信息写入时机

• 3.2 Offset写入时机

1 . 前言

在之前的文章中, 我们有讲解过消息是如何从Producer存储到Broker中的, 也分析了消息的结构, 如果不清楚的话,可以看看 [图解Kafka的RecordBatch结构]

那么作为消费者, 消费了某条消息之后又是如何记录我已经消费了这条消息呢？

如果消费者重启了, 那么又是如何获知我上次消费到了哪个消息呢？

既然要解决上述的问题, 那么我们肯定要把我们已经消费的消息给记录下来, 不可能把整个Record给记录下来吧,那也太大了, 但是我们知道消息存储在具体的分区的时候都是有自己的Offset的, 并且是有序的offset, 所以我们只需要记录下当前消费到了哪个Offset就行了。

这个Offset保存在哪里呢？

在 0.9 版本之前，这个信息是记录在zookeeper内的。但是因为频繁的对zookeeper进行读写,压力太大集群不稳 所以在0.9之后的版本，offset保存在__consumer_offsets 这个内部的Topic内了。

那么这个是怎么储存的呢?

因为本质上还是把消费的offset当做一条普通的消息发送到内部的Topic __consumer_offsets 中, 所以存储数据的时候, 也就是存到Record消息结构的 Key Value 中。

根据版本的迭代,存储的Key Value 的Schema也有一些略微不一样, 接下来我会把每个版本的Schema都讲解一下。

值得一提的是, 这个__consumer_offsets 保存的并不仅仅是offset的信息, 实际上它分为两种类型的数据

一种是 GROUP_METADATA

这种类型的数据, 存放的是消费组相关的元数据信息

一种是 OFFSET_COMMIT

这种类型的数据,存放的是就是上面说的Offset消费偏移量数据

接下来我们看看他们的数据结构

2 . 存储结构

我们下面要说的存储结构说的是Record数据结构里面的 Key Value

2.1 GROUP_METADATA 消费组元信息

GROUP_METADATA 存储的是消费组的元信息, 比如groupId 、protocol_type、protocol、leader、members 等等。

比如说有很多的消费组 ， 每个消费组有很多消费者客户端member，每个member的 client.id、clien.host等等信息。

这个消息的Key Value数据如下

2.1.1 GROUP_METADATA_KEY

Key的结构

GroupMetadataKey结构

这里的GroupMetadataKey的最开始的2个字节存储的是version数据

1. version=2 表示的是这个消息类型就是GROUP_METADATA类型
2. version=0或1 表示的是OFFSET_COMMIT类型。

除了version,后面的数据才是Schema, 这个Schema只有一个字段就是 group。

注意: 在数据结构上 version 并不是属于Schema内的。

这里的key就是group.id

2.1.2 GROUP_METADATA_VALUE

GroupMetaDataValue 的数据结构变更的有点多, 有好几次升级, 版本信息存放在version里面,并且占用2个字节

特别注意的是: version跟其他的并不属于同一个Schema里面。

1 .Verion0 -- 基础版本,也是所有版本的共有属性

这个是最原始的版本,后续的版本都是在这个基础上做加法

Kafka版本：kafak version < 0.10.1Schema：

这里是每个结构的共有属性， 后面每个版本讲解的时候只会讲差异

Member结构

SyncGroupRequest请求参数assignment属性的结构看下图虚线部分的assignment

        private List<TopicPartition> partitions;
        private ByteBuffer userData;

JoinGroupRquest请求参数

subscription属性的结构看下图红线框起来的部分 metadata

2 .Verion1

Kafka版本：0.10.1 < kafka version <2.1Schema：

相比于version = 0 的版本，在member里面新增了一个字段rebalance_timeout

2 .Verion2

Kafka版本：2.1 < kafka version <2.3Schema：

跟V1的版本区别是 新增了一个字段 current_state_timestamp

2 .Verion3

Kafka版本：2.3 < kafka version Schema：

跟V2的版本区别是 member里面新增了一个字段 group_instance_id

2.2 OFFSET_COMMIT 消费偏移量信息

这种类型的数据就是跟存储的每个消费组每个分区消费的offset的数据。

2.2.1 OFFSET_COMMIT_KEY

version=0或1 表示的是OFFSET_COMMIT类型。

version=2 表示的是上面提过的GROUP_METADATA类型

因为可能会有很多个不同的消费组消费同一个分区, 并且他们直接是相互独立的, 所以为了区别他们,在存储消费组offset的时候, key的值实际上是 由group.id+topic+partition 组成的。

并且一个分区在同一个消费组内只会被其中一个消费者消费, 这样就不会出现多个消费者重复消费的问题了。

2.2.2 OFFSET_COMMIT_VALUE

Value是存储的消费组的消费偏移量Offset，这个Offset是针对每个group.id+topic+partition 维度的。

除了Offset信息,还存了其他的一些信息, 请看下图Value结构的变更历史

我们主要看Version=3的一个版本

3 . 写入时机

3.1 Group元信息写入时机

元信息的写入源码入口在

GroupMetadataManager#storeGroup

  /**
  * 公众号：石臻臻的杂货铺
  * Vx: shiyanzu001
  * 领取20万字《Kafka运维实战宝典》
  **/
  def storeGroup(group: GroupMetadata,
                 groupAssignment: Map[String, Array[Byte]],
                 responseCallback: Errors => Unit): Unit = {
    //读取分区数据,获取消息的结构版本             
    getMagic(partitionFor(group.groupId)) match {
      case Some(magicValue) =>
        // We always use CREATE_TIME, like the producer. The conversion to LOG_APPEND_TIME (if necessary) happens automatically.
        val timestampType = TimestampType.CREATE_TIME
        val timestamp = time.milliseconds()
        // 获取Group元信息的Key结构
        val key = GroupMetadataManager.groupMetadataKey(group.groupId)
        // 获取Group元信息的Value结构
        val value = GroupMetadataManager.groupMetadataValue(group, groupAssignment, interBrokerProtocolVersion)
  //构造消息体
        val records = {
          val buffer = ByteBuffer.allocate(AbstractRecords.estimateSizeInBytes(magicValue, compressionType,
            Seq(new SimpleRecord(timestamp, key, value)).asJava))
          val builder = MemoryRecords.builder(buffer, magicValue, compressionType, timestampType, 0L)
          builder.append(timestamp, key, value)
          builder.build()
        }
  //构造TopicPartition，消息将会发往这个分区
        val groupMetadataPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))
        val groupMetadataRecords = Map(groupMetadataPartition -> records)
        val generationId = group.generationId

        //当日志写入成功之和调用这个回调函数，并将 创建的group插入缓存
        def putCacheCallback(responseStatus: Map[TopicPartition, PartitionResponse]): Unit = {

          // 省略很多代码..
          }

          responseCallback(responseError)
        }
        //将数据写入到Log中
        appendForGroup(group, groupMetadataRecords, putCacheCallback)

      case None =>
        responseCallback(Errors.NOT_COORDINATOR)
        None
    }
  }

这段代码很简单, 就是构造一下GroupMetadata的消息体并写入__consumer_offset对应的分区中。我们主要还是要来了解一下在什么情况下才会写入

我们看调用链,发现有三个地方调用了它

3.1.1 onCompleteJoin 当所有的Member都成功Join之后调用

这个的执行条件是: 当Group当前是Empty的状态,一个Member也没有,那么这个时候写入一个Group元信。

GroupCoordinator#onCompleteJoin

3.1.2 doSyncGroup 当Leader Member接收到SyncGroup请求

一般SyncGroup请求都是在JoinGroup完成了之后，所有的Member都会向组协调器发送SyncGroup。

但是Leader Member会带上计算好的新的分区分配方案给协调器

所以协调器当发现是Leader Member给自己发送SyncGroup请求之后,他会把相应的元数据给写入到_consumer_offset中。(因为Group的元信息有变更了)

GroupCoordinator#doSyncGroup

上面的一个入参assignments列表, 是SyncGroup请求带进来的,具体的结构请看下图右边虚线框;

上图虚线内的单个assignment的数据结构是

        private List<TopicPartition> partitions;
        private ByteBuffer userData;

想了解 SyncGroupRequest请看：KafkaConsumer SyncGroupRequest详解

3.1.3 updateStaticMemberAndRebalance 更新静态成员

在静态成员JoinGroup的时候, 会走单独的静态成员更新流程

GroupCoordinator#updateStaticMemberAndRebalance

判断group的下一代的selectedProtocol(所有memmber都支持并排序最靠前)是否会改变

1. 如果有变化，应该触发rebalance，让group的assignment和selectProtocol一致
2. 如果没有，则简单地存储group来持久化更新的静态成员

就是这里的2的情况,去更新一下静态成员的相关信息,主要就是更新 memberId，其他的都没有什么好变更的。分配方案也不会有改变。

GROUP_METADATA_VALUE_SCHEMA版本选择

GroupMetadataManager#groupMetadataValue

  def groupMetadataValue(groupMetadata: GroupMetadata,
                         assignment: Map[String, Array[Byte]],
                         apiVersion: ApiVersion): Array[Byte] = {

    val (version, value) = {
      if (apiVersion < KAFKA_0_10_1_IV0)
        (0.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V0))
      else if (apiVersion < KAFKA_2_1_IV0)
        (1.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V1))
      else if (apiVersion < KAFKA_2_3_IV0)
        (2.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V2))
      else
        (3.toShort, new Struct(GROUP_METADATA_VALUE_SCHEMA_V3))
    }


// 省略。。。

3.2 Offset写入时机

handleJoinGroupRequest

(略)

本文由哈喽比特于2年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/mxTjj3wTXxdOKwEIP3j-7Q