How to test Kafka Streams applications with Spring Kafka?

Question

I am writing a streaming application with Kafka Streams, Spring-Kafka and Spring Boot. I cannot find any information how to properly test stream processing done by Kafka Streams DSL while using Spring-Kafka. Documentation mentions EmbeddedKafkaBroker but there seems to be no information on how to handle testing for example state stores.

Just to provide some simple example of what I would like to test. I have a following bean registered (where Item is avro generated):


    @Bean
    public KTable<String, Long> itemTotalKTable(StreamsBuilder streamsBuilder) {
        return streamsBuilder
                .stream(ITEM_TOPIC,
                        Consumed.with(Serdes.String(), itemAvroSerde))
                .mapValues((id, item) -> item.getNumber())
                .groupByKey()
                .aggregate(
                        () -> 0L,
                        (id, number, agg) -> agg + number,
                        Materialized.with(Serdes.String(), Serdes.Long()));
    }

What is a proper way to test that all item numbers are aggregated?

score 2 · Answer 1 · answered Sep 03 '19 at 15:44

Spring Kafka for Kafka Streams support doesn't bring any extra API, especially in streams building and their processing.

We have opened recently for ourselves that there is a good kafka-streams-test-utils library to be used in unit tests without any Kafka broker start (even embedded).

In several our tests we have something like this:

    KStream<String, String> stream = builder.stream(INPUT);
    stream
            .transform(() -> enricher)
            .to(OUTPUT);

    Properties config = new Properties();
    config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
    config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
    TopologyTestDriver driver = new TopologyTestDriver(builder.build(), config);

    ConsumerRecordFactory<String, String> recordFactory = new ConsumerRecordFactory<>(new StringSerializer(),
            new StringSerializer());
    driver.pipeInput(recordFactory.create(INPUT, "key", "value"));
    ProducerRecord<byte[], byte[]> result = driver.readOutput(OUTPUT);
    assertThat(result.headers().lastHeader("foo")).isNotNull();

I believe there should some API in that TopologyTestDriver to deal with the mentioned state store.

If anybody needs, I shared a GitHub repo that shows how to integrate Spring Kafka and the test library `kafka-streams-test-utils`: https://github.com/rcardin/spring-boot-kafka-stream-test — riccardo.cardin, May 22 '20 at 07:52

score 0 · Answer 2 · answered Sep 03 '19 at 16:16

Maybe you could make a method which takes your KTable as a param and calls .toStream().to(topicname,Produced.with(keyserde, valueserde)) on it, then you could do the following:

MyTopologyBuilder builder = new MyTopologyBuilder();

testDriver = new TopologyTestDriver(builder.build(), config);

ConsumerRecord<byte[], byte[]> input = createStepRecord(key, record);

testDriver.pipeInput(input);

ProducerRecord<String, String> out testDriver.readOutput(topic, new StringDeserializer(), ew AvroDeserializer<>(MyClass.class);

assertThat(out.key(), is(key));
assertEquals(myPredefinedValue, out.value());
assertEquals(5, out.value().getMyList().size());

This should work, but there could be more elegant ways I guess.

How to test Kafka Streams applications with Spring Kafka?

2 Answers2