Kafka Streams with Spring Boot: Real-Time Pipelines the Spring Way

In the Kafka Streams series, we built stream processing apps with plain Java: a main method, a Properties object, a shutdown hook, and a hand-rolled JSON serde. It works beautifully — and it's the right way to understand what's happening.

But most of us don't ship bare main methods. We ship Spring Boot apps. We want our config in application.yml, our lifecycle managed by the framework, our metrics in Micrometer, and our JSON handled by something we didn't write at 2 AM.

Good news: Spring Kafka wraps Kafka Streams without hiding it. The topology code stays exactly what you learned in the series. Spring just takes over everything around it.

In this post we build one complete, runnable app: an order analytics service. Orders arrive through a REST endpoint, flow through Kafka, and two real-time pipelines pop out the other side — running revenue per user, and fraud alerts when someone bursts 5+ orders in a minute.

If you're new to Kafka Streams concepts — KStreams vs KTables, windows, state stores — read Part 1 and Part 2 first. This post focuses on the Spring layer, not the concepts.

What Spring Actually Does For You

Everything below is stuff you'd otherwise write by hand:

Plain Kafka Streams	With Spring Boot
`Properties` object built in code	`application.yml`
`new KafkaStreams(...)`, start it yourself, register a shutdown hook	The app context starts and stops the streams with your app
Hand-rolled JSON serde	`JsonSerde` from Spring Kafka, wired to your `ObjectMapper`
`KafkaProducer` setup for test input	`KafkaTemplate`, auto-configured from yml
Metrics DIY	One listener wires Kafka Streams into Micrometer/Actuator
Ordering of shutdown, state cleanup on restart	Handled gracefully by the framework's lifecycle

One thing Spring does not change: the topology. StreamsBuilder, KStream, KTable, windows, state stores — identical to the series. That also means everything from Part 3: production concerns (serde traps, event time, exactly-once, error handling) applies unchanged.

Step 1: The Project

Generate a project at start.spring.io (Java 17+, Spring Boot 3.x) with the Spring for Apache Kafka dependency, then add kafka-streams — Spring's dependency management handles the version, and it must match the spring-kafka version, so resist the urge to pin your own:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- @EnableKafkaStreams needs this on the classpath -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
    </dependency>
    <!-- for testing, no broker needed -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams-test-utils</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

No version tags anywhere — deliberate. Spring Boot's BOM keepsspring-kafka and kafka-streams in lockstep. Mixing a hand-picked Kafka version with Spring's managed one is a classic source of NoSuchMethodError ghosts.

Step 2: Configuration in `application.yml`

This replaces the Properties object from Part 1:

spring:
  application:
    name: order-analytics
  kafka:
    bootstrap-servers: localhost:9092

    # Producer side (our REST endpoint uses this)
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

    # Streams side (our pipelines use this)
    streams:
      application-id: order-analytics-app
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        processing.guarantee: exactly_once_v2
        num.stream.threads: 2

Notice how the Part 3 production settings — exactly-once, thread count — are now just two lines of YAML. Also note application-id: if you leave it out, Spring falls back to spring.application.name. Same rule as ever, though: pick it deliberately and never change it on a running app.

The nested class name for the default serdes isSerdes$StringSerde — with the dollar sign. That's Java's inner-class naming, and it trips people up on the first day.

And the bootstrap class, with the one annotation that turns everything on:

@SpringBootApplication
@EnableKafkaStreams
public class OrderAnalyticsApplication {

    public static void main(String[] args) {
        SpringApplication.run(OrderAnalyticsApplication.class, args);
    }
}

@EnableKafkaStreams makes Spring auto-configure a StreamsBuilder bean and a factory that starts your topology when the context starts — and stops it cleanly when it shuts down. No KafkaStreams object, no shutdown hook, no lifecycle code.

Step 3: The Event and the REST Endpoint

One record, one controller. A KafkaTemplate is auto-configured from the producer block in the yml:

public record Order(
        String orderId,
        String userId,
        double amount,
        Instant timestamp
) {}

