Kafka offset management

0、offset管理

• offset是一个long型数值，唯一对应一条message。
• 消费者会向__consumer__offset的这个特殊topic提交偏移量。具体group的消费情况，记录在这个topic的哪个partition，计算公式如下：

brokerId = Math.abs(groupId.hashCode()) % partitionCount 
// parititionCount 是__consumer_offsets的 partition 数，默认是50个

1、自动提交offset

设置enable.auto.commit为true，则会自动提交偏移量
。consumer在调用poll()方法的时候，会检查{auto.commit.interval.ms}(默认5s)的间隔，如果满足则会自动提交offset。但是可能导致消息被重复消费。例如每5s提交偏移量一次，但是在第3s的时候发送了再均衡，这个时候偏移量不会被提交上去，但是之前的3s消息就会被重复消费。

2、手动提交

（1）commitSync，同步提交

commitSync()函数提交偏移量并阻塞等待server端响应。如果没有发生不可恢复的错误，commitSync()会一直尝试直到成功

    while(true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        
        for (ConsumerRecord<String, String> record : records) {
        }
        try {
            consumer.commitSync();
        } catch (Exception e) {
        }
    }

（2）commitAsync()，异步提交

• commitAsync()函数会异步提交请求，但是commitAsync()只会尝试一次，如果失败就不会再发送。因为可能在第一次提交偏移量失败以后，消费者消费了新的一批数据并准备提交offset，如果新的offset提交成功（新的offset比旧的offset大），旧的offset再提交成功，这个时候发生再均衡，就回出现重复消息。commitAsync()支持回调函数。

（3）同步与异步组合提交

      try {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(5);
            for (ConsumerRecord<String, String> record : records) {
            }
            consumer.commitAsync(new OffsetCommitCallback() {
                private int marker = atomicInteger.incrementAndGet();
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
                                       Exception exception) {
                    // 当发生错误的时候要重试的时候进行校验。利用上面的atomicInteger的incrementAndGet方法进行递增保证，当失败的时候check本次提交是否是最新的提交。
                    if (exception != null) {
                        if (marker == atomicInteger.get()) consumer.commitAsync(this);
                    }
                }
            });
        }
    } catch (WakeupException e) {
        // ignore for shutdown
    } finally {
        consumer.commitSync(); //Block
        consumer.close();
        System.out.println("Closed consumer and we are done");
    }

(4)提交特定offset

        Map<TopicPartition,OffsetAndMetadata> currentOffsets = Maps.newHashMap();
        int count = 0;
        try {
            while(true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("topic = %s,partition = %s,offset = %s,customer = %s",
                            record.topic(),record.partition(),record.offset(),record.key());
                    currentOffsets.put(new TopicPartition(record.topic(),record.partition()),
                            new OffsetAndMetadata(record.offset() + 1));
                    if( count % 1000 == 0) {
                        // no callback
                        consumer.commitAsync(currentOffsets,null);
                    }
                    count ++;
                }
            }
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }

3、offset缓存

在broker上，有一份关于offset的缓存，记录了(topic,groupid)维度的最新offset信息。consumer client在commit offset的时候，offset会同步更新这份缓存以及写入对应的log文件中。当这台broker宕机的时候，对应的partition-replica在的broker就会从log文件中重建缓存。