60-Day Kafka 4 Learning Plan · Week 5 — Day 35 of 60

---

60-Day Kafka 4 Learning Plan · Week 5 — Kafka Streams (capstone)
Sources: Kafka: The Definitive Guide Ch.11 · kafka.apache.org/documentation/streams/tutorial

Goal

Combine all Week 5 concepts into one working Spring Boot project. In this story, we build a complete real-time order analytics pipeline. Raw order events arrive on labs.events.v2 (Avro-encoded, from Phase 2 of this series). 

The topology filters confirmed orders, enriches them with user profile data from a GlobalKTable, aggregates per-user statistics into a RocksDB state store, and exposes the results through a REST endpoint using Kafka Streams Interactive Queries.

What we’re building

The pipeline has seven steps:

labs.events.v2   (Avro LabEvent)
  │
  │  filter: type == "ORDER_CONFIRMED"
  ▼
  │  selectKey: payload.userId     ← repartition trigger
  ▼
  │  join: users GlobalKTable      ← no co-partitioning required
  ▼
  │  aggregate: OrderStats         ← RocksDB state store
  ▼
labs.order-stats (JSON OrderStats)
  │
  │  GET /api/stats/{userId}       ← Interactive Queries
  ▼

Each step introduces a distinct Kafka Streams concept. Together they cover the full stateful stream processing model.

Let’s code

Project structure

kafka-order-analytics-demo/
├── docker-compose.yml
├── Dockerfile
├── pom.xml
└── src/
    ├── main/
    │   ├── avro/
    │   │   └── lab-event.avsc
    │   ├── java/com/boottechsolutions/orderanalytics/
    │   │   ├── OrderAnalyticsApplication.java
    │   │   ├── config/
    │   │   │   ├── KafkaStreamsConfig.java
    │   │   │   ├── TopicConfig.java
    │   │   │   └── DataSeeder.java
    │   │   ├── domain/
    │   │   │   ├── OrderPayload.java
    │   │   │   ├── UserProfile.java
    │   │   │   ├── EnrichedOrder.java
    │   │   │   └── OrderStats.java
    │   │   ├── serde/
    │   │   │   └── JsonSerde.java
    │   │   ├── streams/
    │   │   │   └── OrderAnalyticsTopology.java
    │   │   └── controller/
    │   │       └── StatsController.java
    │   └── resources/
    │       └── application.yml
    └── test/
        └── java/com/boottechsolutions/orderanalytics/
            └── OrderAnalyticsTopologyTest.java

Step 1: Dependencies and configuration

pom.xml

Key dependencies:

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.5.15</version>
</parent>

<properties>
    <java.version>21</java.version>
    <avro.version>1.11.3</avro.version>
    <confluent.version>7.6.0</confluent.version>
</properties>

<repositories>
    <repository>
        <id>confluent</id>
        <url>https://packages.confluent.io/maven/</url>
    </repository>
</repositories>

<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>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>${avro.version}</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>${confluent.version}</version>
        <exclusions>
            <!-- Let Spring Boot BOM control kafka-clients version -->
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-streams-avro-serde</artifactId>
        <version>${confluent.version}</version>
        <exclusions>
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-streams</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.springframework.retry</groupId>
        <artifactId>spring-retry</artifactId>
    </dependency>
    <!-- test -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams-test-utils</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-maven-plugin</artifactId>
            <version>${avro.version}</version>
            <executions>
                <execution>
                    <phase>generate-sources</phase>
                    <goals><goal>schema</goal></goals>
                    <configuration>
                        <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                        <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
                        <stringType>String</stringType>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

The kafka-clients exclusions prevent a dependency version conflict: Confluent 7.6.0 pulls in Kafka 3.6.x clients while Spring Boot 3.5.x manages Kafka 3.9.x. Excluding from Confluent lets Spring Boot’s BOM control the version — the serializers remain binary-compatible.

