Testing and Debugging Kafka Streams Applications

Kafka Streams lets you build stateful, real-time processing directly inside a Java or Scala service, with no separate processing cluster to operate. That convenience comes with a cost: local state stores, changelog topics, repartition topics, and multi-threaded task assignment all live inside your application, so failures rarely reproduce with a simple unit test. This guide is a code-driven, version-pinned walkthrough of how to test and debug Streams applications: deterministic unit tests with TopologyTestDriver, integration tests with an embedded broker, state-store introspection, exception handling, exactly-once validation, and production profiling.

The examples target Apache Kafka 3.7.x. Where behavior is version-specific (notably exactly-once and the exception-handler signatures), that is called out explicitly.

Unit testing topologies with TopologyTestDriver

The bulk of your coverage should be deterministic unit tests that run without a broker. TopologyTestDriver drives a built Topology synchronously: you pipe input records in, and assert on output records and state-store contents. Because it processes each record synchronously, commit.interval.ms and cache.max.bytes.buffering have no effect inside the driver — it behaves as if a commit and flush happen after every record. That makes tests fast and reproducible, but it also means the driver does not exercise commit batching, caching, or transaction boundaries; those need integration tests (covered below).

Add the test-utils artifact (test scope only):

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams-test-utils</artifactId>
    <version>3.7.0</version>
    <scope>test</scope>
</dependency>

A stateless topology — filter out empty payloads, take the first comma-separated field, and forward — is the simplest case. Note that the default.*.serde config values are class names, so Serdes.StringSerde.class.getName() is the idiomatic form (a raw Serdes.String().getClass() works too, but reads worse):

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Produced;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.Properties;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

public class EventFilterTopologyTest {

    private TopologyTestDriver testDriver;
    private TestInputTopic<String, String> inputTopic;
    private TestOutputTopic<String, String> outputTopic;

    @Before
    public void setUp() {
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());

        StreamsBuilder builder = new StreamsBuilder();
        builder.stream("raw-events", Consumed.with(Serdes.String(), Serdes.String()))
               .filter((key, value) -> value != null && !value.isEmpty())
               .mapValues(value -> value.split(",")[0])
               .to("clean-events", Produced.with(Serdes.String(), Serdes.String()));

        testDriver = new TopologyTestDriver(builder.build(), config);
        inputTopic = testDriver.createInputTopic("raw-events",
                Serdes.String().serializer(), Serdes.String().serializer());
        outputTopic = testDriver.createOutputTopic("clean-events",
                Serdes.String().deserializer(), Serdes.String().deserializer());
    }

    @After
    public void tearDown() {
        testDriver.close();
    }

    @Test
    public void testFilterAndMap() {
        inputTopic.pipeInput("key1", "value1,extra");
        inputTopic.pipeInput("key2", null);   // filtered out
        inputTopic.pipeInput("key3", "value3,data");

        assertEquals("value1", outputTopic.readValue());
        assertEquals("value3", outputTopic.readValue());
        assertTrue(outputTopic.isEmpty());     // nothing left
    }
}

outputTopic.isEmpty() is the correct way to assert “no more output”; calling readValue() on a drained topic throws NoSuchElementException rather than returning null, so don’t assert on a null read.

Testing stateful operations and state stores

For stateful topologies you must assert on the internal state, not just the output. TopologyTestDriver exposes stores directly via getKeyValueStore(), getWindowStore(), and getSessionStore(). If your store carries timestamps (the default for DSL aggregations), prefer getTimestampedKeyValueStore() for full access — the plain getKeyValueStore() view drops the timestamp and is best avoided for timestamped stores.

A groupByKey().count(Materialized.as("event-counts")) topology produces a KeyValueStore<String, Long>:

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

@Test
public void testStatefulCount() {
    // Topology built in setUp(): groupByKey().count(Materialized.as("event-counts"))
    inputTopic.pipeInput("A", "event1");
    inputTopic.pipeInput("A", "event2");
    inputTopic.pipeInput("B", "event3");

    KeyValueStore<String, Long> store = testDriver.getKeyValueStore("event-counts");

    assertEquals(Long.valueOf(2), store.get("A"));
    assertEquals(Long.valueOf(1), store.get("B"));
    assertNull(store.get("C"));
}

Time-driven logic needs care. There are two clocks: event time (the record timestamp) and wall-clock time (used by Punctuators scheduled with PunctuationType.WALL_CLOCK_TIME). You set event time by passing a timestamp to pipeInput(...); you advance the wall clock with advanceWallClockTime(Duration), which fires any wall-clock punctuations and lets you test scheduled emits without Thread.sleep. Note that advanceWallClockTime() does not push event-time windows forward — windowing and grace periods are driven by the timestamps of the records you feed in.

import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import java.time.Duration;
import java.time.Instant;

