Skip to content
JavaAgentic

Type at least two characters. Try “RAG”, “pgvector” or “tool calling”.

Kafka Streams & Stream Processing

Stream processing without a cluster: KStream and KTable semantics, stateless and stateful operations, windowing, joins, exactly-once v2 and testing with TopologyTestDriver.

Advanced5 min readUpdated
On this page

Kafka Streams is a library, not a cluster. Your application reads from topics, transforms, and writes back — with state, windowing and fault tolerance handled by Kafka itself. No Flink, no Spark, no extra infrastructure.

Key Takeaways

  • KStream is a sequence of facts; KTable is the current value per key.
  • Stateless operations are cheap; stateful ones create local stores backed by changelog topics.
  • Parallelism comes from partitions, exactly as with a consumer group.
  • exactly_once_v2 covers Kafka-to-Kafka only, not external writes.
  • TopologyTestDriver tests a whole topology with no broker running.

Streams and tables

The same records mean different things depending on interpretation. Choosing wrongly produces answers that look plausible and are wrong.

A GlobalKTable is a third option: fully replicated to every instance rather than partitioned. That makes it joinable on any key without co-partitioning, which is ideal for small reference data — currency rates, country lookups, product categories. It is unsuitable for anything large, since every instance holds the whole thing.

A topology

OrderStreamsTopology.java
@Configuration
@EnableKafkaStreams
public class OrderStreamsTopology {
 
    @Bean
    public KStream<String, OrderEvent> orderStream(StreamsBuilder builder) {
 
        KStream<String, OrderEvent> orders =
                builder.stream("orders", Consumed.with(Serdes.String(), orderSerde()));
 
        // Small, slow-changing reference data: replicate it everywhere so the
        // join needs no co-partitioning.
        GlobalKTable<String, Customer> customers =
                builder.globalTable("customers", Consumed.with(Serdes.String(), customerSerde()));
 
        orders
            .filter((key, order) -> order.status() == OrderStatus.PLACED)
            .join(customers,
                  (orderId, order) -> order.customerId(),      // key extractor
                  (order, customer) -> order.enrichedWith(customer))
            .peek((k, v) -> log.debug("enriched order {}", k))
            .to("orders-enriched", Produced.with(Serdes.String(), enrichedSerde()));
 
        // Windowed revenue per channel, emitted once per closed window.
        orders
            .filter((key, order) -> order.status() == OrderStatus.PAID)
            .groupBy((key, order) -> order.channel(),
                     Grouped.with(Serdes.String(), orderSerde()))
            .windowedBy(TimeWindows.ofSizeAndGrace(
                    Duration.ofMinutes(5), Duration.ofMinutes(1)))   // grace for late arrivals
            .aggregate(
                    () -> 0L,
                    (channel, order, total) -> total + order.totalMinorUnits(),
                    Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("revenue-store")
                            .withValueSerde(Serdes.Long()))
            // Emit only the final result per window instead of every update.
            .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
            .toStream((windowedKey, total) -> windowedKey.key())
            .to("revenue-by-channel", Produced.with(Serdes.String(), Serdes.Long()));
 
        return orders;
    }
}

suppress is worth highlighting. Without it, an aggregation emits an updated result on every input record, so a five-minute window with a thousand orders produces a thousand downstream messages instead of one. That is usually not what a consumer expects, and it is a common source of surprise volume.

Windowing

WindowShapeUse for
TumblingFixed size, no overlapHourly totals, daily reports
HoppingFixed size, overlappingRolling 5-minute average, updated every minute
SessionGap-based, variable lengthUser activity sessions
SlidingOnly for joinsCorrelating two streams within a time bound

Grace periods handle late arrivals. ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)) keeps a window open for one extra minute after it closes, so a record delayed by network or rebalance still lands in the right bucket. Longer grace means more accuracy and more memory; the right value comes from measuring your actual arrival delay distribution rather than guessing.

Joins and co-partitioning

A KStream-KTable or KStream-KStream join requires co-partitioning: both topics must have the same number of partitions and the same partitioning strategy, so records with the same key land on the same instance. Kafka Streams will fail at startup rather than silently produce wrong results, which is a mercy — but it means a join sometimes requires repartitioning one side first with selectKey followed by through.

