Sprint Cloud - Kafka Streams Binder - Testing KafkaStreamsProcessor

Question

I've been trying to write unit tests for a KafkaStreamsProcessor. This is the code processor code

@EnableBinding(KafkaStreamsProcessor.class)
public class StockProcessor {

    private static final Log LOG = LogFactory.getLog(StockProcessor.class);

    @Autowired
    private EddieClient client;

    @Autowired
    private InventoryRepository inventoryRepository;

    @Autowired
    private PermissionRepository permissionRepository;

    /**
     * Receive message from input queue
     * Apply business logic
     * Send to output queue
     *
     * @param inputMessage the message
     * @return outputMessage
     */
    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public KStream<?, OutputMessage> process(KStream<?, InputMessage> inputMessage){
        return inputMessage
            .map((key, value) -> {
                LOG.info("::: processing message...");
               //  ... business logic
                return new KeyValue<>(key, outputMessage);
            });
    }
}

aplication.yml

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers:
              - ${NX_KAFKA_SERVERS}
      bindings:
        input:
          destination: ${NX_INPUT_TOPIC}
          content-type: application/json
          group: ${NX_PULL_GROUP_ID}
        output:
          destination: ${NX_OUTPUT_TOPIC}
          content-type: application/json
          group: ${NX_PUSH_GROUP_ID}

This is what I've read and tried to do in unit test

public class StockProcessorTest {

    private static final String INPUT_TOPIC = "input-topic";
    private static final String OUTPUT_TOPIC = "output-topic";

    @SpyBean
    private StockProcessor stockProcessor;

    @MockBean
    private EddieClient client;

    @MockBean
    private InventoryRepository inventoryRepository;

    @MockBean
    private PermissionRepository permissionRepository;

    private TopologyTestDriver topologyTestDriver;
    private TestInputTopic<String, InputMessage> inputTopic;
    private TestOutputTopic<String, OutputMessage> outputTopic;

    private Topology topology;
    private Properties config;

    KStream<String, InputMessage> inputMessageStream;
    KStream<String, OutputMessage> outputMessageStream;

    @Before
    public void setup() {
        config = new Properties();
        config.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "app_id");
        config.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "foo:1234");

        StreamsBuilder streamsBuilder = new StreamsBuilder();
        inputMessageStream = streamsBuilder.stream(INPUT_TOPIC);
        stockProcessor.process(inputMessageStream).to(OUTPUT_TOPIC);

        topology = streamsBuilder.build();
        topologyTestDriver = new TopologyTestDriver(topology, config);

        //???
    }
}

I don't really know if I'am going to right path here. I am using Jackson serializer. How can I create the inputTopic and outputTopic and test my business logic?

I can provide any further detail needed. Thanks in advance

Answer 1

Here are some examples for how to unit test a Kafka Streams application based on Spring Cloud Stream - https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/kafka-streams-samples/kafka-streams-word-count/src/test/java/kafka/streams/word/count/WordCountProcessorApplicationTests.java

Also, here is a test suite that has some advanced examples: https://github.com/spring-cloud/spring-cloud-stream-samples/tree/master/kafka-streams-samples/kafka-streams-inventory-count/src/test/java/kafka/streams/inventory/count

These examples have details about how to use Serdes with the test driver.

Please check them out and see if they satisfy your testing requirements.