title: How to set port in KafkaEmbedded when unit testing spring-kafka consumer
body: <p>I am using <code>spring-boot-starter-parent</code> version <code>1.5.0.RELEASE</code>, <code>spring-kafka</code> version <code>1.0.0.RELEASE</code> and <code>spring-kafka-test</code> version <code>1.0.0.RELEASE</code> in an application which consumes messages from a <code>Kakfa 0.9</code> cluster. I have a unit test for my consumer which used <code>KafkaEmbedded</code> but it fails since the broker port is picked up randomly. Is there a way I can set this broker property without changing versions? Or which versions should I use so as to not break anything?</p>
  <p>Here is the code for the <code>KafkaListener</code> and <code>KafkaConsumerTest</code>.</p>
  <h3><strong>Listener.java</strong></h3>
  <pre><code>@Service
  public class Listener {
  
      private static final Logger logger = LoggerFactory.getLogger(Listener.class);
      private CountDownLatch latch = new CountDownLatch(1);
  
      @KafkaListener(topics = &quot;topic&quot;, group = &quot;group&quot;, containerFactory = &quot;kafkaListenerContainerFactory&quot;)
      public void consumeClicks(@Payload String msg, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition, @Header(KafkaHeaders.OFFSET) Integer offset, Acknowledgment ack) throws Exception {
          logger.info(msg);
          latch.countDown();
          ack.acknowledge();
      }
  
      public CountDownLatch getLatch() {
          return latch;
      }
  }
  
  </code></pre>
  <h3><strong>KafkaConsumerTest.java</strong> (<em>EDIT</em>)</h3>
  <pre><code>@DirtiesContext
  @SpringBootTest(classes = {SpringApplication.class})
  @RunWith(SpringRunner.class)
  public class KafkaConsumerTest {
      private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerTest.class);
      private static String TEST_TOPIC = &quot;topic&quot;;
  
      @ClassRule
      public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC);
  
      public KafkaTemplate&lt;String, String&gt; template;
  
      @Autowired
      private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
  
      @Autowired
      private Listener listener;
  
      @Before
      public void init(){
          System.setProperty(&quot;spring.kafka.bootstrap-servers&quot;, embeddedKafka.getBrokersAsString());
          Map&lt;String, Object&gt; senderProps = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString());
          senderProps.put(&quot;key.serializer&quot;, StringSerializer.class);
          ProducerFactory&lt;String, String&gt; producerFactory = new DefaultKafkaProducerFactory&lt;String, String&gt;(senderProps);
          template = new KafkaTemplate&lt;&gt;(producerFactory);
          template.setDefaultTopic(TEST_TOPIC);
      }
  
      @Test
      public void testConsume() throws Exception {
          String record = &quot;message&quot;;
          template.sendDefault(TEST_TOPIC, record);
          logger.debug(&quot;test-consume sent record {}&quot;, record);
          listener.getLatch().await(1000, TimeUnit.MILLISECONDS);
          Assert.assertEquals(listener.getLatch().getCount(), 0);
      }
  }
  
  </code></pre>