Using Kafka With JUnit

One of the neat features that the Spring Kafka project provides, apart from an easier-to-use abstraction over raw Kafka Producer and Consumer, is a way to use Kafka in tests. It does this by providing an embedded version of Kafka that can be set up and torn down very easily. 

All that a project needs in order to include this support is the spring-kafka-test module for a Gradle build in the following way (as seen here): 

1testCompile "org.springframework.kafka:spring-kafka-test:1.1.2.BUILD-SNAPSHOT"          

(Note that I am using a snapshot version of the project as this has support for Kafka 0.10+.)

With this dependency in place, an embedded Kafka can be spun up in a test using the @ClassRule of JUnit:

@ClassRule public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(
  2, true, 2, "messages");

This would start up a Kafka Cluster with two brokers with a topic called "messages" using two partitions. The class rule would make sure that a Kafka cluster is spun up before the tests are run and then shut down at the end of it.

Here is how a sample how it looks with a raw Kafka producer and consumer using an embedded Kafka cluster. The embedded Kafka can be used for retrieving the properties required by the Kafka producer and consumer:

Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
KafkaProducer<Integer, String> producer = newKafkaProducer<>(senderProps);
producer.send(newProducerRecord<>("messages", 0, 0, "message0")).get();
producer.send(newProducerRecord<>("messages", 0, 1, "message1")).get();
producer.send(newProducerRecord<>("messages", 1, 2, "message2")).get();
producer.send(newProducerRecord<>("messages", 1, 3, "message3")).get();     
Map<String, Object> consumerProps =
  KafkaTestUtils.consumerProps("sampleRawConsumer", "false", embeddedKafka);
consumerProps.put("auto.offset.reset", "earliest");   
finalCountDownLatch latch = newCountDownLatch(4);
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(() -> {    
  KafkaConsumer<Integer, String> kafkaConsumer = newKafkaConsumer<>(consumerProps);
  kafkaConsumer.subscribe(Collections.singletonList("messages"));    
  try{       
    while(true) {            
    ConsumerRecords<Integer, String> records = kafkaConsumer.poll(100);            for(ConsumerRecord<Integer, String> record : records) {                LOGGER.info("consuming from topic = {}, partition = {}, offset = {}, key = {}, value = {}",                        record.topic(), record.partition(), record.offset(), record.key(), record.value());                latch.countDown();            }        }    } finally{        kafkaConsumer.close();    }});                
assertThat(latch.await(90, TimeUnit.SECONDS)).isTrue();

A little more comprehensive test is available here.