Reactor Kafka的Commit offset提交时机与消息顺序处理问题咨询

问题解答 #

关于offset提交认知的正误判断 #

你的认知是错误的。你当前使用的receiveAutoAck()方法的默认逻辑是：只要Kafka消费者成功拉取到消息，就会自动提交offset，完全不感知后续链式调用中的消息转换、写入目标Topic等操作是否成功。如果发送到目标Topic的过程报错，这条消息的offset已经被提交，就会直接丢失，不会被重新消费。
只有当你使用手动提交offset的模式，自行控制offset提交的时机时，才有可能实现「全链路操作全部执行成功后再提交offset」的逻辑。

单条处理成功再处理下一条的实现方案 #

改造要点如下：

• 把自动ack的receiveAutoAck()替换为手动ack的receive()方法，自己控制offset提交，提前确认消费者配置中enable.auto.commit = false
• 不要用doOnNext执行发送到Kafka的异步逻辑：doOnNext是副作用操作，不会等待内部异步操作完成就会向下游传递信号，会导致发送还没完成就已经开始处理下一条消息的问题。改用concatMap串行处理逻辑，保证前一条消息的全流程处理完成后，才会拉取下一条消息
• 发送到目标Topic成功后，再手动提交当前消息的offset，保证不会丢消息

改造后代码示例如下：

// 提前确认KafkaConsumerTemplate配置了手动提交模式：enable.auto.commit = false
public Flux<String> consume(String destTopic) {
    return kafkaConsumerTemplate
            .receive() // 改用手动ack模式
            .doOnNext(consumerRecord -> log.info("received key={}, value={} from topic={}, offset={}",
                    consumerRecord.key(),
                    consumerRecord.value(),
                    consumerRecord.topic(),
                    consumerRecord.offset())
            )
            // concatMap保证串行处理，前一条处理完成才处理下一条
            .concatMap(consumerRecord -> 
                    // 发送到目标Topic，sendToKafka需返回Mono类型的异步结果，等待发送成功
                    sendToKafka(consumerRecord, destTopic)
                            // 发送成功后手动提交当前offset
                            .then(consumerRecord.receiverOffset().commit())
                            // 返回消息值向下游传递
                            .thenReturn(consumerRecord.value())
            )
            .doOnError(throwable -> log.error("Error while consuming : {}", throwable.getMessage()));
}

如果需要严格保证单条串行无乱序，建议在消费者配置中设置max.poll.records = 1，避免一次拉取多条消息带来的顺序问题。

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