Spring Kafka 与嵌入式 Kafka 集成测试

我有一个 Spring Boot 应用程序，它有一个消费者从一个集群中的主题消费并生成到不同集群中的另一个主题。

现在我正在尝试使用 Spring 嵌入式 Kafka 编写集成测试用例，但遇到了问题KafkaTemplate could not be registered. A bean with that name has already been defined in class path resource

消费级

@Service
public class KafkaConsumerService {

@Autowired
private KafkaProducerService kafkaProducerService;

@KafkaListener(topics = "${kafka.producer.topic}")
public void professor(List<Professor> pro) {
    pro.forEach(kafkaProducerService::produce);
    
   }

}

制作人班

@Service
public class KafkaProducerService {

@Value("${kafka.producer.topic}")
private String topic;

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

public void produce(Professor pro) {
    kafkaTemplate.send(topic,"professor",pro);
  }

 }

在我的测试用例中我想覆盖KafkaTemplate这样当我打电话时kafkaConsumerService.professor中的方法Test它应该将数据生成到嵌入式 Kafka 中，我应该验证它。

测试配置

@TestConfiguration
@EmbeddedKafka(partitions = 1, controlledShutdown = false,
brokerProperties = {"listeners=PLAINTEXT://localhost:3333", "port=3333"})
public class KafkaProducerConfigTest {

@Autowired
 KafkaEmbedded kafkaEmbeded;

@Autowired
KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

@Before
public void setUp() throws Exception {
  for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
    ContainerTestUtils.waitForAssignment(messageListenerContainer, 
    kafkaEmbeded.getPartitionsPerTopic());
  }
}

@Bean
public ProducerFactory<String, Object> producerFactory() {
    return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(kafkaEmbeded));
}

@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
    KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<>(producerFactory());
    return kafkaTemplate;
   }

 }

测试班

@EnableKafka
@SpringBootTest(classes = {KafkaProducerConfigTest.class})
@RunWith(SpringRunner.class)
public class KafkaProducerServiceTest {

@Autowired
private KafkaConsumerService kafkaConsumerService;

@Test
public void testReceive() throws Exception {
     kafkaConsumerService.professor(Arrays.asList(new Professor()));
     
     //How to check messages is sent to kafka?
}

 }

Error

 The bean 'kafkaTemplate', defined in com.kafka.configuration.KafkaProducerConfigTest, could not be registered. 
 A bean with that name has already been defined in class path resource [com/kafka/configuration/KafkaProducerConfig.class] and overriding is disabled.
 Consider renaming one of the beans or enabling overriding by setting spring.main.allow-bean-definition-overriding=true

有人可以帮助我如何验证发送到嵌入式 Kafka 服务器的消息吗？

Note我收到一些已弃用的警告

KafkaEmbedded 类型已弃用

KafkaEmbedded 类型的 getPartitionsPerTopic() 方法已弃用

KafkaTestUtils 类型中的方法 ProducerProps(KafkaEmbedded) 已弃用

启动2.1默认情况下禁用 bean 覆盖 https://github.com/spring-projects/spring-boot/wiki/Spring-Boot-2.1-Release-Notes#bean-overriding.

默认情况下禁用 Bean 覆盖，以防止 Bean 被意外覆盖。如果您依赖覆盖，则需要设置spring.main.allow-bean-definition-overriding to true.

关于弃用；请参阅 javadoc@EmbeddedKafka。它被替换为EmbeddedKafkaBroker.