Spring Kafka从单broker接收消息转发至另一broker失败问题咨询

问题根因 #

你自定义了指向9093 broker的KafkaTemplate实例forwarderKafkaClient，但业务代码中实际注入使用的是Spring Boot自动装配的默认KafkaTemplate。默认KafkaTemplate会复用全局Kafka配置的bootstrap地址，因此始终将消息发送到9092端口。

解决方案 #

第一步：指定注入自定义的KafkaTemplate #

你定义的forwarderKafkaClient方法名即为该Bean的默认名称，在业务类注入KafkaTemplate时，添加@Qualifier注解指定该Bean名称即可：

// 原有注入模板的代码修改为如下形式
@Autowired
@Qualifier("forwarderKafkaClient")
private KafkaTemplate<String, AuditLog> template;

第二步：（可选）关闭默认Kafka自动装配 #

如果项目中不需要使用Spring Boot默认提供的Kafka相关实例，可以在启动类上排除Kafka自动配置类，避免出现多实例混淆问题：

@SpringBootApplication(exclude = {KafkaAutoConfiguration.class})
public class YourApplication {
    public static void main(String[] args) {
        SpringApplication.run(YourApplication.class, args);
    }
}

验证方法 #

注入完成后可通过如下代码确认模板配置的目标broker地址是否正确：

Map<String, Object> configs = ((DefaultKafkaProducerFactory<String, AuditLog>) template.getProducerFactory()).getConfigurationProperties();
System.out.println(configs.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));

输出结果为127.0.0.1:9093即代表配置生效。

内容的提问来源于stack exchange，提问作者halil ibrahim binol