Kafka Consumer Offset解析

Kafka __consumer_offsets是一个特殊的存储元数据的Topic
数据格式可以想象成一个 KV 格式的消息，key 就是一个三元组：group.id+topic+分区号，而 value 就是 offset 的值。

查看方式:使用kafka自带的读取类
./bin/kafka-console-consumer.sh --topic __consumer_offsets --partition 01 --bootstrap-server xxx:9092 --formatter "kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter" --from-beginning --max-messages 30

一般情况下， 使用 OffsetsMessageFormatter 打印的格式可以概括为：
"[%s,%s,%d]::[OffsetMetadata[%d,%s],CommitTime %d,ExpirationTime %d]".format(group, topic, partition, offset, metadata, commitTimestamp, expireTimestamp)

数据内容:
[flink-payment-alert_query_time_1576066085229,payment-result-count,4]::NULL
[flink-payment-alert_query_time_1576066085229,payment-result-count,3]::NULL
[flink-payment-alert_query_time_1576066085229,payment-result-count,9]::NULL

另外一种是
[work_default_yw.int.spring.cxyw.blackgold.kafka.orderdomain.core.sub,work_default_yw.int.spring.cxyw.blackgold.kafka.orderdomain.topic,0]::OffsetAndMetadata(offset=19, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1636939024066, expireTimestamp=None)
[work_default_yw.int.spring.cxyw.blackgold.kafka.orderdomain.core.sub,work_default_yw.int.spring.cxyw.blackgold.kafka.orderdomain.topic,0]::OffsetAndMetadata(offset=19, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1636939028621, expireTimestamp=None)
[work_default_yw.int.spring.cxyw.blackgold.kafka.orderdomain.core.sub,work_default_yw.int.spring.cxyw.blackgold.kafka.orderdomain.topic,0]::OffsetAndMetadata(offset=19, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1636939033680, expireTimestamp=None)

还有一种是
[ProcessEngineBusinessProcess,CasBusinessTopic,1]::[OffsetMetadata[99649027,NO_METADATA],CommitTime 1636930671854,ExpirationTime 1637017071854]
[ProcessEngineBusinessProcess,CasBusinessTopic,0]::[OffsetMetadata[99650360,NO_METADATA],CommitTime 1636930671854,ExpirationTime 1637017071854]
[ProcessEngineBusinessProcess,CasBusinessTopic,3]::[OffsetMetadata[99640798,NO_METADATA],CommitTime 1636930672471,ExpirationTime 1637017072471]

分别解释一下:

1. NULL表示offset过期了 也就是墓碑消息
   offset的过期机制在源码里是定时任务执行的
   private[group] def cleanupGroupMetadata(): Unit = {
   val startMs = time.milliseconds()
   val offsetsRemoved = cleanupGroupMetadata(groupMetadataCache.values, group => {
   group.removeExpiredOffsets(time.milliseconds())
   })
   info(s"Removed {time.milliseconds() - startMs} milliseconds.")
   }

在 Kafka 中有一个名为“delete-expired-group-metadata”的定时任务来负责清理过期的消费位移，这个定时任务的执行周期由参数 offsets.retention.check.interval.ms 控制，默认值为600000，即10分钟。这和普通的topic的不太一样

还有 metadata，一般情况下它的值要么为 null 要么为空字符串，OffsetsMessageFormatter 会把它展示为 NO_METADATA，否则就按实际值进行展示。

2. expireTimestamp=None并不说明消息过期了 而是服务器端改进了根据expireTimestamp过期消息后不再设置这个值

看一下源码里这些类的结构
case class OffsetAndMetadata(offsetMetadata: OffsetMetadata,
commitTimestamp: Long = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP,
expireTimestamp: Long = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP) {

case class OffsetMetadata(offset: Long, metadata: String = OffsetMetadata.NoMetadata) {
override def toString = "OffsetMetadata[%d,%s]"
.format(offset,
if (metadata != null && metadata.length > 0) metadata else "NO_METADATA")
}

@Deprecated
public static final long DEFAULT_TIMESTAMP = -1L; // for V0, V1

另外0.11.0之后对应的数据格式版本是V2,这个版本的消息相比于v0和v1的版本而言改动很大，同时还参考了Protocol Buffer而引入了变长整型（Varints）和ZigZag编码。

3. ExpirationTime 1637017072471(具体时间戳)
   这个值和commitTimestamp对比来看,结果就是服务端offsets.retention.minutes的配置值,低版本才会设置这个值

另外:
offset为什么会有墓碑消息?
因为offset本身也会过期清理.受offsets.retention.minutes 这个配置的影响
看下官网介绍
After a consumer group loses all its consumers (i.e. becomes empty) its offsets will be kept for this retention period before getting discarded. For standalone consumers (using manual assignment), offsets will be expired after the time of last commit plus this retention period.
当group里的consumer全部下线后过offsets.retention.minutes 时间后offset就会被删除
val OffsetsRetentionMinutes: Int = 7 * 24 * 60 // 默认7天
默认2.0之前是1天,2.0及以后是7天 这个官方真是..要么就改为2天,结果直接改为7天,改动不可谓不大,而且active的group不会过期

附:https://cwiki.apache.org/confluence/display/KAFKA/KIP-186%3A+Increase+offsets+retention+default+to+7+days

另外active的group无法修改consumer offset?
Usually we do not allow committed offset changes while a group is active because we do not have a mechanism to notify the group of the change.
原因是无法通知到组成员consumer offset的变更