How to do integration testing of KStream topology using spring-boot EmbeddedKafka?

Question

I have a simple spring-boot KStream topology that transforms a string from lowercase to uppercase. I want my integration test to launch an embedded kafka, and then test the topology. I would like to know if it possible to write integration tests like these using spring @EmbeddedKafka?

I have seen several examples using @EmbeddedKafka with simple consumers using @KafkaListener but not any that uses KStream.

I tried attempting to test the following topology to transform from incoming text stream from lowercase to uppercase.

Here's the topology:

    @Configuration
    public class UppercaseStream {
    
        private static final String LOWERCASE_TOPIC = "t.lower.case";
        private static final String UPPERCASE_TOPIC = "t.upper.case";
    
        @Bean
        @Qualifier("kStreamPromoToUppercase")
        public KStream<String, String> kStreamPromoToUppercase(StreamsBuilder builder) {
    
            KStream<String, String> sourceStream = builder
                    .stream(LOWERCASE_TOPIC, Consumed.with(Serdes.String(), Serdes.String()));
    
            sourceStream.print(Printed.<String, String>toSysOut().withLabel("Original KStream..."));
    
            KStream<String, String> upperCaseStream = sourceStream.mapValues(text -> text.toUpperCase());
    
            upperCaseStream.print(Printed.<String, String>toSysOut().withLabel("Uppercase KStream..."));
    
            upperCaseStream.to(UPPERCASE_TOPIC);
    
            return upperCaseStream;
        }
    }

The unit test that tests the topology is:

    @TestInstance(TestInstance.Lifecycle.PER_CLASS)
    public class UpperCaseTopologyTest {
        TopologyTestDriver testDriver;
    
        @AfterAll
        void tearDown(){
            testDriver.close();
        }
    
        @Test
        @DisplayName("should transform lowercase to uppercase words")
        void shouldTransformLowercaseWords() {
            //Given
            StreamsBuilder builder = new StreamsBuilder();
            new UppercaseStream().kStreamPromoToUppercase(builder);
    
            Topology topology = builder.build();
            // setup test driver
            Properties props = new Properties();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-stream");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    
            //Create a Topology Test Driver 
            testDriver = new TopologyTestDriver(topology, props);
            TestInputTopic<String, String> inputTopic = testDriver.createInputTopic("t.lower.case", new Serdes.StringSerde().serializer(), new Serdes.StringSerde().serializer());
            TestOutputTopic<String, String> outputTopic = testDriver.createOutputTopic("t.upper.case", new Serdes.StringSerde().deserializer(), new Serdes.StringSerde().deserializer());
    
            //When
            inputTopic.pipeInput("test");
    
            //Then
            assertThat(outputTopic.readValue()).isEqualTo("TEST");
        }
    }

I want to write an integration test that first launches an embedded kafka server and then test the UppercaseStream topology.

I tried the following:

    @SpringBootTest
    @DirtiesContext
    @EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" })
    class EmbeddedKafkaIntegrationTest {
    
        @Autowired
        public KafkaTemplate<String, String> template;
        @Autowired
        private KafkaConsumer consumer;
        private KafkaStreams kafkaStreams;
    
        @Value("${test.topic}")
        private String topic;
    
        @Autowired
        private KafkaStreamsConfiguration kafkaStreamsConfiguration;
        
        @Test
        public void should_transform_lowercase_to_uppercase() throws Exception {
            //Create a StreamsBuilder
            StreamsBuilder streamsBuilder = new StreamsBuilder();
            streamsBuilder.stream(topic, Consumed.with(new Serdes.StringSerde(), new Serdes.StringSerde()));
    
            //Add a topology
            new UppercaseStream().kStreamPromoToUppercase(streamsBuilder);
            kafkaStreams = new KafkaStreams(streamsBuilder.build(), kafkaStreamsConfiguration.asProperties());
    
            kafkaStreams.start();
            template.send(topic, "test");
            consumer.getLatch().await(10000, TimeUnit.MILLISECONDS);
    
            assertThat(consumer.getLatch().getCount(), equalTo(0L));
            assertThat(consumer.getPayload(), containsString("TEST"));
        }
    
        @After
        public void tearDown() {
            if (kafkaStreams!= null) kafkaStreams.close();
    
        }
    }

The test fails the assertion. I am not sure how to get kStreamPromoToUppercase bean. I am not sure if I am trying following the correct approach.

Answer 1

There were a few things missing from the integration test.

A couple NewTopic kafka client admin objects were needed to represent an input and an output topics

@Bean public NewTopic createInputTopic() { return new NewTopic(inputTopic,Optional.of(1), Optional.empty()); } The other one is for the output topic

@Bean public NewTopic createOutputTopic() { return new NewTopic(outputTopic,Optional.of(1), Optional.empty()); }

The rest of the test remains more of less the same. As suggested by @Garry I used the kafka consumer.

    @SpringBootTest
    @EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" })
    class KStreamSampleApplicationTests {
    
    
        private final KafkaProperties kafkaProperties;
        private final String inputTopic;
        private final String outputTopic;
    
        @Autowired
        public KStreamSampleApplicationTests(KafkaProperties kafkaProperties, Environment env) {
            this.kafkaProperties = kafkaProperties;
            this.inputTopic = env.getProperty("spring.kafka.input-lowercase-topic");
            this.outputTopic = env.getProperty("spring.kafka.output-uppercase-topic");
    
        }
        
        @Test
        @DisplayName("should test uppercaseStream topology")        
        void shouldTestUppercaseStreamTopology() {
            //Given
            Map<String, Object> producerProps = new HashMap<>(KafkaTestUtils.producerProps(
                    String.join(",", kafkaProperties.getBootstrapServers())));
    
            //Create a kafka producer
            Producer<String, String> producer = new DefaultKafkaProducerFactory<>(producerProps, new StringSerializer(), new StringSerializer()).createProducer();
    
    
            Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(String.join(",", kafkaProperties.getBootstrapServers()), "testGroup", "true");
            consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    
            //Create a Consumer client
            Consumer<String, String> consumer = new DefaultKafkaConsumerFactory<>(consumerProps, new StringDeserializer(), new StringDeserializer()).createConsumer();
            consumer.subscribe(Collections.singleton(outputTopic));
    
            //When
            producer.send(new ProducerRecord<>(inputTopic, "test"));
            producer.flush();
    
            //Then
            assertThat(producer).isNotNull();
            //And
            ConsumerRecords<String, String> rec = consumer.poll(Duration.ofSeconds(3));
            Iterable<ConsumerRecord<String, String>> records = rec.records(outputTopic);
            Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
    
            if (!iterator.hasNext()) Assertions.fail();
    
            ConsumerRecord<String, String> next = iterator.next();
            assertThat(next.value()).isEqualTo("TEST");
        }
    }

Here's the gist of the full refactored solution.

Answer 2

It's not clear what your KafkaConsumer is; presumably some wrapper around a Consumer<K, V>.

Perhaps your consumer doesn't have auto.offset.reset=earliest. This is a common mistake with this kind of test where there is a race where the consumer might start after the record has been sent; the default is latest so you won't get such records.

The framework has a number of Kafka Streams test cases that use the embedded kafka broker.

https://github.com/spring-projects/spring-kafka/tree/master/spring-kafka/src/test/java/org/springframework/kafka/streams