@Test
public void testWindowedAggregation() {
    Instant base = Instant.parse("2024-01-01T00:00:00Z"); // fixed, not Instant.now()
    inputTopic.pipeInput("sensor-1", "25.5", base);
    inputTopic.pipeInput("sensor-1", "26.0", base.plusSeconds(30));

    // Fire any wall-clock punctuator (e.g. a scheduled flush/emit)
    testDriver.advanceWallClockTime(Duration.ofMinutes(5));

    WindowStore<String, Double> store = testDriver.getWindowStore("sensor-avg");
    try (WindowStoreIterator<Double> it = store.fetch("sensor-1", base, base.plusMinutes(1))) {
        assertTrue(it.hasNext());
        // assert on the windowed aggregate values...
    }
}

Use a fixed base instant rather than Instant.now() so window-boundary assertions stay deterministic, and close the WindowStoreIterator — like a RocksDB iterator, it holds resources.

Integration testing with an embedded broker

Some behavior only appears against a real broker: rebalancing, repartition/changelog topic creation, isolation levels, and end-to-end serde compatibility. kafka-streams-test-utils does not ship an embedded broker — it provides only TopologyTestDriver. For an in-JVM broker, use Spring Kafka Test (EmbeddedKafkaBroker), which since Kafka 4.0 runs in KRaft mode with no ZooKeeper.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;

import java.util.Map;

import static org.junit.jupiter.api.Assertions.assertEquals;

@EmbeddedKafka(partitions = 1, topics = {"input-topic", "output-topic"})
public class IntegrationTest {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    public void testEndToEndFlow() {
        Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps)) {
            producer.send(new ProducerRecord<>("input-topic", "key1", "value1"));
            producer.flush();
        }

        // Start your KafkaStreams app here against embeddedKafka.getBrokersAsString(),
        // then wait for it to reach RUNNING before consuming.

        Map<String, Object> consumerProps =
                KafkaTestUtils.consumerProps("test-group", "true", embeddedKafka);
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic");
            ConsumerRecord<String, String> record =
                    KafkaTestUtils.getSingleRecord(consumer, "output-topic");
            assertEquals("transformed-value1", record.value());
        }
    }
}

Integration tests are slow and flaky-prone, so keep them for critical paths — topology startup, dead-letter routing, exactly-once — and rely on TopologyTestDriver for the rest. The DSL-versus-Processor-API choice also shapes your test surface; see Kafka Streams DSL vs Processor API: When to Use Each. For background on the state semantics you are validating, see Stateful Stream Processing with Kafka Streams.

Debugging live topologies in production

Streams exposes JMX metrics and runtime APIs so you can inspect a running application without restarting it. The main levers are JMX metrics, the cluster-wide metadata API, and interactive queries against state stores.

The JmxReporter is registered by default, so client and stream metrics are already exposed over JMX. To ship them elsewhere you typically run the Prometheus JMX Exporter as a Java agent, or add a MetricsReporter implementation via metric.reporters:

// Only needed to ADD a reporter; JmxReporter is on by default.
props.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG,
          "com.example.MyMetricsReporter");

The MBean object names form a four-level hierarchy (client → thread → task → node/store). The exact patterns are:

• Client level: kafka.streams:type=stream-metrics,client-id=<id> — alive/failed thread counts and client state.
• Thread level: kafka.streams:type=stream-thread-metrics,thread-id=<id> — process-rate, process-latency-avg, poll-latency-avg, commit-latency-avg, and similar.
• Task level: kafka.streams:type=stream-task-metrics,thread-id=<id>,task-id=<id> — per-task throughput and process-latency.
• State-store level: kafka.streams:type=stream-state-metrics,thread-id=<id>,task-id=<id>,<scope>-id=<storeName>, where <scope> is the store type, e.g. rocksdb-state, rocksdb-window-state, or in-memory-state. Store metrics default to the DEBUG recording level, so set metrics.recording.level=DEBUG to see them.

For state inspection, query the running store with KafkaStreams#store(...) — the same interactive-query mechanism you would use to serve reads to clients:

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

public class DebugEndpoint {
    private final KafkaStreams streams;

    public DebugEndpoint(KafkaStreams streams) {
        this.streams = streams;
    }

    public Long getCountForKey(String storeName, String key) {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            StoreQueryParameters.fromNameAndType(storeName, QueryableStoreTypes.keyValueStore())
        );
        return store.get(key);
    }
}

A local query only returns keys whose partitions are hosted on this instance. To find which instance owns a key across the cluster, use KafkaStreams#queryMetadataForKey(...); to enumerate all instances, use KafkaStreams#metadataForAllStreamsClients() (the replacement for the older allMetadata()). Note metadataForLocalThreads() is a different method — it returns local ThreadMetadata, not cluster-wide instance metadata.

Handling serialization and deserialization errors

A single malformed (“poison-pill”) record can otherwise kill a stream thread, because the default DeserializationExceptionHandler is LogAndFailExceptionHandler. In production you usually want LogAndContinueExceptionHandler, or a custom handler that routes the bad record to a dead-letter topic.

In Kafka 3.7 the handler signature takes a ProcessorContext from org.apache.kafka.streams.processor:

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.processor.ProcessorContext;
import java.util.Map;

