/// Resources / Kafka

Testing Kafka streams in Java: produce, consume and assert in a Cucumber scenario

Published 2026-11-056 min readL3git Development

Event-driven services are hard to test from the outside because nothing comes back on the request. The assertion has to be a message appears on a topic within a timeout. Here is the small amount of Java that makes that a readable Gherkin step, and the pitfalls that make most first attempts flaky.

What the scenario should read like

Scenario: An order event reaches the fulfilment topic
  When I post to "/orders" with body:
    """
    { "sku": "A-100", "qty": 2 }
    """
  Then the response status should be 201
  And I should receive a kafka message on the topic "orders.created" containing "A-100"

The REST step is ordinary REST Assured. The last line is the Kafka assertion. Everything about brokers, deserialisers and polling lives in a support class, not in the feature file.

The support class

Two methods are enough: produce a record, and consume until a record matches or a timeout expires. This is the shape of the KafkaClient that L3git generates into API projects when a Kafka step is used (the kafka-clients dependency is added to the pom only then):

public void produce(String bootstrap, String topic, String key, String value) {
    Properties p = new Properties();
    p.put("bootstrap.servers", bootstrap);
    p.put("key.serializer",   StringSerializer.class.getName());
    p.put("value.serializer", StringSerializer.class.getName());
    try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
        producer.send(new ProducerRecord<>(topic, key, value)).get();   // wait for the ack
    }
}

public String consumeExpect(String bootstrap, String topic, String contains, long timeoutMs) {
    Properties p = new Properties();
    p.put("bootstrap.servers", bootstrap);
    p.put("group.id", "l3git-test-" + UUID.randomUUID());               // fresh group per scenario
    p.put("auto.offset.reset", "earliest");
    p.put("key.deserializer",   StringDeserializer.class.getName());
    p.put("value.deserializer", StringDeserializer.class.getName());
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p)) {
        consumer.subscribe(List.of(topic));
        long end = System.currentTimeMillis() + timeoutMs;
        while (System.currentTimeMillis() < end) {
            for (ConsumerRecord<String, String> r : consumer.poll(Duration.ofMillis(500))) {
                if (contains.isBlank() || r.value().contains(contains)) return r.value();
            }
        }
    }
    throw new AssertionError("No message on " + topic + " containing '" + contains + "' within " + timeoutMs + "ms");
}

The step definitions

@When("I produce to the kafka topic {string}:")
public void produce(String topic, String message) throws Exception {
    kafka.produce(cfg.broker(), topic, "", message);
}

@Then("I should receive a kafka message on the topic {string} containing {string}")
public void consume(String topic, String contains) {
    kafka.consumeExpect(cfg.broker(), topic, contains, 10_000);
}

Four things that make Kafka tests flaky, and the fix for each

1. Shared consumer group. Two scenarios in the same group split partitions and one of them never sees the record. Use a random group.id per scenario, as above.
2. Reading from latest. If the producer wins the race, the consumer subscribes after the record landed and waits forever. auto.offset.reset=earliest with a fresh group reads from the start of retention, and the contains match ignores older records.
3. No ack on produce. send() is asynchronous. Call .get() so the step does not pass before the broker has the record.
4. One global timeout. Ten seconds is right on a laptop and wrong on a loaded CI broker. Make it a property (kafka.timeout.ms) and raise it per environment, not per test.

Running it in CI

The broker address comes from an environment variable or healing.properties, never the feature file, so the same scenario runs against a laptop Redpanda, a staging cluster or a Testcontainers Kafka in Jenkins. Because the output is plain Java in a Maven project there is nothing extra to license for CI; the Kafka steps run on every agent in the matrix like any other scenario.

In L3git these two steps are recorded in the API tab (topic, broker, key, body, timeout) and the Java above is generated for you, alongside gRPC via server reflection and WebSocket steps. See the proof page for a real run against a live broker and guide 10 for setup.

Questions