@RestController
@RequestMapping("/orders")
public class OrderController {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderController(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @PostMapping
    public ResponseEntity<Void> placeOrder(@RequestBody Order order) {
        // key = userId → all of a user's orders land in the same partition, in order
        kafkaTemplate.send("orders", order.userId(), order);
        return ResponseEntity.accepted().build();
    }
}

Two details worth a second look:

The key is userId. Same key → same partition → per-user order, the foundation everything downstream relies on (Part 1's golden rule).
send() is async. We return 202 Accepted — "taken, processing" — not "processed". That honesty is the whole point of event-driven endpoints.

The JsonSerializer from the yml turns each Order into JSON bytes automatically. Done — we have an event producer in ~20 lines.

Step 4: The Topology

Here's the Spring-wrapped version of everything from the series. Spring Kafka ships its own JsonSerde (backed by the same serializer family as the producer side), so we skip the hand-rolled one from Part 3:

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.serializer.JsonSerde;

import java.time.Duration;

@Configuration
public class OrderTopology {

    public static final String ORDERS = "orders";
    public static final String USER_REVENUE = "user-revenue";
    public static final String FRAUD_ALERTS = "fraud-alerts";

    public static final JsonSerde<Order> ORDER_SERDE = new JsonSerde<>(Order.class);

    // Spring injects the auto-configured StreamsBuilder here.
    // When this configuration loads, the pipeline is registered — and
    // @EnableKafkaStreams starts it with the application.
    public OrderTopology(StreamsBuilder builder) {
        definePipeline(builder);
    }

    public static void definePipeline(StreamsBuilder builder) {

        KStream<String, Order> orders =
                builder.stream(ORDERS, Consumed.with(Serdes.String(), ORDER_SERDE));

        buildRevenuePerUser(orders);
        buildFraudAlerts(orders);
    }

    // Pipeline 1: running total revenue per user → "user-revenue"
    private static void buildRevenuePerUser(KStream<String, Order> orders) {
        orders
                .groupByKey()
                .aggregate(
                        () -> 0.0,
                        (userId, order, total) -> total + order.amount(),
                        Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("revenue-by-user")
                                .withValueSerde(Serdes.Double()))
                .toStream()
                .to(USER_REVENUE, Produced.with(Serdes.String(), Serdes.Double()));
    }

    // Pipeline 2: 5+ orders inside any 1-minute window → one final alert
    private static void buildFraudAlerts(KStream<String, Order> orders) {
        orders
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeAndGrace(
                        Duration.ofMinutes(1), Duration.ofSeconds(10)))
                .count()
                .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
                .toStream()
                .filter((userId, count) -> count >= 5)
                .map((windowedUserId, count) -> KeyValue.pair(
                        windowedUserId.key(),          // unwrap Windowed<String> → String
                        "Suspicious burst: " + count + " orders in 1 minute"))
                .to(FRAUD_ALERTS, Produced.with(Serdes.String(), Serdes.String()));
    }
}

If you read the series, every line should feel familiar — this is Part 2's aggregate and Part 2's windowed count + suppress, nearly verbatim. The only Spring-specific moves are:

The constructor injection of StreamsBuilder — Spring hands us the builder it owns, so it can manage the resulting topology's lifecycle. (The Spring Kafka docs also show a @Bean returning a KStream; same effect, pick whichever reads better to you.)
JsonSerde<Order> — Spring's serde instead of our hand-rolled Jackson one.
definePipeline is a plain static method — deliberately, so we can test the topology without booting Spring. You'll see why in a minute.

The Part 3 trap still applies here. aggregate stores Double values in a state store — without .withValueSerde(Serdes.Double()), the default String serde tries to serialize a Double and the app dies at runtime. Spring manages the lifecycle; it does not rescue you from serde mistakes. That's a feature: the failure modes are the ones you already know.

Two honest notes on the fraud pipeline, straight from the series:

Why suppress? Without it, the fraud-alerts topic gets an update per order (1, 2, 3, 4, 5...). With it, each window emits exactly one final record after it closes — the alert fires once, when the verdict is in.
The alert waits for the window to close. A burst at 10:00:00 produces its alert roughly at 10:01:10 (window end + grace). That's correct-by-design for "final answer" semantics; if you need faster reactions, shrink the window or drop suppress and react to intermediate counts.

Step 5: Run the Whole Thing

Start Kafka and create the topics (same one-liner from Part 1):

docker run -d --name kafka -p 9092:9092 apache/kafka:4.0.0

for topic in orders user-revenue fraud-alerts; do
  docker exec kafka /opt/kafka/bin/kafka-topics.sh \
    --bootstrap-server localhost:9092 --create --topic $topic --partitions 3
done

Start the app, fire 5 quick orders at one user:

for i in {1..5}; do
  curl -X POST http://localhost:8080/orders \
    -H "Content-Type: application/json" \
    -d "{\"orderId\":\"o-$i\",\"userId\":\"user-1\",\"amount\":25.50,\"timestamp\":\"2026-08-25T10:00:00Z\"}"
done

Watch revenue update live:

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic user-revenue \
  --from-beginning --property print.key=true --property key.separator=" -> "

user-1 -> 25.5
user-1 -> 51.0
user-1 -> 76.5
user-1 -> 102.0
user-1 -> 127.5

Then watch the fraud verdict arrive — about 70 seconds after the burst, when the window closes:

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic fraud-alerts --from-beginning

Suspicious burst: 5 orders in 1 minute

A REST endpoint, real-time aggregation, and windowed fraud detection — one runnable Spring Boot app.

Step 6: Testing Without a Broker

Because definePipeline is a plain static method, we can test the whole topology with TopologyTestDriver (Part 3's approach) — no Kafka, no Docker, no Spring context, milliseconds per test:

class OrderTopologyTest {

    @Test
    void flagsSuspiciousBursts() {
        StreamsBuilder builder = new StreamsBuilder();
        OrderTopology.definePipeline(builder);

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "nowhere:9092"); // never contacted

        try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {

            TestInputTopic<String, Order> orders = driver.createInputTopic(
                    OrderTopology.ORDERS,
                    new StringSerializer(), new JsonSerializer<Order>());

            TestOutputTopic<String, String> alerts = driver.createOutputTopic(
                    OrderTopology.FRAUD_ALERTS,
                    new StringDeserializer(), new StringDeserializer());

            // 5 orders, 10 seconds apart, starting at 10:00 — all inside one window
            List<KeyValue<String, Order>> burst = new ArrayList<>();
            Instant ts = Instant.parse("2026-08-25T10:00:00Z");
            for (int i = 1; i <= 5; i++) {
                burst.add(new KeyValue<>("user-1", new Order("o-" + i, "user-1", 25.0, ts)));
                ts = ts.plusSeconds(10);
            }
            orders.pipeKeyValueList(burst,
                    Instant.parse("2026-08-25T10:00:00Z"), Duration.ofSeconds(10));

            // Window hasn't closed yet → suppress holds the result back
            assertThat(alerts.isEmpty()).isTrue();

            // Fast-forward past window end + grace — no real waiting!
            driver.advanceWallClockTime(Duration.ofMinutes(2));

            assertThat(alerts.readValue()).contains("Suspicious burst: 5");
        }
    }
}

That last trick is the payoff of testable topology design: advanceWallClockTime jumps straight past the window close. Testing "did the alert fire after 70 seconds" takes 70 microseconds of test time, not 70 seconds of wall clock.

Production Tuning: It's All YAML Now

Every Part 3 setting maps to one or two lines in the same config block. A production-ready streams section looks like this:

spring:
  kafka:
    streams:
      application-id: order-analytics-app
      properties:
        # Parallelism: tasks = input partitions, spread across these threads
        num.stream.threads: 3