application.yml

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    streams:
      application-id: order-analytics-app
      properties:
        schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        num.stream.threads: 2
        commit.interval.ms: 1000
        state.dir: /tmp/kafka-streams/order-analytics
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}

server:
  port: ${SERVER_PORT:8090}

analytics:
  topics:
    events: labs.events.v2
    users: labs.users
    order-stats: labs.order-stats
  stores:
    order-stats: order-stats-store

num.stream.threads: 2 means the topology runs on two threads within the same JVM. Partitions are distributed across threads. For this demo with 3 partitions on labs.events.v2, two threads process one or two partitions each.

Step 2: Kafka Streams configuration

@EnableKafkaStreams activates Spring Kafka’s Kafka Streams integration. It looks for a bean named DEFAULT_STREAMS_CONFIG_BEAN_NAME of type KafkaStreamsConfiguration:

@SpringBootApplication
@EnableKafkaStreams
@EnableRetry
@EnableAsync
public class OrderAnalyticsApplication {
    public static void main(String[] args) {
        SpringApplication.run(OrderAnalyticsApplication.class, args);
    }
}

@Configuration
public class KafkaStreamsConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.streams.application-id}")
    private String applicationId;

    @Value("${spring.kafka.streams.properties.schema.registry.url}")
    private String schemaRegistryUrl;

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
                org.apache.kafka.common.serialization.Serdes.StringSerde.class);
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
                org.apache.kafka.common.serialization.Serdes.StringSerde.class);
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
        props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams/order-analytics");
        props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                LogAndContinueExceptionHandler.class);
        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public KafkaStreamsInteractiveQueryService kafkaStreamsInteractiveQueryService(
            StreamsBuilderFactoryBean streamsBuilderFactoryBean) {
        KafkaStreamsInteractiveQueryService service =
                new KafkaStreamsInteractiveQueryService(streamsBuilderFactoryBean);
        service.setRetryTemplate(RetryTemplate.builder()
                .maxAttempts(10)
                .fixedBackoff(500)
                .build());
        return service;
    }

    @Bean
    public StreamsBuilderFactoryBeanConfigurer streamsBuilderConfigurer() {
        return factoryBean -> factoryBean.setStateListener(
                (newState, oldState) ->
                        log.info("Kafka Streams state: {} → {}", oldState, newState));
    }

    public String getSchemaRegistryUrl() { return schemaRegistryUrl; }
}

LogAndContinueExceptionHandler prevents a single malformed message from crashing the entire topology. In production, replace with a dead-letter topic handler.

KafkaStreamsInteractiveQueryService (added in Spring Kafka 3.2.0) wraps the low-level KafkaStreams.store() call with retry logic. The state store is not immediately queryable when the application starts — the Streams app goes through REBALANCING → RUNNING before the store becomes accessible. The retry template absorbs the InvalidStateStoreException during startup.KafkaStreamsInteractiveQueryService (added in Spring Kafka 3.2.0) wraps the low-level KafkaStreams.store() call with retry logic. The state store is not immediately queryable when the application starts — the Streams app goes through REBALANCING → RUNNING before the store becomes accessible. The retry template absorbs the InvalidStateStoreException during startup.

Step 3: Data models

LabEvent Avro schema (src/main/avro/lab-event.avsc)

This is the Phase 2 schema, unchanged:

{
  "type": "record",
  "name": "LabEvent",
  "namespace": "com.labs.events",
  "fields": [
    { "name": "eventId",    "type": "string" },
    { "name": "type",       "type": "string" },
    { "name": "payload",    "type": "string" },
    { "name": "timestamp",  "type": "long"   },
    { "name": "source",     "type": "string" },
    { "name": "couponCode", "type": ["null", "string"], "default": null }
  ]
}

For order events in Phase 3, the fields carry this meaning:

The Avro Maven plugin generates com.labs.events.LabEvent at build time from the .avsc file. All field types are String (not CharSequence) because of <stringType>String</stringType> in the plugin config.