GlobalKTable sidesteps this entirely, which is why it is the right choice for small lookup data even when the partition counts happen to match today.

Exactly-once

application.yml
spring:
  kafka:
    streams:
      application-id: order-enrichment          # also the consumer group id
      properties:
        processing.guarantee: exactly_once_v2
        num.standby.replicas: 1                 # warm failover for state stores
        state.dir: /var/lib/kafka-streams       # MUST survive restarts
        commit.interval.ms: 100
        # Larger cache means fewer downstream updates, at the cost of latency.
        statestore.cache.max.bytes: 10485760

exactly_once_v2 wraps consume, state update and produce in one Kafka transaction, so a failure mid-processing rolls all three back. It is genuinely exactly-once — within Kafka.

The boundary matters. A database write inside a processor is not in that transaction. If the transaction aborts after the database write succeeded, the write stands while the Kafka side is rolled back and reprocessed. Keep external side effects out of stream processors where you can, and make them idempotent where you cannot.

state.dir on a persistent volume is the other setting people miss. Without it, every restart rebuilds every state store by replaying its changelog topic from the beginning — minutes of downtime for a large store, on every deploy.

Interactive queries

State stores can be queried directly, turning a stream processor into a low-latency read API:

InteractiveQuery.java
@RestController
public class RevenueController {
 
    private final StreamsBuilderFactoryBean factory;
 
    @GetMapping("/api/revenue/{channel}")
    public ResponseEntity<Long> revenue(@PathVariable String channel) {
        KafkaStreams streams = factory.getKafkaStreams();
 
        // The key may live on another instance — find out which.
        KeyQueryMetadata metadata = streams.queryMetadataForKey(
                "revenue-store", channel, Serdes.String().serializer());
 
        if (!metadata.activeHost().equals(thisHost)) {
            // Forward to the owning instance rather than returning nothing.
            return remote.fetch(metadata.activeHost(), channel);
        }
 
        var store = streams.store(StoreQueryParameters.fromNameAndType(
                "revenue-store", QueryableStoreTypes.keyValueStore()));
        return ResponseEntity.ok((Long) store.get(channel));
    }
}

The forwarding step is what makes this usable. State is partitioned across instances, so any instance may be asked for a key it does not hold; without an RPC layer between instances, a third of your queries return nothing.

Testing

TopologyTest.java
@Test
void aggregatesRevenuePerChannel() {
    var props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
 
    try (var driver = new TopologyTestDriver(buildTopology(), props)) {
        var input = driver.createInputTopic("orders",
                Serdes.String().serializer(), orderSerde().serializer());
        var output = driver.createOutputTopic("revenue-by-channel",
                Serdes.String().deserializer(), Serdes.Long().deserializer());
 
        input.pipeInput("ORD-1", paidOrder("web", 1000));
        input.pipeInput("ORD-2", paidOrder("web", 2500));
        // Advance past the window plus grace so suppression emits.
        driver.advanceWallClockTime(Duration.ofMinutes(6));
 
        assertThat(output.readKeyValuesToMap()).containsEntry("web", 3500L);
    }
}

TopologyTestDriver runs the entire topology in-process with no broker, no containers and no waiting. Being able to advance wall-clock time deterministically is what makes windowing and suppression testable at all.

What to take away

Model your data as a stream of facts or a table of current values, and be deliberate about which. Use suppress so windowed aggregations emit once rather than continuously. Give state stores a persistent directory and a standby replica. Remember exactly-once stops at the Kafka boundary, and test topologies with TopologyTestDriver rather than a cluster.

Frequently Asked Questions

KStream or KTable?
A KStream is a sequence of independent facts — each record is an event that happened. A KTable is a changelog where each key has a current value and a new record replaces the old one. Ask whether a second record for the same key adds to history or supersedes it.
Why is my state store rebuilding on every restart?
Without a state directory that survives restarts, the store is rebuilt by replaying its changelog topic, which is slow for large stores. Mount a persistent volume for state.dir, and set num.standby.replicas so another instance already holds a warm copy for failover.
Does exactly_once_v2 make everything exactly-once?
It makes read-process-write within Kafka atomic — consuming, updating state and producing commit together. It does not extend to external systems. A write to a database inside a stream processor is outside the transaction and still needs idempotency.

Related tutorials