        # Exactly-once (skip it if duplicates are harmless — it costs throughput)
        processing.guarantee: exactly_once_v2

        # Warm failover for big state stores (Part 3)
        num.standby.replicas: 1

        # Local state location (defaults to a temp dir — pin it in prod)
        state.dir: /var/lib/order-analytics/state

        # Poison-pill policy from Part 3: skip and log instead of dying
        default.deserialization.exception.handler: com.example.analytics.config.LogAndSkipHandler

And for observability, one bean wires Kafka Streams' metrics into Micrometer (add the actuator starter to expose them):

@Bean
StreamsBuilderFactoryBeanConfigurer metricsConfigurer(MeterRegistry registry) {
    return factory -> factory.addKafkaStreamsMicrometerListener(
            new KafkaStreamsMicrometerListener(registry));
}

From there, consumer lag, process rate, and state store metrics show up in actuator endpoints or your dashboard like any other Spring metric — no custom exporters.

Where to Go From Here

You've now seen Kafka Streams from two angles: raw (the series) and Spring-flavored (this post). The topology code was identical — that's the strongest argument for learning the core library first and the integration second.

If you want to keep going:

Concepts you used today, explained properly: KStreams, KTables, joins and windows
Everything production: serdes, event time, exactly-once, DLQs, scaling, testing — Part 3
Going deeper with Spring Kafka: the Spring Kafka reference documentation covers @KafkaListener, error handlers, and transactions (the messaging side, as opposed to the streams side)

The natural next step is the same one from Part 3: take one real topic from your world, wrap it in this exact skeleton — @EnableKafkaStreams, one @Configuration topology, tests with TopologyTestDriver — and ship something small.

Key Takeaways:

Spring Kafka wraps Kafka Streams without changing it — the topology code is identical to plain Kafka Streams

@EnableKafkaStreams gives you an auto-configured StreamsBuilder and manages the entire lifecycle (start with the app, clean stop, graceful shutdown)

All config — exactly-once, threads, standbys, exception handlers — lives in spring.kafka.streams.* in application.yml

Use Spring's JsonSerde instead of hand-rolled serdes, and let Boot's BOM manage versions — never pin Kafka versions yourself

Keep the topology in a plain static method so TopologyTestDriver tests run without a broker or a Spring context

advanceWallClockTime lets you test windowed/suppressed pipelines instantly — no sleeping in tests

Key your events deliberately (userId here): same key → same partition is what makes grouping, aggregating, and windowing correct

Good news: Spring Kafka wraps Kafka Streams without hiding it. The topology code stays exactly what you learned in the series. Spring just takes over everything around it.

If you're new to Kafka Streams concepts — KStreams vs KTables, windows, state stores — read Part 1 and Part 2 first. This post focuses on the Spring layer, not the concepts.

What Spring Actually Does For You

Everything below is stuff you'd otherwise write by hand:

Plain Kafka Streams	With Spring Boot
`Properties` object built in code	`application.yml`
`new KafkaStreams(...)`, start it yourself, register a shutdown hook	The app context starts and stops the streams with your app
Hand-rolled JSON serde	`JsonSerde` from Spring Kafka, wired to your `ObjectMapper`
`KafkaProducer` setup for test input	`KafkaTemplate`, auto-configured from yml
Metrics DIY	One listener wires Kafka Streams into Micrometer/Actuator
Ordering of shutdown, state cleanup on restart	Handled gracefully by the framework's lifecycle

Step 1: The Project

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- @EnableKafkaStreams needs this on the classpath -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
    </dependency>
    <!-- for testing, no broker needed -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams-test-utils</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Step 2: Configuration in `application.yml`

This replaces the Properties object from Part 1:

spring:
  application:
    name: order-analytics
  kafka:
    bootstrap-servers: localhost:9092