OrderPayload

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderPayload(
        String userId,
        double amount,
        int quantity,
        String productId
) {
    public OrderPayload() { this(null, 0.0, 0, null); }
}

UserProfile

@JsonIgnoreProperties(ignoreUnknown = true)
public record UserProfile(
        String userId,
        String username,
        String email,
        String tier     // BRONZE | SILVER | GOLD
) {
    public UserProfile() { this(null, null, null, "BRONZE"); }
}

The labs.users topic is key-compacted, keyed by userId. The GlobalKTable reads this topic on startup and maintains the latest value per key locally.

EnrichedOrder

@JsonIgnoreProperties(ignoreUnknown = true)
public record EnrichedOrder(
        String orderId,
        String userId,
        String username,
        String tier,
        double amount,
        int quantity,
        String productId,
        String couponCode,
        long timestamp
) {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    public static EnrichedOrder from(LabEvent event, UserProfile profile) {
        OrderPayload payload = parsePayload(event.getPayload());
        String username = profile != null ? profile.username() : "unknown";
        String tier     = profile != null ? profile.tier()     : "UNKNOWN";
        return new EnrichedOrder(
                event.getEventId(),
                payload.userId() != null ? payload.userId() : event.getEventId(),
                username, tier,
                payload.amount(), payload.quantity(), payload.productId(),
                event.getCouponCode(), event.getTimestamp()
        );
    }

    private static OrderPayload parsePayload(String json) {
        try {
            return MAPPER.readValue(json, OrderPayload.class);
        } catch (Exception e) {
            return new OrderPayload();
        }
    }
}

The join callback passes null for profile when no user record exists for a userId. EnrichedOrder.from() handles this gracefully rather than failing the topology.

OrderStats

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderStats(
        String userId,
        String username,
        String tier,
        long orderCount,
        double totalAmount,
        double avgOrderValue,
        double maxOrderAmount,
        int totalQuantity,
        long lastUpdatedAt
) {
    public OrderStats() { this(null, null, "UNKNOWN", 0L, 0.0, 0.0, 0.0, 0, 0L); }

    public static OrderStats empty() { return new OrderStats(); }

    public OrderStats accumulate(EnrichedOrder order) {
        long newCount   = orderCount + 1;
        double newTotal = totalAmount + order.amount();
        int newQty      = totalQuantity + order.quantity();
        return new OrderStats(
                order.userId(), order.username(), order.tier(),
                newCount, newTotal, newTotal / newCount,
                Math.max(maxOrderAmount, order.amount()),
                newQty, System.currentTimeMillis()
        );
    }
}

empty() is the aggregate initializer. accumulate() is the aggregator function — it produces a new immutable record for each incoming order. RocksDB stores the latest value per key. The accumulator must be pure (no side effects) because Kafka Streams may reprocess from the changelog topic during recovery.

JsonSerde

A generic Jackson-backed Serde<T> for non-Avro types:

public class JsonSerde<T> implements Serde<T> {

    private final ObjectMapper mapper = new ObjectMapper();
    private final Class<T> targetType;

    public JsonSerde(Class<T> targetType) { this.targetType = targetType; }

    @Override
    public Serializer<T> serializer() {
        return (topic, data) -> {
            if (data == null) return null;
            try {
                return mapper.writeValueAsBytes(data);
            } catch (Exception e) {
                throw new SerializationException("JSON serialization failed", e);
            }
        };
    }

    @Override
    public Deserializer<T> deserializer() {
        return (topic, data) -> {
            if (data == null) return null;
            try {
                return mapper.readValue(data, targetType);
            } catch (IOException e) {
                throw new SerializationException("JSON deserialization failed", e);
            }
        };
    }
}

Step 4: Building the topology

The topology is defined as a @Bean method in a @Configuration class. Spring Kafka’s StreamsBuilderFactoryBean calls the StreamsBuilder bean to build the Topology and manages the KafkaStreams lifecycle.

