Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

The testing pyramid is a representation of the most widely accepted testing methods. It illustrates how teams should distribute their testing efforts.

Figure 6.7: The testing pyramid

At the bottom of the pyramid, tests are faster, specific, simpler, and easier to write. As you go up the pyramid, tests cover broader scopes but become more difficult to develop, and maintain.

In EDAs using a Pub-Sub architecture, you implement Kafka producers to publish events, and Kafka consumers to subscribe to events. Decoupling the business logic from these producers and consumers is vital to achieve isolated, easy-to-write unit tests, and it helps to make tests independent of the Kafka infrastructure.

Decoupled business logic code should not depend on specific event messaging systems, such as Kafka. After decoupling, business logic code units should normally emit and listen to events. Consequently, your unit tests will generally test the following:

To write tests, you can use JUnit. JUnit is the de facto standard for unit tests in Java. In other languages, similar libraries exist. For example, consider the following Java code, which implements a simple mechanism to process book orders.

public class BookStore {

    ...omitted...

    public BookOrderCreated order(Book book, CustomerInfo customerInfo)
            throws OutOfStockException {

        if (inventory.has(book)) {
            Order order = new Order(book, customerInfo);
            ordersRegistry.register(order);

            return new BookOrderCreated(order);
        }

        throw new OutOfStockException(book);
    }
}

The order method returns a BookOrderCreated event, if the book is available in the inventory. Note that the method's only responsibility is to verify the business rule and emit an event if the condition is met. The code does not include any Kafka dependencies or Kafka producer code.

You can implement a unit test for such a scenario as follows:

@Test
public void testProcessOrderReturnsBookOrderCreated() {
    // Given a book, included in the inventory
    Book book = buildTestBook();
    CustomerInfo customerInfo = buildTestCustomerInfo();
    Inventory inventory = buildTestInventory().add(book);

    // When the store processes an book order
    BookOrderCreated event = store.order(book, customerInfo);

    // Then the produced event corresponds to the ordered book
    assertBookOrderCreatedForBook(event, book);
}

The unit test only covers the BookStore#order method. Both the BookStore class and the test are unaware of the Kafka infrastructure layer, which is implemented in a different application layer. Other dependencies, such as the inventory, are implemented as test doubles.

If you develop a Kafka Streams application, then you can use the Kafka Streams test utils package to unit test topologies. The test utils package decouples tests from Kafka, by providing methods to define test doubles for Kafka topics. By using test doubles to simulate streams, tests become easier to implement and do not require a running Kafka instance.

You can install Kafka Streams test utils by including the kafka-streams-test-utils artifact as a Maven dependency.

To test a topology, pass the topology to the TopologyTestDriver constructor. The TopologyTestDriver object creates a test double of a Kafka infrastructure and executes the topology on top of this simulated infrastructure. Next, by using the TopologyTestDriver methods, you specify the data that the test should expect in each topic.

For example, consider the following method, which builds a Kafka Streams topology.

public Topology getBookOrdersTopology() {
    KSTream<String, BookOrderCreated> ordersStream = builder.stream(
        "book-orders-created"
    )

    ordersStream
        //...transformations omitted...
        .to("book-orders-confirmed")

    ordersStream
        //...transformations omitted...
        .count(Materialized.as("book-orders-counts"))

    return builder.build();
}

The getBookOrdersTopology method reads a stream of events from the book-orders-created topic, transforms the stream, and materializes results in the book-orders-confirmed topic and the book-orders-counts store.

You can unit test this topology as follows:

import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.test.TestRecord;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.state.KeyValueStore;

...class omitted...

@Test
public void testTopologyConfirmsBookOrders() {
    // Given 
    TopologyTestDriver testDriver = new TopologyTestDriver(getBookOrdersTopology());
    TestInputTopic<String, BookOrderCreated> inputTopic = testDriver.createInputTopic(
        "book-orders-created",
        stringSerializer,
        bookOrderCreatedSerializer
    );
    TestOutputTopic<String, BookOrderCreated> outputTopic = testDriver.createOutputTopic(
        "book-order-confirmed",
        stringDeserializer,
        bookOrderConfirmedDeserializer
    );

    // When 
    BookOrderCreated inputEvent = createTestOrder("book-1");
    inputTopic.pipeInput("order-42", inputEvent);

    // Then 
    TestRecord<String, BookOrderConfirmed> record = outputTopic.readRecord();
    BookOrderConfirmed outputEvent = record.getValue();
    assertThat(outputEvent.bookId, "book-1");
}

The TopologyTestDriver object also allows you to handle and verify stores.

import org.apache.kafka.streams.state.KeyValueStore;

...class omitted...

@Test
public void testTopologyCountsBookOrders() {
    // Given 
    testDriver = new TopologyTestDriver(getBookOrdersTopology());
    inputTopic = testDriver.createInputTopic(
        "book-orders-created",
        stringSerializer,
        bookOrderCreatedSerializer
    );
    KeyValueStore<String, Integer> bookCountStore = testDriver.getKeyValueStore(
        "book-orders-counts"
    )

    // When 
    inputTopic.pipeInput("order-32", createTestOrder("book-1"));
    inputTopic.pipeInput("order-33", createTestOrder("book-1"));

    // Then 
    assertThat(bookCountStore.get("book-1"), equalTo(2));
}