    # Producer side (our REST endpoint uses this)
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

    # Streams side (our pipelines use this)
    streams:
      application-id: order-analytics-app
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        processing.guarantee: exactly_once_v2
        num.stream.threads: 2

The nested class name for the default serdes isSerdes$StringSerde — with the dollar sign. That's Java's inner-class naming, and it trips people up on the first day.

And the bootstrap class, with the one annotation that turns everything on:

@SpringBootApplication
@EnableKafkaStreams
public class OrderAnalyticsApplication {

    public static void main(String[] args) {
        SpringApplication.run(OrderAnalyticsApplication.class, args);
    }
}

Step 3: The Event and the REST Endpoint

One record, one controller. A KafkaTemplate is auto-configured from the producer block in the yml:

public record Order(
        String orderId,
        String userId,
        double amount,
        Instant timestamp
) {}

@RestController
@RequestMapping("/orders")
public class OrderController {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderController(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @PostMapping
    public ResponseEntity<Void> placeOrder(@RequestBody Order order) {
        // key = userId → all of a user's orders land in the same partition, in order
        kafkaTemplate.send("orders", order.userId(), order);
        return ResponseEntity.accepted().build();
    }
}

Two details worth a second look:

The key is userId. Same key → same partition → per-user order, the foundation everything downstream relies on (Part 1's golden rule).
send() is async. We return 202 Accepted — "taken, processing" — not "processed". That honesty is the whole point of event-driven endpoints.

The JsonSerializer from the yml turns each Order into JSON bytes automatically. Done — we have an event producer in ~20 lines.

Step 4: The Topology

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.serializer.JsonSerde;

import java.time.Duration;

@Configuration
public class OrderTopology {

    public static final String ORDERS = "orders";
    public static final String USER_REVENUE = "user-revenue";
    public static final String FRAUD_ALERTS = "fraud-alerts";

    public static final JsonSerde<Order> ORDER_SERDE = new JsonSerde<>(Order.class);

    // Spring injects the auto-configured StreamsBuilder here.
    // When this configuration loads, the pipeline is registered — and
    // @EnableKafkaStreams starts it with the application.
    public OrderTopology(StreamsBuilder builder) {
        definePipeline(builder);
    }

    public static void definePipeline(StreamsBuilder builder) {

        KStream<String, Order> orders =
                builder.stream(ORDERS, Consumed.with(Serdes.String(), ORDER_SERDE));

        buildRevenuePerUser(orders);
        buildFraudAlerts(orders);
    }

    // Pipeline 1: running total revenue per user → "user-revenue"
    private static void buildRevenuePerUser(KStream<String, Order> orders) {
        orders
                .groupByKey()
                .aggregate(
                        () -> 0.0,
                        (userId, order, total) -> total + order.amount(),
                        Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("revenue-by-user")
                                .withValueSerde(Serdes.Double()))
                .toStream()
                .to(USER_REVENUE, Produced.with(Serdes.String(), Serdes.Double()));
    }

    // Pipeline 2: 5+ orders inside any 1-minute window → one final alert
    private static void buildFraudAlerts(KStream<String, Order> orders) {
        orders
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeAndGrace(
                        Duration.ofMinutes(1), Duration.ofSeconds(10)))
                .count()
                .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
                .toStream()
                .filter((userId, count) -> count >= 5)
                .map((windowedUserId, count) -> KeyValue.pair(
                        windowedUserId.key(),          // unwrap Windowed<String> → String
                        "Suspicious burst: " + count + " orders in 1 minute"))
                .to(FRAUD_ALERTS, Produced.with(Serdes.String(), Serdes.String()));
    }
}

If you read the series, every line should feel familiar — this is Part 2's aggregate and Part 2's windowed count + suppress, nearly verbatim. The only Spring-specific moves are:

The constructor injection of StreamsBuilder — Spring hands us the builder it owns, so it can manage the resulting topology's lifecycle. (The Spring Kafka docs also show a @Bean returning a KStream; same effect, pick whichever reads better to you.)
JsonSerde<Order> — Spring's serde instead of our hand-rolled Jackson one.
definePipeline is a plain static method — deliberately, so we can test the topology without booting Spring. You'll see why in a minute.

Two honest notes on the fraud pipeline, straight from the series:

Why suppress? Without it, the fraud-alerts topic gets an update per order (1, 2, 3, 4, 5...). With it, each window emits exactly one final record after it closes — the alert fires once, when the verdict is in.
The alert waits for the window to close. A burst at 10:00:00 produces its alert roughly at 10:01:10 (window end + grace). That's correct-by-design for "final answer" semantics; if you need faster reactions, shrink the window or drop suppress and react to intermediate counts.

Step 5: Run the Whole Thing

Start Kafka and create the topics (same one-liner from Part 1):

docker run -d --name kafka -p 9092:9092 apache/kafka:4.0.0

for topic in orders user-revenue fraud-alerts; do
  docker exec kafka /opt/kafka/bin/kafka-topics.sh \
    --bootstrap-server localhost:9092 --create --topic $topic --partitions 3
done

Start the app, fire 5 quick orders at one user:

for i in {1..5}; do
  curl -X POST http://localhost:8080/orders \
    -H "Content-Type: application/json" \
    -d "{\"orderId\":\"o-$i\",\"userId\":\"user-1\",\"amount\":25.50,\"timestamp\":\"2026-08-25T10:00:00Z\"}"
done

Watch revenue update live:

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic user-revenue \
  --from-beginning --property print.key=true --property key.separator=" -> "

user-1 -> 25.5
user-1 -> 51.0
user-1 -> 76.5
user-1 -> 102.0
user-1 -> 127.5

Then watch the fraud verdict arrive — about 70 seconds after the burst, when the window closes:

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic fraud-alerts --from-beginning

Suspicious burst: 5 orders in 1 minute

A REST endpoint, real-time aggregation, and windowed fraud detection — one runnable Spring Boot app.

Step 6: Testing Without a Broker

Because definePipeline is a plain static method, we can test the whole topology with TopologyTestDriver (Part 3's approach) — no Kafka, no Docker, no Spring context, milliseconds per test:

class OrderTopologyTest {

    @Test
    void flagsSuspiciousBursts() {
        StreamsBuilder builder = new StreamsBuilder();
        OrderTopology.definePipeline(builder);

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "nowhere:9092"); // never contacted

        try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {

            TestInputTopic<String, Order> orders = driver.createInputTopic(
                    OrderTopology.ORDERS,
                    new StringSerializer(), new JsonSerializer<Order>());

            TestOutputTopic<String, String> alerts = driver.createOutputTopic(
                    OrderTopology.FRAUD_ALERTS,
                    new StringDeserializer(), new StringDeserializer());

            // 5 orders, 10 seconds apart, starting at 10:00 — all inside one window
            List<KeyValue<String, Order>> burst = new ArrayList<>();
            Instant ts = Instant.parse("2026-08-25T10:00:00Z");
            for (int i = 1; i <= 5; i++) {
                burst.add(new KeyValue<>("user-1", new Order("o-" + i, "user-1", 25.0, ts)));
                ts = ts.plusSeconds(10);
            }
            orders.pipeKeyValueList(burst,
                    Instant.parse("2026-08-25T10:00:00Z"), Duration.ofSeconds(10));

            // Window hasn't closed yet → suppress holds the result back
            assertThat(alerts.isEmpty()).isTrue();

            // Fast-forward past window end + grace — no real waiting!
            driver.advanceWallClockTime(Duration.ofMinutes(2));

            assertThat(alerts.readValue()).contains("Suspicious burst: 5");
        }
    }
}

Production Tuning: It's All YAML Now

Every Part 3 setting maps to one or two lines in the same config block. A production-ready streams section looks like this:

spring:
  kafka:
    streams:
      application-id: order-analytics-app
      properties:
        # Parallelism: tasks = input partitions, spread across these threads
        num.stream.threads: 3