@Configuration
public class OrderAnalyticsTopology {

    public static final String STATS_STORE_NAME = "order-stats-store";
    private static final String ORDER_CONFIRMED = "ORDER_CONFIRMED";

    @Bean
    public KStream<String, LabEvent> orderAnalyticsPipeline(StreamsBuilder builder) {

        SpecificAvroSerde<LabEvent> labEventSerde = buildLabEventSerde();
        JsonSerde<UserProfile>  userProfileSerde  = new JsonSerde<>(UserProfile.class);
        JsonSerde<EnrichedOrder> enrichedSerde    = new JsonSerde<>(EnrichedOrder.class);
        JsonSerde<OrderStats>  orderStatsSerde    = new JsonSerde<>(OrderStats.class);

        // Step 1 — GlobalKTable
        GlobalKTable<String, UserProfile> usersTable = builder.globalTable(
                usersTopic,
                Consumed.with(Serdes.String(), userProfileSerde),
                Materialized.as("users-store")
        );

        // Step 2 — Source
        KStream<String, LabEvent> source = builder.stream(
                eventsTopic,
                Consumed.with(Serdes.String(), labEventSerde)
        );

        // Step 3 — Filter
        KStream<String, LabEvent> confirmed = source
                .filter((key, event) ->
                        event != null && ORDER_CONFIRMED.equalsIgnoreCase(event.getType()));

        // Step 4 — Re-key by userId
        KStream<String, LabEvent> keyedByUser = confirmed
                .selectKey((orderKey, event) -> extractUserId(event));

        // Step 5 — Left-join with GlobalKTable
        // leftJoin: profile is null when no user exists — EnrichedOrder handles it.
        // inner join() silently drops events for unrecognised user IDs.
        KStream<String, EnrichedOrder> enriched = keyedByUser
                .leftJoin(
                        usersTable,
                        (streamKey, event) -> streamKey,   // lookup key = userId
                        (event, profile) -> EnrichedOrder.from(event, profile)
                );

        // Step 6 — Aggregate per userId
        KTable<String, OrderStats> statsTable = enriched
                .groupByKey(Grouped.with(Serdes.String(), enrichedSerde))
                .aggregate(
                        OrderStats::empty,
                        (userId, order, acc) -> acc.accumulate(order),
                        Materialized.<String, OrderStats, KeyValueStore<Bytes, byte[]>>
                                as(STATS_STORE_NAME)
                                .withKeySerde(Serdes.String())
                                .withValueSerde(orderStatsSerde)
                );

        // Step 7 — Sink
        statsTable.toStream()
                .to(orderStatsTopic, Produced.with(Serdes.String(), orderStatsSerde));

        return source;
    }
}

Why each step matters

filter() before selectKey()

filter() is a stateless 1→0 or 1→1 operation. It never triggers repartition. Applying it before selectKey() reduces the volume of records that cause repartition and join overhead. Filtering after selectKey() still works correctly but wastes the repartition bandwidth on records you were about to discard.

selectKey() and the repartition marker

selectKey() changes the stream’s key. Kafka Streams marks the stream internally as “dirty” — needing repartition. The physical repartition write-through to an internal topic happens automatically before the first stateful operation downstream (the GlobalKTable join here). The internal topic is named {application-id}-{some-suffix}-repartition.

This is the correct place to re-key. Doing it after the join is too late — the join’s lookup key must match the stream key at join time.

join vs leftJoin with GlobalKTable

KStream.join(GlobalKTable) is an inner join: records with no matching GlobalKTable entry are silently dropped. For order events where the userId isn’t in the users table (new users, seeding race condition, corrupted payload), this means lost data with no error.

KStream.leftJoin(GlobalKTable) keeps every stream record. The joiner receives null for the profile argument when no match exists. EnrichedOrder.from(event, null) substitutes "unknown" and "UNKNOWN" as defaults. This is almost always the right choice for enrichment joins — fail open, not silent.