public class DeadLetterDeserializationHandler implements DeserializationExceptionHandler {
    @Override
    public DeserializationHandlerResponse handle(ProcessorContext context,
                                                  ConsumerRecord<byte[], byte[]> record,
                                                  Exception exception) {
        System.err.printf("Deserialization error on %s-%d@%d: %s%n",
                record.topic(), record.partition(), record.offset(), exception.getMessage());
        // Produce to a dead-letter topic with your own producer if desired.
        return DeserializationHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(Map<String, ?> configs) {}
}

Register it:

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
          DeadLetterDeserializationHandler.class.getName());

Two forward-looking notes if you are on a newer release: Kafka 3.9 (KIP-1033) deprecates this ProcessorContext overload in favor of one taking ErrorHandlerContext, which exposes only safe metadata; and KIP-1034 adds a built-in dead-letter queue via errors.dead.letter.queue.topic.name, so you may not need to hand-roll the producer above. Symmetrically, errors thrown while producing are handled by a ProductionExceptionHandler registered with DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG. Together these handlers let a topology degrade gracefully on bad data instead of halting.

Validating exactly-once semantics

Exactly-once (EOS) makes offset commits and output writes a single Kafka transaction. Enable it with processing.guarantee:

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);   // internal topics
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1); // faster failover, optional

EXACTLY_ONCE_V2 (the constant value exactly_once_v2, from KIP-447) was introduced in Kafka 2.8 and made the recommended mode in 3.0. It requires brokers on 2.5+. The earlier EXACTLY_ONCE (v1) and the transitional EXACTLY_ONCE_BETA constants are deprecated — use v2. Setting replication.factor=3 matters because EOS writes commit markers to internal topics that must survive a broker loss; on a single-broker test cluster, use 1. See the Streams configuration reference for the full set of producer/consumer settings EOS turns on automatically (idempotent producer, read_committed consumers, and a shorter default commit.interval.ms).

TopologyTestDriver cannot validate EOS: it never opens a transaction or simulates a crash. Test EOS against an embedded broker or a staging cluster by killing the application mid-processing and restarting it, then asserting the output topic contains each expected record exactly once with no duplicates. Read the output with an isolation level of read_committed so aborted transactions are excluded. Watch transaction.timeout.ms: too short and you get spurious aborts under load; too long and a genuinely stuck transaction blocks read_committed consumers until it expires. Note that Streams manages transactional.id for you — do not set it manually.

Profiling and performance tuning

Performance problems usually show up as growing consumer lag, rising processing latency, or ballooning state-store disk usage. Start with end-to-end lag using kafka-consumer-groups.sh (the consumer group ID equals your application.id):

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-streams-app \
  --describe

Sustained lag means the topology cannot keep up. Common causes and fixes:

• Insufficient parallelism / skew. A topology scales to at most one task per input partition, so threads beyond that partition count sit idle, and a hot key concentrates load on one task. Increase input-topic partitions (this requires re-keying/reprocessing for keyed state) and size num.stream.threads to the partition count.
• Expensive per-record work. Synchronous external calls, heavy transforms, or large range scans dominate process-latency-avg. Move blocking I/O out of the processing path or batch it, and add custom metrics around the hot section.
• State-store / changelog contention. Slow disks make RocksDB flushes and changelog restores stall processing. Put the state.dir on fast local SSD, tune RocksDB via a rocksdb.config.setter, and increase commit.interval.ms (under EOS this also lengthens transaction windows) to amortize commits. To recover a corrupt or oversized local store, call KafkaStreams#cleanUp() before start() (or delete the state.dir); the store is rebuilt from its changelog.

A separate, offline operation is the application reset tool, kafka-streams-application-reset.sh. It does not clear local state stores — it resets the input-topic offsets and deletes the application’s internal (changelog and repartition) topics so you can reprocess from scratch. All instances must be stopped first, or you risk corrupting state; pair it with cleanUp() on each instance to wipe local state on the next start.

For CPU hotspots, attach a low-overhead profiler such as async-profiler to the running JVM (recent releases ship the asprof launcher; older ones use profiler.sh):

asprof -d 30 -f /tmp/streams-profile.html <PID>   # or: ./profiler.sh -d 30 -f ... <PID>

The resulting flame graph quickly surfaces whether time is spent in your processors, in serde code, or inside RocksDB. Cross-reference it with the thread-level process-latency-avg and poll-latency-avg metrics to separate processing cost from consumer-poll cost.

Conclusion

Reliable Kafka Streams operations come from layering the right tool at each level: TopologyTestDriver for fast, deterministic logic tests; an embedded broker for the rebalancing, serde, and transactional behavior the driver cannot reproduce; JMX metrics and interactive queries for live introspection; and explicit exception handlers so bad data degrades gracefully instead of halting a thread. Pin your expectations to the version you run — exactly-once mode, the exception-handler signatures, and the reset/cleanup semantics all changed across recent releases — and you can operate stateful stream processing in production with confidence.

Related Pages

• Stream Processing