Spring Kafka注入ReplyingKafkaTemplate提示找不到Bean如何解决

你在Spring Boot项目中尝试参照KafkaTemplate的注入使用方式，在REST控制器中引入ReplyingKafkaTemplate实现Kafka请求-应答交互，编写的控制器代码如下：

@RestController
public class TestController {

    @Autowired
    private ReplyingKafkaTemplate<Object, KafkaExampleRecord, KafkaExampleRecord> replyingTemplate;

    @PostMapping("/test/request")
    public void requestReply(@RequestBody KafkaExampleRecord record) throws ExecutionException, InterruptedException, TimeoutException {
        ProducerRecord<Object, KafkaExampleRecord> producerRecord = new ProducerRecord<>("mytopic", record);
        RequestReplyFuture<Object, KafkaExampleRecord, KafkaExampleRecord> replyFuture = replyingTemplate.sendAndReceive(producerRecord);

        SendResult<Object, KafkaExampleRecord> sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);

        ConsumerRecord<Object, KafkaExampleRecord> consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);
    }
}

项目启动时抛出如下依赖注入异常：

Field replyingTemplate in com.blah.KafkaController required a bean of type 'org.springframework.kafka.requestreply.ReplyingKafkaTemplate' that could not be found.

你已经通过如下配置类开启了Kafka相关支持，所有Kafka连接、生产/消费者参数均已写入application.yml配置文件：

@Configuration
@EnableKafka
public class KafkaConfig {

}

核心疑问：

• 还需要补充什么配置才能正常注入ReplyingKafkaTemplate？
• 是否必须手动定义ReplyingKafkaTemplate对应的Bean？你认为手动定义该Bean并无必要。

Spring Kafka官方自动配置默认不会提供ReplyingKafkaTemplate类型的Bean，你必须手动完成该Bean的定义，不存在自动装配的可能。
原因很简单：ReplyingKafkaTemplate的正常运行强依赖业务侧自定义配置，包括专用的回复消息监听容器、专属消费组、回复主题、超时规则等，框架无法给出适配所有业务场景的通用默认值，因此不会自动创建该Bean。

你只需要在现有KafkaConfig配置类中补充两个Bean定义即可：

• 定义专用于接收回复消息的监听容器，该容器由ReplyingKafkaTemplate自行管理生命周期，不要给它配置普通的@KafkaListener监听逻辑：

@Bean
public ConcurrentMessageListenerContainer<Object, KafkaExampleRecord> kafkaReplyContainer(
        ConcurrentKafkaListenerContainerFactory<Object, KafkaExampleRecord> factory) {
    ConcurrentMessageListenerContainer<Object, KafkaExampleRecord> container =
            factory.createContainer("你的回复消息专用Topic名称");
    container.getContainerProperties().setGroupId("kafka-reply-group");
    container.setAutoStartup(false);
    return container;
}

• 定义ReplyingKafkaTemplate实例，关联上述回复容器：

@Bean
public ReplyingKafkaTemplate<Object, KafkaExampleRecord, KafkaExampleRecord> replyingKafkaTemplate(
        ProducerFactory<Object, KafkaExampleRecord> producerFactory,
        ConcurrentMessageListenerContainer<Object, KafkaExampleRecord> kafkaReplyContainer) {
    ReplyingKafkaTemplate<Object, KafkaExampleRecord, KafkaExampleRecord> template =
            new ReplyingKafkaTemplate<>(producerFactory, kafkaReplyContainer);
    // 可按需配置全局默认回复超时时间，替代每次调用传入的超时参数
    template.setDefaultReplyTimeout(Duration.ofSeconds(10));
    return template;
}

注意：不要复用业务侧处理普通消息的Kafka监听容器作为回复容器，混用会导致消息抢占、应答丢失、消费位点异常等问题。

配置完成后，你原有控制器中的注入和调用逻辑不需要做任何修改，即可正常启动运行。

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