GlobalKTable vs KTable

The user profile dataset is small and changes infrequently — GlobalKTable is the right choice. A KTable join with labs.users would require that topic to have the same partition count as labs.events.v2 and that partitions are co-located by key. GlobalKTable eliminates that constraint entirely.

groupByKey() vs groupBy()

groupByKey() uses the existing stream key (userId, after selectKey()). It does not trigger a repartition because the key has not changed since the last repartition.

groupBy(newKey) changes the key and always triggers a new repartition. Use groupByKey() whenever the stream key already matches the grouping key.

Named state store

Materialized.as(STATS_STORE_NAME) names the store order-stats-store. This name is required by Interactive Queries — the REST endpoint uses it to locate the store at runtime. Without it, Kafka Streams generates an internal name that changes between deployments.

Step 5: Avro Serde configuration

SpecificAvroSerde<LabEvent> uses the Schema Registry to serialize and deserialize LabEvent records. It must be configured with the Schema Registry URL before use:

private SpecificAvroSerde<LabEvent> buildLabEventSerde() {
    SpecificAvroSerde<LabEvent> serde = new SpecificAvroSerde<>();
    serde.configure(
            Map.of(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
                    streamsConfig.getSchemaRegistryUrl()),
            false   // false = value serde (not key serde)
    );
    return serde;
}

The boolean false tells the serde it is being used for values, not keys. Key serdes and value serdes register schemas under different subject names ({topic}-key vs {topic}-value).

Step 6: Interactive Queries

Interactive Queries allow the application to read directly from the embedded RocksDB state store without going back through Kafka. The state store is local to the instance — in a multi-instance deployment, queries must be routed to the instance that owns the partition for a given key.

Spring Kafka’s KafkaStreamsInteractiveQueryService (available since Spring Kafka 3.2.0) simplifies local queries and adds retry logic for the startup window when the store is not yet queryable:

@GetMapping("/{userId}")
public ResponseEntity<OrderStats> getUserStats(@PathVariable String userId) {
    ReadOnlyKeyValueStore<String, OrderStats> store =
            interactiveQueryService.retrieveQueryableStore(
                    OrderAnalyticsTopology.STATS_STORE_NAME,
                    QueryableStoreTypes.keyValueStore()
            );
    OrderStats stats = store.get(userId);
    return stats != null
            ? ResponseEntity.ok(stats)
            : ResponseEntity.notFound().build();
}

@GetMapping
public ResponseEntity<List<OrderStats>> getAllStats() {
    ReadOnlyKeyValueStore<String, OrderStats> store =
            interactiveQueryService.retrieveQueryableStore(
                    OrderAnalyticsTopology.STATS_STORE_NAME,
                    QueryableStoreTypes.keyValueStore()
            );
    List<OrderStats> result = new ArrayList<>();
    try (KeyValueIterator<String, OrderStats> it = store.all()) {
        it.forEachRemaining(kv -> result.add(kv.value));
    }
    return ResponseEntity.ok(result);
}

The KeyValueIterator implements Closeable — always close it after use. An unclosed iterator holds a RocksDB read handle and may block compaction.

Store availability during rebalance

When Kafka reassigns partitions (scale-out, restart, crash recovery), the Streams app transitions to REBALANCING. The state store is not queryable during this window. Queries during rebalance throw InvalidStateStoreException.

The retry template on KafkaStreamsInteractiveQueryService handles this:

service.setRetryTemplate(RetryTemplate.builder()
        .maxAttempts(10)
        .fixedBackoff(500)
        .build());

For production, expose the store health state via Actuator or a custom health indicator that checks KafkaStreams.state() == RUNNING.

Step 7: Data seeder

The DataSeeder seeds labs.users (user profiles, JSON) and labs.events.v2 (order events, Avro) on application startup. This makes the demo self-contained — no Phase 2 services need to be running.

