Spring Cloud Stream Kafka Binder如何编程重置多分区Kafka Topic offset到指定值

该需求完全可以实现，Spring Cloud Stream Kafka binder 原生支持编程式控制消费偏移量，可按以下步骤实现：

一、基础配置调整（推荐前置修改，从根源减少异常） #

为了避免后续再次出现数据库写入失败但偏移量已提交的问题，先调整消费者配置：

• 关闭自动提交偏移量：spring.cloud.stream.kafka.bindings.<你的消费绑定名称>.consumer.auto-commit-offset: false
• 开启手动确认模式：spring.cloud.stream.kafka.bindings.<你的消费绑定名称>.consumer.ack-mode: manual

配置修改后，只有在数据库写入逻辑执行成功后，你再手动调用确认方法提交偏移量，未写入成功的消息不会被标记为已消费，应用重启后会自动从最后一次提交成功的偏移量开始消费。

二、编程式重置偏移量实现 #

如果当前已经出现了偏移量已经提交到20、需要回退到10的场景，可通过以下两种方式实现：

方式1：实现ConsumerSeekAware接口（推荐） #

Spring Cloud Stream 提供的ConsumerSeekAware接口可以直接拿到偏移量控制回调，你可以在检测到数据库恢复后触发偏移量重置，示例代码如下：

@Component
public class CustomKafkaConsumer implements ConsumerSeekAware {
    // 存储待重置的目标偏移量，数据库恢复后可将该值设置为10
    private volatile Long targetResetOffset = null;
    private final ThreadLocal<ConsumerSeekCallback> callbackThreadLocal = new ThreadLocal<>();

    @StreamListener(value = "你的消费绑定名称")
    public void processMessage(String message,
                               @Header(KafkaHeaders.OFFSET) Long currentOffset,
                               @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                               @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
                               Acknowledgment ack) {
        // 检测到有偏移量重置需求时，先执行重置逻辑
        if (targetResetOffset != null) {
            callbackThreadLocal.get().seek(topic, partition, targetResetOffset);
            targetResetOffset = null;
            return;
        }
        try {
            // 你的数据库写入逻辑
            dbSaveOperation(message);
            // 写入成功后手动提交偏移量
            ack.acknowledge();
        } catch (Exception e) {
            // 数据库写入失败时，记录需要重置的偏移量
            targetResetOffset = currentOffset;
        }
    }

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        callbackThreadLocal.set(callback);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        // 分区重新分配时如果有重置需求，也可在此触发
        if (targetResetOffset != null) {
            assignments.keySet().forEach(tp -> callback.seek(tp.topic(), tp.partition(), targetResetOffset));
        }
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
    }
}

方式2：调用Kafka原生Consumer API #

你也可以直接获取当前消费端的Consumer实例，调用原生seek方法重置偏移量：

// 拿到当前消费者实例后，指定topic、分区、目标偏移量即可
consumer.seek(new TopicPartition("你的Topic名称", 分区编号), 10);

注意事项 #

• 如果你的Topic有多个分区，需要为每个分区分别指定要重置的偏移量
• 重置偏移量后，会从目标位置开始重新消费所有后续消息，注意做好幂等处理，避免重复写入数据库

小提示：你可以在数据库中记录最后一次成功消费的偏移量，每次应用启动时读取该值，自动触发偏移量重置，完全无需人工介入处理这类异常。

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