Spring Kafka 2.8手动将Commit/Read offset保存至Zookeeper方案咨询

核心思路 #

Spring Kafka 2.8默认将偏移量提交至Kafka内置的__consumer_offsets主题，本身不直接支持偏移量备份到Zookeeper。要实现需求，需手动捕获消费偏移量，再通过Zookeeper客户端API将偏移量写入指定ZK节点，同时通过定时任务或消费回调完成定期备份。

具体步骤 #

1. 引入Zookeeper客户端依赖 #

在Maven/Gradle中添加兼容的ZK客户端依赖，示例Maven配置：

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.7.1</version> <!-- 选择与Kafka集群匹配的版本 -->
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>

2. 封装Zookeeper偏移量操作工具类 #

实现连接ZK、读写偏移量节点的工具类：

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import java.util.concurrent.CountDownLatch;

public class ZkOffsetBackupUtil {
    private static final String ZK_OFFSET_ROOT = "/kafka/consumer/offsets/backup";
    private ZooKeeper zk;

    public ZkOffsetBackupUtil(String zkConnectString) throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        this.zk = new ZooKeeper(zkConnectString, 30000, event -> {
            if (event.getState() == ZooKeeper.States.CONNECTED) {
                latch.countDown();
            }
        });
        latch.await();
        // 初始化根节点
        if (zk.exists(ZK_OFFSET_ROOT, false) == null) {
            zk.create(ZK_OFFSET_ROOT, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    // 保存消费者组-主题-分区的偏移量
    public void saveOffset(String groupId, String topic, int partition, long offset) throws Exception {
        String nodePath = String.format("%s/%s/%s/%d", ZK_OFFSET_ROOT, groupId, topic, partition);
        createParentNodes(nodePath);
        zk.setData(nodePath, String.valueOf(offset).getBytes(), -1);
    }

    // 读取备份的偏移量
    public long getOffset(String groupId, String topic, int partition) throws Exception {
        String nodePath = String.format("%s/%s/%s/%d", ZK_OFFSET_ROOT, groupId, topic, partition);
        if (zk.exists(nodePath, false) == null) {
            return -1; // 无备份时返回-1，可按需调整
        }
        byte[] data = zk.getData(nodePath, false, null);
        return Long.parseLong(new String(data));
    }

    // 递归创建父节点
    private void createParentNodes(String nodePath) throws Exception {
        String[] pathSegments = nodePath.split("/");
        StringBuilder currentPath = new StringBuilder();
        for (int i = 1; i < pathSegments.length - 1; i++) {
            currentPath.append("/").append(pathSegments[i]);
            if (zk.exists(currentPath.toString(), false) == null) {
                zk.create(currentPath.toString(), new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        }
    }

    // 关闭ZK连接
    public void close() throws InterruptedException {
        if (zk != null) {
            zk.close();
        }
    }
}

3. 在Spring Kafka消费者中实现定期备份 #

有两种常用实现方式：

方式一：定时任务批量备份

通过@Scheduled定时捕获当前消费的偏移量并备份，同时在分区 revoke 时触发即时备份：

import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class OffsetBackupScheduler implements ConsumerSeekAware {
    private final ConsumerFactory<?, ?> consumerFactory;
    private final ZkOffsetBackupUtil zkOffsetBackupUtil;
    private final Map<TopicPartition, Long> currentOffsets = new ConcurrentHashMap<>();

    public OffsetBackupScheduler(ConsumerFactory<?, ?> consumerFactory, ZkOffsetBackupUtil zkOffsetBackupUtil) {
        this.consumerFactory = consumerFactory;
        this.zkOffsetBackupUtil = zkOffsetBackupUtil;
    }

    // 监听分区分配，更新偏移量缓存
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        currentOffsets.putAll(assignments);
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> idlePartitionOffsets, ConsumerSeekCallback callback) {
        currentOffsets.putAll(idlePartitionOffsets);
    }

    // 分区回收时即时备份
    @Override
    public void onPartitionsRevoked(Map<TopicPartition, Long> partitions) {
        currentOffsets.putAll(partitions);
        backupOffsets();
    }

    // 每5分钟执行一次备份（可调整频率）
    @Scheduled(fixedRate = 300000)
    public void scheduledBackup() {
        backupOffsets();
    }

    private void backupOffsets() {
        String groupId = consumerFactory.getConfigurationProperties().get("group.id").toString();
        for (Map.Entry<TopicPartition, Long> entry : currentOffsets.entrySet()) {
            TopicPartition tp = entry.getKey();
            long offset = entry.getValue();
            try {
                zkOffsetBackupUtil.saveOffset(groupId, tp.topic(), tp.partition(), offset);
            } catch (Exception e) {
                // 日志记录异常，避免影响主流程
                e.printStackTrace();
            }
        }
    }
}

方式二：消费后即时备份

在消费方法中，处理完消息后直接备份当前偏移量（适合需要精准备份的场景）：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageConsumer {
    private final ZkOffsetBackupUtil zkOffsetBackupUtil;
    private static final String GROUP_ID = "your-consumer-group-id";

    public KafkaMessageConsumer(ZkOffsetBackupUtil zkOffsetBackupUtil) {
        this.zkOffsetBackupUtil = zkOffsetBackupUtil;
    }

    @KafkaListener(topics = "your-topic", groupId = GROUP_ID)
    public void consume(String message,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
                        @Header(KafkaHeaders.OFFSET) Long offset) {
        // 业务逻辑处理
        processMessage(message);

        // 备份偏移量：注意+1，因为Kafka偏移量代表下一条要消费的位置
        try {
            zkOffsetBackupUtil.saveOffset(GROUP_ID, topic, partition, offset + 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void processMessage(String message) {
        // 你的业务代码
    }
}

4. 关键注意事项 #

• 版本兼容：确保ZK客户端版本与Kafka集群使用的ZK版本匹配，避免连接或数据结构异常。
• 偏移量一致性：备份的偏移量需与Kafka中提交的偏移量保持一致，通常需将当前消费偏移量+1。
• 异常处理：ZK操作可能出现超时、节点不存在等异常，需添加重试或日志记录，避免阻塞消费流程。
• 资源释放：在应用关闭时通过@PreDestroy注解关闭ZK连接，防止资源泄漏。

内容的提问来源于stack exchange，提问作者Ivan Mirchev