@Component
public class DataSeeder implements ApplicationRunner {

    private static final List<UserProfile> USERS = List.of(
            new UserProfile("user-1", "alice",   "alice@example.com",   "GOLD"),
            new UserProfile("user-2", "bob",     "bob@example.com",     "SILVER"),
            new UserProfile("user-3", "charlie", "charlie@example.com", "BRONZE")
    );

    @Override
    public void run(ApplicationArguments args) throws Exception {
        seedUserProfiles();
        Thread.sleep;   // allow GlobalKTable to load before events arrive
        seedOrderEvents();
    }

    private void seedOrderEvents() {
        KafkaTemplate<String, LabEvent> avroTemplate = buildAvroTemplate();
        ORDER_SEEDS.forEach(seed -> {
            String orderId = UUID.randomUUID().toString();
            String payload = buildPayload(seed.userId(), seed.amount(),
                                          seed.quantity(), seed.productId());
            LabEvent event = LabEvent.newBuilder()
                    .setEventId(orderId)
                    .setType(seed.type())
                    .setPayload(payload)
                    .setTimestamp(System.currentTimeMillis())
                    .setSource("order-service")
                    .setCouponCode(seed.couponCode())
                    .build();
            avroTemplate.send(eventsTopic, orderId, event);
        });
    }
}

The Thread after seeding user profiles gives the GlobalKTable consumer a moment to load the compacted topic before order events arrive. In production this is unnecessary — the GlobalKTable is always current because it continuously tails the topic. For a demo on startup, the brief delay prevents a race where the first events arrive before the GlobalKTable has finished its initial load.

The DataSeeder creates its own KafkaTemplate<String, LabEvent> configured with KafkaAvroSerializer. Configuring the Avro producer outside the main Spring auto-config avoids wiring conflicts with the String producer used for user profiles.

Step 8: Topic provisioning

The TopicConfig class creates topics via Spring Kafka’s NewTopic beans, which KafkaAdmin provisions automatically on startup:

@Bean
public NewTopic labsUsersTopic() {
    return TopicBuilder.name(usersTopic)
            .partitions(1)
            .replicas(1)
            .compact()      // log compaction: keep latest value per key
            .build();
}

labs.users is compacted because it holds reference data. Compaction retains the latest value per userId key and removes superseded records. The GlobalKTable reads this compacted topic to rebuild its local view — compaction reduces the volume of data the GlobalKTable must replay on restart.

Step 9: Testing with TopologyTestDriver

TopologyTestDriver runs the topology in memory against simulated time. No broker, no Schema Registry, no threads — tests complete in milliseconds.

For Avro Serdes in tests, use MockSchemaRegistryClient. It fulfills schema registration calls in memory without a real registry:

class OrderAnalyticsTopologyTest {

    @BeforeEach
    void setUp() {
        MockSchemaRegistryClient mockRegistry = new MockSchemaRegistryClient();
        SpecificAvroSerde<LabEvent> labEventSerde = new SpecificAvroSerde<>(mockRegistry);
        labEventSerde.configure(
                Map.of(SCHEMA_REGISTRY_URL_CONFIG, "mock://test"), false
        );

        // Build the same topology as production
        StreamsBuilder builder = new StreamsBuilder();
        // ... topology setup ...

        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-order-analytics");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
        config.put(SCHEMA_REGISTRY_URL_CONFIG, "mock://test");

        testDriver = new TopologyTestDriver(builder.build(), config);
    }

    @Test
    void singleConfirmedOrder_createsStats() throws Exception {
        seedUser("user-1", "alice", "GOLD");
        pipeEvent("order-1", "ORDER_CONFIRMED", "user-1", 100.0, 2);

        KeyValueStore<String, OrderStats> store =
                testDriver.getKeyValueStore(OrderAnalyticsTopology.STATS_STORE_NAME);
        OrderStats stats = store.get("user-1");

        assertThat(stats.orderCount()).isEqualTo(1);
        assertThat(stats.totalAmount()).isCloseTo(100.0, within(0.01));
        assertThat(stats.username()).isEqualTo("alice");
        assertThat(stats.tier()).isEqualTo("GOLD");
    }

