如何通过REST接口重置Spring Kafka的Offset并实现消息重放

核心问题修复思路 #

你当前的问题是AbstractConsumerSeekAware提供的getSeekCallbackFor仅返回了seek回调对象，没有实际调用回调的seek方法传入目标offset，我们可以通过扩展消费者服务的seek方法、改造REST接口的暂停&seek&恢复逻辑实现需求。

完整代码实现 #

1. 改造Kafka消费者服务 #

@Service
public class KafKaConsumerService extends AbstractConsumerSeekAware {

    @KafkaListener(id = "foo", topics = "mytopic-3", concurrency = "3", groupId = "mytopic-1-groupid")
    public void consumeFromTopic1(@Payload @Valid Foo message, ConsumerRecordMetadata c)  {
        dbservice.processInDB(message);
    }

    /**
     * 对外暴露的指定分区offset重置方法
     * @param topic 目标topic
     * @param partition 目标分区号
     * @param targetOffset 要重置到的offset值
     */
    public void seekToOffset(String topic, int partition, long targetOffset) {
        TopicPartition tp = new TopicPartition(topic, partition);
        ConsumerSeekCallback callback = getSeekCallbackFor(tp);
        if (callback == null) {
            throw new IllegalArgumentException("对应分区" + tp + "未找到可用的seek回调，请确认分区已被当前消费者实例订阅");
        }
        // 执行seek操作
        callback.seek(topic, partition, targetOffset);
    }

    // 扩展：重置指定分区到最早偏移量
    public void seekToBeginning(String topic, int partition) {
        TopicPartition tp = new TopicPartition(topic, partition);
        ConsumerSeekCallback callback = getSeekCallbackFor(tp);
        if (callback != null) {
            callback.seekToBeginning(topic, partition);
        }
    }

    // 扩展：重置指定分区到最新偏移量
    public void seekToEnd(String topic, int partition) {
        TopicPartition tp = new TopicPartition(topic, partition);
        ConsumerSeekCallback callback = getSeekCallbackFor(tp);
        if (callback != null) {
            callback.seekToEnd(topic, partition);
        }
    }
}

2. 改造REST维护接口 #

@RestController
public class KafkaMaintenanceController {

    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @Autowired
    private KafKaConsumerService kafKaConsumerService;

    @GetMapping("/KafkaConsumerMaintainence")
    public void resetOffset(@RequestParam String topic, @RequestParam int partition, @RequestParam long targetOffset) {
        MessageListenerContainer container = kafkaListenerEndpointRegistry.getListenerContainer("foo");
        if (container == null) {
            throw new IllegalArgumentException("未找到id为foo的监听器容器");
        }
        try {
            // 1. 暂停监听器容器
            container.pause();
            // 等待容器完全暂停，避免seek操作被后续的poll覆盖
            while (!container.isContainerPaused()) {
                Thread.sleep(50);
            }
            // 2. 执行指定分区的offset重置，批量重置多分区可循环调用seek方法
            kafKaConsumerService.seekToOffset(topic, partition, targetOffset);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("重置offset过程被中断", e);
        } finally {
            // 3. 恢复容器消费，无论操作是否成功都避免容器一直阻塞
            container.resume();
        }
    }
}

注意事项 #

• 若消费者组为多实例部署，当前接口仅会重置当前实例负责分区的offset，如需全量重置可直接通过AdminClient修改消费者组偏移量后重启所有消费实例
• 可根据业务需求扩展入参，支持批量重置多分区offset、重置到最早/最新偏移量等能力
• 生产环境请为该维护接口添加权限校验，避免误操作导致重复消费

内容的提问来源于stack exchange，提问作者Pale Blue Dot