title: spring Kafka integration testing with embedded Kafka
body: <p>I have spring boot application that had a consumer consumes from topic in one cluster and produces to another topic in different cluster.</p>
  <p>Now I'm trying to write integration test case using spring embedded Kafka but having an issue <code>KafkaTemplate could not be registered. A bean with that name has already been defined in class path resource</code></p>
  <p><strong>Consumer Class</strong></p>
  <pre><code>@Service
  public class KafkaConsumerService {
  
  @Autowired
  private KafkaProducerService kafkaProducerService;
  
  @KafkaListener(topics = &quot;${kafka.producer.topic}&quot;)
  public void professor(List&lt;Professor&gt; pro) {
      pro.forEach(kafkaProducerService::produce);
      
     }
  
  }
  </code></pre>
  <p><strong>Producer Class</strong></p>
  <pre><code>@Service
  public class KafkaProducerService {
  
  @Value(&quot;${kafka.producer.topic}&quot;)
  private String topic;
  
  @Autowired
  private KafkaTemplate&lt;String, Object&gt; kafkaTemplate;
  
  public void produce(Professor pro) {
      kafkaTemplate.send(topic,&quot;professor&quot;,pro);
    }
  
   }
  </code></pre>
  <p>In my Test cases I want to override <code>KafkaTemplate</code> so that when i call <code>kafkaConsumerService.professor</code> method in <code>Test</code> it should produce the data into embedded Kafka and i should validate it.</p>
  <p><strong>Test config</strong></p>
  <pre><code>@TestConfiguration
  @EmbeddedKafka(partitions = 1, controlledShutdown = false,
  brokerProperties = {&quot;listeners=PLAINTEXT://localhost:3333&quot;, &quot;port=3333&quot;})
  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&lt;String, Object&gt; producerFactory() {
      return new DefaultKafkaProducerFactory&lt;&gt;(KafkaTestUtils.producerProps(kafkaEmbeded));
  }
  
  @Bean
  public KafkaTemplate&lt;String, Object&gt; kafkaTemplate() {
      KafkaTemplate&lt;String, Object&gt; kafkaTemplate = new KafkaTemplate&lt;&gt;(producerFactory());
      return kafkaTemplate;
     }
  
   }
  </code></pre>
  <p><strong>Test class</strong></p>
  <pre><code>@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?
  }
  
   }
  </code></pre>
  <p><strong>Error</strong></p>
  <pre><code> 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
  </code></pre>
  <hr />
  <p>And also can some one help me how to validate messages sent into embedded Kafka server?</p>
  <p><strong>Note</strong> I'm having some deprecated warnings</p>
  <blockquote>
  <p>The type KafkaEmbedded is deprecated</p>
  <p>The method getPartitionsPerTopic() from the type KafkaEmbedded is deprecated</p>
  <p>The method producerProps(KafkaEmbedded) from the type KafkaTestUtils is deprecated</p>
  </blockquote>