    @Test
    void nonConfirmedOrders_areFiltered() throws Exception {
        seedUser("user-3", "charlie", "BRONZE");
        pipeEvent("order-X", "ORDER_PENDING",   "user-3", 200.0, 1);
        pipeEvent("order-Y", "ORDER_CANCELLED", "user-3", 200.0, 1);

        KeyValueStore<String, OrderStats> store =
                testDriver.getKeyValueStore(OrderAnalyticsTopology.STATS_STORE_NAME);
        assertThat(store.get("user-3")).isNull();
        assertThat(outputTopic.isEmpty()).isTrue();
    }

    @Test
    void unknownUser_orderStillProcessed() {
        // No profile for user-99 — EnrichedOrder uses "unknown" / "UNKNOWN"
        pipeEvent("order-Z", "ORDER_CONFIRMED", "user-99", 75.0, 1);

        OrderStats stats = testDriver
                .getKeyValueStore(OrderAnalyticsTopology.STATS_STORE_NAME)
                .get("user-99");

        assertThat(stats).isNotNull();
        assertThat(stats.username()).isEqualTo("unknown");
    }
}

The TopologyTestDriver directly pipes input records into topics and reads output records without going through a real broker. The state store is accessible synchronously after pipeInput() returns — no eventual consistency issues in tests.

Running the demo

With Docker Compose

cd kafka-order-analytics-demo
docker compose up --build

The compose file starts Kafka 4.0.0, Schema Registry 7.6.0, a kafka-init container that creates topics, and the order-analytics application. On first startup:

1. Kafka and Schema Registry become healthy
2. kafka-init creates labs.events.v2, labs.users, labs.order-stats
3. order-analytics starts, seeds user profiles and order events
4. The topology begins processing

Query the stats endpoint

After 10–15 seconds (allow the Streams app to reach RUNNING and process the seeded events):

# All user stats
curl http://localhost:8090/api/stats | jq .

# Single user
curl http://localhost:8090/api/stats/user-1 | jq .

Expected response for user-1:

{
  "userId": "user-1",
  "username": "alice",
  "tier": "GOLD",
  "orderCount": 3,
  "totalAmount": 638.49,
  "avgOrderValue": 212.83,
  "maxOrderAmount": 399.0,
  "totalQuantity": 4,
  "lastUpdatedAt": 1722912345678
}

Inspect the Kafka topics

# Watch order-stats sink topic
docker exec kafka-1 /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic labs.order-stats \
  --from-beginning

# Describe the internal repartition topic
docker exec kafka-1 /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --list | grep repartition

The repartition topic appears automatically after selectKey(). It is named order-analytics-app-KSTREAM-KEY-SELECT-...-repartition.

Local development (no Docker)

Start Kafka and Schema Registry separately, then:

mvn spring-boot:run

Performance considerations

RocksDB memory

Kafka Streams uses RocksDB for state stores by default. Each store allocates a write buffer and block cache. With default settings, a state store with millions of keys can use 200+ MB of memory. Configure via RocksDBConfigSetter:

public class CustomRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options,
                          Map<String, Object> configs) {
        BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
        tableConfig.setBlockCacheSize(32 * 1024 * 1024L); // 32 MB per store
        options.setTableFormatConfig(tableConfig);
        options.setWriteBufferSize(8 * 1024 * 1024L);     // 8 MB
    }
}

Register it via StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG.

selectKey and repartition cost

selectKey() writes every matching record to an internal repartition topic before the join. For high-throughput streams, this doubles the write amplification — every CONFIRMED order is written twice (once to labs.events.v2, once to the repartition topic).