        # Exactly-once (skip it if duplicates are harmless — it costs throughput)
        processing.guarantee: exactly_once_v2

        # Warm failover for big state stores (Part 3)
        num.standby.replicas: 1

        # Local state location (defaults to a temp dir — pin it in prod)
        state.dir: /var/lib/order-analytics/state

        # Poison-pill policy from Part 3: skip and log instead of dying
        default.deserialization.exception.handler: com.example.analytics.config.LogAndSkipHandler

And for observability, one bean wires Kafka Streams' metrics into Micrometer (add the actuator starter to expose them):

@Bean
StreamsBuilderFactoryBeanConfigurer metricsConfigurer(MeterRegistry registry) {
    return factory -> factory.addKafkaStreamsMicrometerListener(
            new KafkaStreamsMicrometerListener(registry));
}

From there, consumer lag, process rate, and state store metrics show up in actuator endpoints or your dashboard like any other Spring metric — no custom exporters.

Where to Go From Here

If you want to keep going:

Concepts you used today, explained properly: KStreams, KTables, joins and windows
Everything production: serdes, event time, exactly-once, DLQs, scaling, testing — Part 3
Going deeper with Spring Kafka: the Spring Kafka reference documentation covers @KafkaListener, error handlers, and transactions (the messaging side, as opposed to the streams side)

Key Takeaways:

Spring Kafka wraps Kafka Streams without changing it — the topology code is identical to plain Kafka Streams

@EnableKafkaStreams gives you an auto-configured StreamsBuilder and manages the entire lifecycle (start with the app, clean stop, graceful shutdown)

All config — exactly-once, threads, standbys, exception handlers — lives in spring.kafka.streams.* in application.yml

Use Spring's JsonSerde instead of hand-rolled serdes, and let Boot's BOM manage versions — never pin Kafka versions yourself

Keep the topology in a plain static method so TopologyTestDriver tests run without a broker or a Spring context

advanceWallClockTime lets you test windowed/suppressed pipelines instantly — no sleeping in tests

Key your events deliberately (userId here): same key → same partition is what makes grouping, aggregating, and windowing correct

Kafka Streams with Spring Boot: Real-Time Pipelines the Spring Way

What Spring Actually Does For You

Step 1: The Project

Step 2: Configuration in `application.yml`

Step 3: The Event and the REST Endpoint

Step 4: The Topology

Step 5: Run the Whole Thing

Step 6: Testing Without a Broker

Production Tuning: It's All YAML Now

Where to Go From Here

Tags

Related Posts

Kafka Streams Part 3: Making It Production-Ready

Kafka Streams Part 1: Stream Processing Made Simple

Kafka Streams Part 2: KStreams, KTables, Joins and Windows

Kafka Streams with Spring Boot: Real-Time Pipelines the Spring Way

What Spring Actually Does For You

Step 1: The Project

Step 2: Configuration in `application.yml`

Step 3: The Event and the REST Endpoint

Step 4: The Topology

Step 5: Run the Whole Thing

Step 6: Testing Without a Broker

Production Tuning: It's All YAML Now

Where to Go From Here

Tags

Related Posts

Kafka Streams Part 3: Making It Production-Ready

Kafka Streams Part 1: Stream Processing Made Simple

Kafka Streams Part 2: KStreams, KTables, Joins and Windows

What Spring Actually Does For You

Step 1: The Project

Step 2: Configuration in application.yml

Step 3: The Event and the REST Endpoint

Step 4: The Topology

Step 5: Run the Whole Thing

Step 6: Testing Without a Broker

Production Tuning: It's All YAML Now

Where to Go From Here

Tags

Related Posts

Kafka Streams Part 3: Making It Production-Ready

Kafka Streams Part 1: Stream Processing Made Simple

Kafka Streams Part 2: KStreams, KTables, Joins and Windows

What Spring Actually Does For You

Step 1: The Project

Step 2: Configuration in application.yml

Step 3: The Event and the REST Endpoint

Step 4: The Topology

Step 5: Run the Whole Thing

Step 6: Testing Without a Broker

Production Tuning: It's All YAML Now

Where to Go From Here

Tags

Related Posts

Kafka Streams Part 3: Making It Production-Ready

Kafka Streams Part 1: Stream Processing Made Simple

Kafka Streams Part 2: KStreams, KTables, Joins and Windows

Step 2: Configuration in `application.yml`

Step 2: Configuration in `application.yml`