Spring Kafka @KafkaListener搭配@SendTo是否支持延迟结果返回

Spring Kafka默认配置下，@SendTo搭配@KafkaListener的请求-应答模式不会自动处理异步类型的延迟返回值，直接返回Future、CompletableFuture、Mono无法正常工作是默认逻辑导致的，并非使用方式错误。

默认返回值处理器不会主动阻塞等待Future任务执行完成，也不会主动订阅Mono这类响应式流触发执行，会直接把返回的异步对象本身序列化后作为响应内容发送，自然无法得到预期的业务结果。

实现延迟应答的可行方案 #

方案1：手动控制响应发送（兼容所有版本） #

不依赖@SendTo的自动转发逻辑，在监听器方法中拿到响应相关的请求头，等异步延迟任务执行完成后，手动通过KafkaTemplate发送响应，这种方式完全不阻塞监听器线程，适配所有Spring Kafka版本：

@KafkaListener(topics = "some")
public void getBatchInfo(String batchCode,
                         @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTopic,
                         @Header(KafkaHeaders.CORRELATION_ID) byte[] correlationId) {
    // 执行你的延迟业务逻辑，拿到异步结果
    CompletableFuture<String> asyncResult = buildBatchInfoAsync(batchCode);
    asyncResult.whenComplete((result, throwable) -> {
        if (throwable != null) {
            // 自行处理异常场景的应答逻辑
            return;
        }
        // 异步任务完成后手动发送应答
        kafkaTemplate.send(new String(replyTopic), null, correlationId, result);
    });
}

方案2：开启框架原生异步应答支持（Spring Kafka 2.9+） #

从2.9版本开始，Spring Kafka正式提供了异步返回值的自动适配能力，只需要在监听器容器工厂中开启asyncReplies配置，就可以直接用@SendTo搭配异步返回值实现延迟结果：

• 确认项目使用的Spring Kafka版本 >= 2.9
• 修改监听器容器工厂配置：

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConsumerFactory<Object, Object> consumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    // 开启异步应答支持，框架会自动处理异步返回值
    factory.getContainerProperties().setAsyncReplies(true);
    return factory;
}

配置完成后，以下写法可以正常工作，框架会自动等待异步任务完成、订阅响应式流拿到最终结果后再发送应答，不需要手动处理消息发送逻辑：

@SendTo
@KafkaListener(topics = "some")
public CompletableFuture<String> getBatchInfo(String batchCode){
    // 直接返回异步结果即可
    return buildBatchInfoAsync(batchCode);
}

// 响应式类型Mono/Flux也同样支持
@SendTo
@KafkaListener(topics = "some")
public Mono<String> getBatchInfoReactive(String batchCode){
    return Mono.fromCallable(() -> buildBatchInfo(batchCode))
            .delayElement(Duration.ofSeconds(3)); // 模拟延迟逻辑
}

注意：普通java.util.concurrent.Future因为不支持完成回调，框架无法在任务结束时触发应答发送，因此即使开启异步应答配置也不支持普通Future作为返回值，建议替换为CompletableFuture实现相同能力。

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