kafka消息重复消费排查

业务使用我们基础部门封装的kafka组件，在一次版本迭代中，我们引入了offset缓存，正是这个缓存，在某种条件触发下，会导致出现消息重复消费现象。

问题描述

业务方反馈，升级组件后，某个topic，出现过几次不规律的从很早之前的offset重复消费消息，现象表现如下图。

查看这些时间点范围的日记发现，都有出现重均衡的情况，所以首先就是怀疑重均衡导致的。

随后看了组件重均衡处理的相关代码，以及出现重复消费的时间点范围内的重均衡日记，锁定了猜测。

场景

此次排查的，出现重复消费现象的topic，只有一个分区，部署4个消费者节点，每个节点开启3个KafkaConsumer，假设给这些消费者编号为：

• 物理节点1：a0,a1,a2 （三个消费者的编号）
• 物理节点2：b0,b1,b2
• 物理节点3：c0,c1,c2
• 物理节点4：d0,d1,d2

猜测可能的原因

从日记表现来看，猜测可能的原因如下：

• 2月3日，a0在缓存中记录消费到了offset=2199634，之后重均衡到了b0；
• 2月10日，可能是升级原因，重均衡到了a0，然后不知道什么原因，a0没开始消费，又重均衡到了其它节点，a0在重均衡之前，a0把缓存中的offset提交(没清理缓存，即offset=2199634)，即提交了2月3日最后一次提交的offset。

只是猜测，当时各种尝试都无法重现出来，可能还受其它条件的影响，因此未被证实。

为什么要缓存offset?

目的是在重均衡时，失去某些分区消费权限的消费者，在重均衡转移后的消费者开始消费这些分区之前，先把最后消费的消息的offset提交，避免接管的消费者产生不必要的重复消费。

消费者每消费一条消息，就会将该消息的offset更新到Map缓存，key为TopicPartition，value为offset。

给消费者注册ConsumerRebalanceListener，在onPartitionsRevoked方法中，将失去消费权限的TopicPartition的offset提交。

正常情况下，不会出现“重均衡到了a0，然后不知道什么原因，a0没开始消费，又重均衡到了其它节点”这种情况，所以在测试组件时，没考虑到这种情况。

修复只需要在ConsumerRebalanceListener的onPartitionsAssigned方法中，移除缓存。

class AutoCommitOffsetRebalanceListener implements ConsumerRebalanceListener {

    private final String group;

    private final KafkaConsumer<String, byte[]> consumer;

    private final Map<TopicPartition, OffsetAndMetadata> offsetNewMap = new ConcurrentHashMap<>();

    public AutoCommitOffsetRebalanceListener(String group, KafkaConsumer<String, byte[]> consumer) {

        this.consumer = consumer;

        this.group = group;

    }

    // 每消费成功一条消息调用 
    void syncOffset(TopicPartition topicPartition, OffsetAndMetadata offsetAndMetadata) {
        offsetNewMap.put(topicPartition,offsetAndMetadata);
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Map<TopicPartition, OffsetAndMetadata> commitMap = new HashMap<>();
        for (TopicPartition topicPartition : partitions) {
            OffsetAndMetadata offsetAndMetadata = offsetNewMap.get(topicPartition);
            if (offsetAndMetadata != null) {
                commitMap.put(topicPartition, offsetAndMetadata.getMetadata());
            }
        }

        // 提交offset
        if (!CollectionUtils.isNullOrEmpty(commitMap)) {
            try {
                this.consumer.commitSync(commitMap);
                LogUtils.info("[onPartitionsRevoked] auto commitSync success. {}", commitMapToString(commitMap));
            } catch (CommitFailedException exception) {
                LogUtils.info("[onPartitionsRevoked] auto commitSync fail. {}, error msg:{}", commitMapToString(commitMap), exception.getMessage());
            }
        }
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // fix cache bug on this
        for (TopicPartition tp : partitions) {
            offsetNewMap.remove(tp);
        }
        LogUtils.info("[onPartitionsAssigned] {}", partitionsToString(partitions));
    }

}

证实猜测

虽然本地debug各种模拟都未能复现，但从日记表现的现象来看，猜测的原因可能性非常大。

本地未能复现“重均衡到了a0，然后不知道什么原因，a0没开始消费，又重均衡到了其它节点”，可能是因为还有其它条件的影响。需要业务方确认消息消费已经做了幂等之后，在将问题修复后，配合升级一个版本，再验证。

升级后，指标数据、日记，都没有再出现重复消费情况，证实了前面提出的猜测。

指标：

日记：