Spring Kafka跨集群转发异常：消息误发至源集群问题排查

1. 确认生产者工厂配置完全隔离 #

先硬核对target生产者工厂的bootstrap-servers配置，确保它指向的是目标集群，绝对不能和source集群的配置混同。检查代码示例：

@Bean
public ProducerFactory<String, Object> targetProducerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    // 这里必须明确写target集群地址，别用可能被覆盖的变量
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "target-kafka-ip:9092");
    configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-prod-target-");
    // 其他事务、序列化配置也单独设置，别复用source的配置Map
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, Object> targetKafkaTemplate(ProducerFactory<String, Object> targetProducerFactory) {
    return new KafkaTemplate<>(targetProducerFactory);
}

同时排查是否用@ConfigurationProperties绑定时，不小心把source集群的bootstrap-servers注入到了target工厂里。

2. 绑定正确的事务管理器 #

事务性生产必须确保用的是target工厂对应的事务管理器，别和source消费者的事务逻辑混在一起：

@Bean
public KafkaTransactionManager<String, Object> targetKafkaTransactionManager(ProducerFactory<String, Object> targetProducerFactory) {
    KafkaTransactionManager<String, Object> tm = new KafkaTransactionManager<>(targetProducerFactory);
    tm.setTransactionSynchronization(SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
    return tm;
}

业务方法里明确指定事务管理器：

@Transactional(transactionManager = "targetKafkaTransactionManager")
public void processAndSend(ConsumerRecord<String, Object> record) {
    // 业务处理逻辑
    targetKafkaTemplate.send("target-topic", processedData);
}

3. 排查元数据共享问题 #

Kafka客户端的Metadata实例是用来缓存集群信息的，如果多个生产者工厂不小心共用了同一个Metadata，就会出现元数据被覆盖的情况：

• 绝对不要在source和target生产者工厂之间复用同一个Map<String, Object>配置对象，每次都要新建或者深度拷贝
• 检查Bean的作用域，别因为单例配置错误，让target工厂拿到了source的配置

4. 清理自动配置残留 #

虽然你说禁用了KafkaAutoConfiguration，还是要排查：

• 有没有spring.kafka.bootstrap-servers这类全局配置藏在application.yml/properties里，某些组件可能会绕过自动配置读取这些值
• 检查环境变量里有没有KAFKA_BOOTSTRAP_SERVERS，Kafka客户端会优先读取系统级变量

5. 追踪元数据更新触发点 #

你提到org.apache.kafka.clients.producer.KafkaProducer#waitOnMetadata导致元数据被改，直接在这个方法上打断点：

• 看是哪个生产者实例触发的更新，打印它的bootstrap-servers配置
• 检查是不是发送时指定的topic在source集群存在，导致客户端错误地去source集群拉取元数据（本质还是生产者的bootstrap地址配错了）

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