Apache kafka embedded kafka junit test - application starting when I run unit test

Question

I am developing an asynchronous mail server in spring boot using kafka.

I have written tests with embedded kafka which starts its own kafka topic in a random port and use it for testing.

When I started this application context is loading and its expecting for kafka cluster in my local. I need to stop application conext from loading. I replicated the code from https://github.com/code-not-found/spring-kafka/blob/master/spring-kafka-unit-test-classrule/src/test/java/com/codenotfound/kafka/producer/SpringKafkaSenderTest.java which works absolutely fine. When I followed same style in my project, I can see actual apllication starting.

SpringKafkaSenderTest .java

package com.mailer.embeddedkafkatests;
import static org.junit.Assert.assertTrue;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.utils.ContainerTestUtils;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;

import com.mailer.model.Mail;
import com.mailer.producer.KafkaMessageProducer;
import com.mailer.serializer.MailSerializer;

@RunWith(SpringRunner.class)
@SpringBootTest
@DirtiesContext
public class SpringKafkaSenderTest {

  private static final Logger LOGGER =
      LoggerFactory.getLogger(SpringKafkaSenderTest.class);

  private static String SENDER_TOPIC = "sender.t";

  @Autowired
  private KafkaMessageProducer sender;

  private KafkaMessageListenerContainer<String, Mail> container;

  private BlockingQueue<ConsumerRecord<String, Mail>> records;

  @ClassRule
  public static EmbeddedKafkaRule embeddedKafka =
      new EmbeddedKafkaRule(1, true, SENDER_TOPIC);

  @Before
  public void setUp() throws Exception {
    // set up the Kafka consumer properties
    Map<String, Object> consumerProperties =
        KafkaTestUtils.consumerProps("sender", "false",
            embeddedKafka.getEmbeddedKafka());
    consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MailSerializer.class);

    // create a Kafka consumer factory
    DefaultKafkaConsumerFactory<String, Mail> consumerFactory =
        new DefaultKafkaConsumerFactory<String, Mail>(
            consumerProperties);//, new StringDeserializer(), new JsonDeserializer<>(Mail.class));

    // set the topic that needs to be consumed
    ContainerProperties containerProperties =
        new ContainerProperties(SENDER_TOPIC);

    // create a Kafka MessageListenerContainer
    container = new KafkaMessageListenerContainer<>(consumerFactory,
        containerProperties);

    // create a thread safe queue to store the received message
    records = new LinkedBlockingQueue<>();

    // setup a Kafka message listener
    container
        .setupMessageListener(new MessageListener<String, Mail>() {
          @Override
          public void onMessage(
              ConsumerRecord<String, Mail> record) {
            LOGGER.debug("test-listener received message='{}'",
                record.toString());
            records.add(record);
          }
        });

    // start the container and underlying message listener
    container.start();

    // wait until the container has the required number of assigned partitions
    ContainerTestUtils.waitForAssignment(container,
        embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic());
  }

  @After
  public void tearDown() {
    // stop the container
    container.stop();
  }

  @Test
  public void testSend() throws InterruptedException {
    // send the message
    Mail mail = new Mail();
    mail.setFrom("vinoth@local.com");
    sender.sendMessage(mail);
    Thread.sleep(4000);
    // check that the message was received
    ConsumerRecord<String, Mail> received =
        records.poll(10, TimeUnit.SECONDS);
    // Hamcrest Matchers to check the value
    assertTrue(received.value().getFrom().equals(mail.getFrom()));
    System.out.println(received.value().getFrom());
//    assertThat(received, hasValue(mail));
    // AssertJ Condition to check the key
//    assertThat(received).has(key(null));
  }
}

Answer 1

Why would you like to stop the spring context from loading? Isn't the purpose of this junit to test your spring application?

In any case just remove the @SpringBootTest annotation and the spring context will not load.