请求/应答语义下如何等待Kafka消费者完成偏移量初始化

针对使用ReplyingKafkaTemplate配置共享应答Topic、group.id为null禁用消费组、初始偏移设为latest时出现的应答丢失问题，以下是可彻底根治的可靠实现方式，不存在概率性失效问题：

核心判定标准 #

不要用"等待首次poll返回"作为初始化完成的判断依据，该节点仅代表消费者完成了元数据拉取，并未完成分区分配和偏移量设置，就是你遇到的poll返回后才重置偏移到latest、跳过已生成应答消息的问题。必须以「分区分配完成+同步执行完偏移量seek操作」作为消费者完全初始化的唯一判定标准。

落地实现步骤 #

• 给应答消息对应的监听器容器绑定自定义的分区分配回调，基于CountDownLatch实现就绪等待：
  1. 初始化一个计数为1的CountDownLatch作为初始化完成的信号标记
  2. 重写分区分配回调逻辑，在消费者拿到应答Topic的分配分区后，同步调用原生消费者的endOffsets()方法获取所有分配分区的最新位点，主动执行seek()操作把消费位点明确设置到当前最新位置，该操作是同步阻塞的，执行完成就不会再出现后续异步重置偏移的问题
  3. seek操作完成后调用countDown()释放信号
• 业务侧调用sendAndReceive()方法前，先调用latch.await()阻塞等待初始化信号，设置合理的超时时间避免永久阻塞，初始化完成后再发起请求。

核心参考代码：

// 初始化就绪信号闩锁
private final CountDownLatch replyConsumerReadyLatch = new CountDownLatch(1);
// 配置的应答Topic名称
private static final String REPLY_TOPIC = "your-shared-reply-topic";

// 应答消息监听器
@KafkaListener(topics = REPLY_TOPIC, groupId = null)
public void handleReply(ConsumerRecord<?, ?> record) {
    // 原有应答消息处理逻辑
}

// 注册分区分配回调
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback seekCallback) {
    // 过滤出当前消费者分配到的应答Topic分区
    Set<TopicPartition> replyPartitions = assignments.keySet()
            .stream()
            .filter(tp -> REPLY_TOPIC.equals(tp.topic()))
            .collect(Collectors.toSet());
    if (!replyPartitions.isEmpty()) {
        // 主动同步seek到各分区最新偏移，彻底替代默认的异步latest偏移设置逻辑
        Map<TopicPartition, Long> latestOffsets = getCurrentConsumer().endOffsets(replyPartitions);
        latestOffsets.forEach(seekCallback::seek);
        // 偏移设置完成，标记消费者就绪
        replyConsumerReadyLatch.countDown();
    }
}

// 业务侧发请求前的等待逻辑
public RequestReplyFuture sendRequest(Object requestPayload) throws InterruptedException, TimeoutException {
    // 最长等待10秒初始化，超时直接抛异常
    if (!replyConsumerReadyLatch.await(10, TimeUnit.SECONDS)) {
        throw new TimeoutException("应答Kafka消费者初始化超时");
    }
    return replyingKafkaTemplate.sendAndReceive(MessageBuilder.withPayload(requestPayload).build());
}

进阶优化 #

如果不想在业务代码里手动加等待逻辑，可以自定义ReplyingKafkaTemplate子类，把上述等待逻辑封装到模板内部：重写sendAndReceive方法，在方法入口处先判断初始化闩锁状态，未就绪则阻塞等待，就绪后再执行原有发送逻辑，对业务代码完全无侵入。

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