Kafka Consumer Offset解析
作者&投稿：第怨 （若有异议请与网页底部的电邮联系）
数据格式可以想象成一个 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]
分别解释一下:
在 Kafka 中有一个名为“delete-expired-group-metadata”的定时任务来负责清理过期的消费位移，这个定时任务的执行周期由参数 offsets.retention.check.interval.ms 控制，默认值为600000，即10分钟。这和普通的topic的不太一样
还有 metadata，一般情况下它的值要么为 null 要么为空字符串，OffsetsMessageFormatter 会把它展示为 NO_METADATA，否则就按实际值进行展示。
看一下源码里这些类的结构
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编码。
另外:
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的变更
拓炊尤力： 很早以前我们组里的Intern写过一个Patch用来GC旧的consumer metadata from ZK:[KAFKA-559] Garbage collect old consumer metadata entries这个最终没有merge进code base,不过你可以考虑拿过来改一改自己用.此外就是新版本0.9里面...
襄城县17086257537： 如何使用python 连接kafka 并获取数据 - ？
拓炊尤力： 连接 kafka 的库有两种类型,一种是直接连接 kafka 的,存储 offset 的事情要自己在客户端完成.还有一种是先连接 zookeeper 然后再通过 zookeeper 获取 kafka 的 brokers 信息, offset 存放在 zookeeper 上面,由 zookeeper 来协调. 我现在使...
襄城县17086257537： kafka获取数据的几种方式 - ？
拓炊尤力： 一、基于Receiver的方式这种方式使用Receiver来获取数据.Receiver是使用Kafka的高层次Consumer API来实现的.receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的,然后Spark Streaming启动的job会去处理那些数据....
襄城县17086257537： 为什么kafka consumer无法注册到zookeeper的consumers里 - ？
拓炊尤力： 因为0.9 之后的kafka使用了新的consumer API进行消费,配置bootstrap-server.旧API虽然兼容,但是会被逐渐取代.参照官方文档2.2 Consumer API As of the 0.9.0 release we have added a new Java consumer to replace our existing high-level ...
襄城县17086257537： kafka集群重启,怎样确 - ？
拓炊尤力： 1、查询当前线程是否包含kafka2、查询当前端口占用线程是否归属kafka3、开启consumer线程查看运行4、借助第三方监控工具查看集群运行情况
襄城县17086257537： kafka的消费者组该怎么删除 - ？
拓炊尤力： 没有删除消费者的需求 除非队列迁移 因为我们消费者的group是固定的 如果你有这种需求的话 调用zk api删除就行了
襄城县17086257537： kafka 0.9版本的consumer为什么不再配置zookeeper.connector - ？
拓炊尤力： 因为0.9 之后的kafka使用了新的consumer API进行消费,配置bootstrap-server.旧API虽然兼容,但是会被逐渐取代.参照官方文档 2.2 Consumer API As of the 0.9.0 release we have added a new Java consumer to replace our ...
襄城县17086257537： 如何为Kafka集群选择合适的主题和分区数量 - ？
拓炊尤力： 如何决定kafka集群中topic,partition的数量,这是许多kafka用户经常遇到的问题.本文列举阐述几个重要的决定因素,以提供一些参考.分区多吞吐量更高 一个话题topic的各个分区partiton之间是并行的.在producer和broker方面,写不同的分区...
襄城县17086257537： kafka offset是否有序 - ？
拓炊尤力： massage: kafka中最基本的传递对象,有固定格式.topic: 一类消息,如page view,click行为等.producer: 产生信息的主体,可以是服务器日志信息等.consumer: 消费producer产生话题消息的主体.broker: 消息处理结点,多个broker组成kafka集群.partition: topic的物理分组,每个partition都是一个有序队列.segment: 多个大小相等的段组成了一个partition.