If the original producer can key events by userId at write time, the selectKey() and its repartition become unnecessary. Consider this a design optimization for high-volume pipelines.

GlobalKTable startup time

On first start or after a state store wipe, the GlobalKTable must read the entire labs.users topic. For a topic with 50,000 compacted user records, this takes a few seconds. For 10 million records, it may take minutes. Monitor startup with KafkaStreams.state().

Commit interval and latency

commit.interval.ms: 1000 means processed offsets are committed to Kafka every second. The Interactive Queries store is updated as records are processed, not on commit boundaries — so the REST endpoint always reflects the latest state even between commits. Lowering the commit interval improves failure recovery at the cost of higher broker write pressure.

Common pitfalls

Store not queryable at startup

InvalidStateStoreException: Cannot get state store order-stats-store because the stream thread is in state REBALANCING is thrown when a query arrives before the Streams app reaches RUNNING. The RetryTemplate on KafkaStreamsInteractiveQueryService handles this during normal startup. If it persists, the topology may be stuck in rebalance — check broker connectivity and consumer group status.

Filter after selectKey wastes repartition bandwidth

Applying filter() after selectKey() works functionally but discards records after they have already been written to the repartition topic. Keep filter() before any key-changing operations.

Multiple selectKey calls

Each selectKey() triggers a repartition. Two selectKey() calls = two internal topics = double repartition overhead. Restructure the topology to do all key changes in one step.

Changelog topic misconfiguration

Every named state store gets a changelog topic: {application-id}-{store-name}-changelog. This topic must have replication-factor >= min.insync.replicas in production. On a single-broker setup (like this demo), replication-factor: 1 is the only valid option.

record types and Jackson

JsonSerde<T> uses Jackson to serialize records. Jackson requires either a no-args constructor or @JsonCreator to deserialize records. The no-args compact constructor (public OrderStats() { this(...); }) satisfies this requirement. Without it, deserialization from the state store changelog fails at startup.

Alternative approaches

Kafka Streams is the right choice when:

• The processing logic lives in the same service that consumes the events
• You need stateful aggregation without a separate processing cluster
• The state dataset fits comfortably in disk (RocksDB handles spill to disk)
• The team already manages Kafka

It is the wrong choice when:

• You need sub-100ms end-to-end processing with exactly-once at extremely high throughput (Flink’s watermark model is more expressive)
• The processing requires joins across more than two topics that are not naturally co-partitionable
• Ops prefers SQL-based definitions over Java topology code

Key Takeaways

• filter() before selectKey() reduces repartition volume. Every key-changing operation triggers an internal repartition write — minimize what crosses it.
• GlobalKTable eliminates co-partitioning requirements for reference data. Its tradeoff is full dataset replication on every instance; keep it under ~100 MB.
• Name state stores explicitly with Materialized.as(name). Interactive Queries require the name to be stable across deployments — auto-generated names change and break the endpoint.

Week 5 concept map

Week 5 — what you can now build

• ✅ Build stream topologies with DSL operators: filter, map, flatMap, branch
• ✅ Aggregate with state: count / reduce / aggregate
• ✅ Window and session analytics: tumbling / hopping / session
• ✅ Enrich events via joins: KTable & GlobalKTable
• ✅ Query state interactively: RocksDB via REST endpoint
• ✅ Ship production Streams apps with exactly-once guarantees

---

The complete source code is available on GitHub.

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

Next

➡️ Day 36 — Connect Basics: Source & Sink Connectors

References

• Kafka Streams Developer Guide
• Kafka Streams Interactive Queries
• Spring Kafka — Kafka Streams Support
• Spring Kafka — KafkaStreamsInteractiveQueryService
• GlobalKTable vs KTable
• RocksDB configuration in Kafka Streams
• Confluent: Kafka Streams Data Transformation
• TopologyTestDriver

Link to Medium blog