title: How to test a kafka consumer against a real kafka broker running on a server?
body: <p>I have difficulty understanding some Kafka concepts in Java Spring Boot. I’d like to test a consumer against a real Kafka broker running on a server, which has some producers that write / have already written data to various topics. I would like to establish a connection with the server, consume the data, and verify or process its content in a test.</p>
  
  <p>An enormous majority of examples (actually all I have seen so far) in the internet refer to embedded kafka, EmbeddedKafkaBroker, and show both a producer and a consumer implemented on one machine, locally. I haven’t found any example that would explain how to make a connection with a remote kafka server and read data from a particular topic.
  I've written some code and I've printed the broker address with:</p>
  
  <pre><code>System.out.println(embeddedKafkaBroker.getBrokerAddress(0));
  </code></pre>
  
  <p>What I got is 127.0.0.1:9092, which means that it is local, so the connection with the remote server has not been established.</p>
  
  <p>On the other hand, when I run the SpringBootApplication I get the payload from the remote broker.</p>
  
  <p>Receiver:</p>
  
  <pre><code>@Component
  public class Receiver {
  
  private static final String TOPIC_NAME = "X";
  
  private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);
  
  private CountDownLatch latch = new CountDownLatch(1);
  
  public CountDownLatch getLatch() {
      return latch;
  }
  
  @KafkaListener(topics = TOPIC_NAME)
  public void receive(final byte[] payload) {
      LOGGER.info("received the following payload: '{}'", payload);
      latch.countDown();
  }
  }
  </code></pre>
  
  <p>Config:</p>
  
  <pre><code>    @EnableKafka
      @Configuration
      public class ByteReceiverConfig {
  
          @Autowired
          EmbeddedKafkaBroker kafkaEmbeded;
  
          @Value("${spring.kafka.bootstrap-servers}")
          private String bootstrapServers;
  
          @Value("${spring.kafka.consumer.group-id}")
          private String groupIdConfig;
  
          @Bean
          public KafkaListenerContainerFactory&lt;?&gt; kafkaListenerContainerFactory() {
              final ConcurrentKafkaListenerContainerFactory&lt;Object, Object&gt; factory =
                      new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
              factory.setConsumerFactory(consumerFactory());
              return factory;
          }
  
          @Bean
          ConsumerFactory&lt;Object, Object&gt; consumerFactory() {
              return new DefaultKafkaConsumerFactory&lt;&gt;(consumerProperties());
          }
  
          @Bean
          Map&lt;String, Object&gt; consumerProperties() {
              final Map&lt;String, Object&gt; properties =
                      KafkaTestUtils.consumerProps("junit-test", "true", this.kafkaEmbeded);
              properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
              properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
              properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
              properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupIdConfig);
              return properties;
          }
  </code></pre>
  
  <p>Test:</p>
  
  <pre><code>        @EnableAutoConfiguration
          @EnableKafka
          @SpringBootTest(classes = {ByteReceiverConfig.class, Receiver.class})
          @EmbeddedKafka
          @ContextConfiguration(classes = ByteReceiverConfig.class)
          @TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
                  "spring.kafka.consumer.group-id=EmbeddedKafkaTest"})
          public class KafkaTest {
  
  
              @Autowired
              private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
  
              @Autowired
              EmbeddedKafkaBroker embeddedKafkaBroker;
  
  
              @Autowired
              Receiver receiver;
  
              @BeforeEach
              void waitForAssignment() {
                  for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
                      System.out.println(messageListenerContainer.getAssignedPartitions().isEmpty());
                      System.out.println(messageListenerContainer.toString());
                      System.out.println(embeddedKafkaBroker.getTopics().size());
                      System.out.println(embeddedKafkaBroker.getPartitionsPerTopic());
                      System.out.println(embeddedKafkaBroker.getBrokerAddress(0));
                      System.out.println(embeddedKafkaBroker.getBrokersAsString());
  
                      ContainerTestUtils.waitForAssignment(messageListenerContainer,
                              embeddedKafkaBroker.getPartitionsPerTopic());
              }
  
              @Test
              public void testReceive() {
  
              }
          }
  </code></pre>
  
  <p>I would like somebody to shed some light on the following issues:</p>
  
  <p>1.Can an instance of  the class EmbeddedKafkaBroker be used to test data that comes from a remote broker, or is it only used for local tests, in which I would procude i.e send data to a topic that I created and consume data myself?</p>
  
  <p>2.Is it possible to write a test class for a real kafka server? For instance to verify if a connection has been establish, or if a data has been read from a specific topic. What annotations, configurations, and classes would be needed in such case?</p>
  
  <p>3.If I only want to consume data, do I have to provide the producer configuration in a config file (it would be strange, but all examples I have encountered so far did it)?</p>
  
  <p>4.Do you know any resources (books, websites etc.) that show real examples of using kafka i.e. with a remote kafka server, with a procuder or a consumer only?</p>