kafka offset的存储

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

1.旧版本offset保存

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结构为：

/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id] --> offset_counter_value

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

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

//broker
../libs/kafka_2.13-2.6.0.jar

//consumer依赖
<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.8.2.1</version>
</dependency>

构建consumer：

public class Test {
   private ConsumerConnector connector() {
        Properties props = new Properties();
        props.put("zookeeper.connect", "127.0.0.1:2181");
        props.put("group.id", "test_group1");
        props.put("zookeeper.session.timeout.ms", "10000");
        props.put("zookeeper.sync.time.ms", "1000");
        props.put("auto.commit.interval.ms", "1000");
        ConsumerConfig config = new ConsumerConfig(props);
        return Consumer.createJavaConsumerConnector(config);
    }

    private void consumer() {
        String topic = "test1";
        ConsumerConnector connector = this.connector();
        Map<String, Integer> map = new HashMap();
        map.put(topic, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>> topic2stream = connector.createMessageStreams(map);
        KafkaStream<byte[], byte[]> stream = (KafkaStream) ((List) topic2stream.get(topic)).get(0);
        ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
        while (iterator.hasNext()) {
            MessageAndMetadata messageAndMetadata = iterator.next();
            System.out.println(messageAndMetadata.partition() + ", " + new String((byte[]) messageAndMetadata.message()));
        }
    }

    public static void main(String[] args) {
        new Test().consumer();
    }
}

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

[zk: localhost:2181(CONNECTED) 0] get /consumers/test_group1/offsets/test1/0
6
[zk: localhost:2181(CONNECTED) 1] get /consumers/test_group1/offsets/test1/1
2
[zk: localhost:2181(CONNECTED) 2] get /consumers/test_group1/offsets/test1/2
2

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

2.新版本offset保存

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

# A comma separated list of directories under which to store log files
log.dirs=/usr/local/var/lib/kafka-logs

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

__consumer_offsets-0            __consumer_offsets-22           __consumer_offsets-36           __consumer_offsets-5
__consumer_offsets-1            __consumer_offsets-23           __consumer_offsets-37           __consumer_offsets-6
...

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

Utils.abs(groupId.hashCode) % numPartitions

其中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'

[test-group,test1,0]::OffsetAndMetadata(offset=6, leaderEpoch=Optional[0], metadata=, commitTimestamp=1605601180391, expireTimestamp=None)
[test-group,test1,2]::OffsetAndMetadata(offset=2, leaderEpoch=Optional[0], metadata=, commitTimestamp=1605601180391, expireTimestamp=None)
[test-group,test1,1]::OffsetAndMetadata(offset=2, leaderEpoch=Optional[0], metadata=, commitTimestamp=1605601180391, expireTimestamp=None)

该数据结构为以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