Kafka Streams — Topology Architecture & Execution Model

A topology is a directed acyclic graph (DAG) of processing nodes. Every Kafka Streams application is, at its core, a compiled topology definition executed by a managed pool of stream threads and tasks.

Source Nodes    ──►   Processor Nodes (transforms)   ──►   Sink Nodes
(read from            (filter, map, aggregate,             (write to
 Kafka topics)         join, branch, etc.)                  Kafka topics)

---

1. Topology Visualization

Kafka Streams Topology: DAG Compilation, Auto-Naming & Sub-Topology Ordering

Kafka Streams Topology DAG (Directed Acyclic Graph)

Overview: Visualizes the compilation of DSL code into physical processor nodes (Source, Processor, Sink).

• Source Processor: Subscribes to the input topic (e.g. orders) and deserializes key-value records.
• Filter Processor: Applies boolean predicates to stream elements, discarding records that do not match.
• MapValues Processor: Performs stateless transformations on record values while preserving the partition key.
• Sink Processor: Publishes the final transformed stream to output topics (e.g. high-value-orders).

Defining Topologies via High-Level DSL

StreamsBuilder builder = new StreamsBuilder();

// Source node: read from Kafka topic
KStream<String, OrderEvent> rawOrders = builder.stream(
    "orders-raw",
    Consumed.with(Serdes.String(), orderSerde)
        .withOffsetResetPolicy(AutoOffsetReset.EARLIEST)
);

// GlobalKTable for reference data join
GlobalKTable<String, Product> products = builder.globalTable(
    "product-catalog",
    Consumed.with(Serdes.String(), productSerde),
    Materialized.<String, Product, KeyValueStore<Bytes, byte[]>>as("product-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(productSerde)
);

// Transformation pipeline
KStream<String, EnrichedOrder> enriched = rawOrders
    .filter((key, order) -> order != null && order.isValid(),
        Named.as("filter-invalid-orders"))
    .join(
        products,
        (orderKey, order) -> order.getProductId(),
        (order, product) -> order.enrichWith(product),
        Named.as("join-product-catalog")
    );

// Build and inspect the topology
Topology topology = builder.build();
System.out.println(topology.describe()); // Always inspect in development

---

2. Stream Branching: split() vs Spring Kafka KafkaStreamBrancher.onTopOf()

Historically in Kafka Streams (< 2.8), splitting a stream used stream.branch(Predicate...), which returned an untyped array KStream<K, V>[]. Developers had to access branches using brittle array indices like branches[0], where adding or reordering predicates silently broke downstream processing.

1. Native Kafka Streams 2.8+ (split() and Branched)

Modern Kafka Streams provides a type-safe, fluent split() API that names branches and supports chained consumers:

enriched.split(Named.as("orders-branch-"))
    .branch((k, v) -> v.isVip(),
        Branched.withConsumer(ks -> ks.to("vip-orders", Produced.with(Serdes.String(), orderSerde))))
    .branch((k, v) -> v.isFraudRisk(),
        Branched.withConsumer(ks -> ks.to("fraud-review", Produced.with(Serdes.String(), orderSerde))))
    .defaultBranch(
        Branched.withConsumer(ks -> ks.to("standard-orders", Produced.with(Serdes.String(), orderSerde))));

2. Spring Kafka KafkaStreamBrancher.onTopOf(stream)

In Spring Boot applications using spring-kafka, org.springframework.kafka.support.KafkaStreamBrancher provides an alternative builder pattern that defines branching rules upfront and attaches them onto an existing KStream via .onTopOf(stream):

KStream<String, Order> orderStream = builder.stream("orders-raw", Consumed.with(Serdes.String(), orderSerde));

new KafkaStreamBrancher<String, Order>()
    .branch((key, order) -> order.getTotal().compareTo(new BigDecimal("500")) >= 0,
        ks -> ks.to("vip-orders", Produced.with(Serdes.String(), orderSerde)))
    .branch((key, order) -> order.isFlaggedForReview(),
        ks -> ks.to("fraud-review-orders", Produced.with(Serdes.String(), orderSerde)))
    .defaultBranch(ks -> ks.to("standard-orders", Produced.with(Serdes.String(), orderSerde)))
    .onTopOf(orderStream); // Binds all branches onto the base stream and returns it

How .onTopOf(...) Works Under the Hood:

1. Separation of Definition from Stream Instance: You can define a reusable KafkaStreamBrancher bean or strategy object and apply it across multiple streams or in unit tests.
2. Terminal vs Continuation Chain: .onTopOf(stream) returns the original orderStream, allowing you to continue chaining downstream operators on the root stream if needed, while each branch lambda consumes its filtered slice.
3. Execution Semantics: Under the hood, KafkaStreamBrancher evaluates predicates sequentially in declaration order. The first predicate returning true routes the record; remaining predicates are skipped for that record.

---

3. Merging Streams (stream.merge()) vs Merging Topologies

┌─────────────────────────────────────────────────────────────────────────┐
│ Stream-Level Merge: streamA.merge(streamB)                              │
│                                                                         │
│ Topic A ──► [ streamA ] ──┐                                             │
│                           ├──► [ merge() ] ──► [ combined KStream ]     │
│ Topic B ──► [ streamB ] ──┘                                             │
│ (Same Key & Value SerDes required; interweaves records in Stream-Time)  │
└─────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────┐
│ Topology-Level Merging: Multiple Pipelines in One StreamsBuilder        │
│                                                                         │
│ Pipeline 1: Topic A ──► [ Process ] ──► Topic Out 1   (Sub-topology 0)  │
│ Pipeline 2: Topic B ──► [ Process ] ──► Topic Out 2   (Sub-topology 1)  │
│                                                                         │
│ Both registered in SAME StreamsBuilder ──► builder.build() = 1 Topology │
│ Managed by 1 KafkaStreams client and 1 Consumer Group (application.id)  │
└─────────────────────────────────────────────────────────────────────────┘

1. Merging Streams (KStream.merge)

streamA.merge(streamB) is a stream-level union operator. It combines records from two independent streams having the exact same Key and Value types into a single downstream KStream:

KStream<String, ClickEvent> webClicks = builder.stream("clicks-web");
KStream<String, ClickEvent> mobileClicks = builder.stream("clicks-mobile");

// Merge creates a single logical stream downstream without an extra topic hop
KStream<String, ClickEvent> allClicks = webClicks.merge(mobileClicks, Named.as("merge-clicks"));

2. Merging Topologies in Spring Boot

In Spring Boot (@EnableKafkaStreams), declaring multiple @Bean methods that each take StreamsBuilder mutates the same shared StreamsBuilder instance. When StreamsBuilderFactoryBean.start() is called, it executes builder.build(), synthesizing all those independent beans into one single composite Topology object!

---

4. Why Topology Naming Is Critical for Production

Kafka Streams Topology: DAG Compilation, Auto-Naming & Sub-Topology Ordering

The Auto-Generated Naming Shift Trap

Overview: Explains why explicit node naming is critical for stateful production deployments.

• Stateful operators (aggregations, joins) compile into state stores (e.g. RocksDB) backed by changelog topics.
• By default, Kafka Streams auto-generates names sequentially: KSTREAM-SOURCE-0000000000 -> KSTREAM-FILTER-0000000001.
• If you insert or reorder a simple filter node, the counter shifts downstream, corrupting local state bindings and changelog topic alignment!

Auto-generated internal names for operators, state stores, and repartition topics look like: KSTREAM-FILTER-0000000002. These names are used as:

• Kafka internal topic names: app-id-KSTREAM-FILTER-0000000002-repartition
• RocksDB state directory names: /tmp/kafka-streams/KSTREAM-MAPVALUES-0000000003
• Changelog topic names: app-id-KSTREAM-AGGREGATE-STATE-STORE-0000000004-changelog

If you add, remove, or reorder any operator in the topology without explicit names, all downstream auto-generated index counters shift. This causes:

Deployment without explicit naming:

  v1 topology:  FILTER-0002 ──► MAPVALUES-0003 ──► AGGREGATE-0004
  v2 topology:  FILTER-0002 ──► FILTER-0003 ──► MAPVALUES-0004 ──► AGGREGATE-0005
                              (added a second filter)

  On startup:
    AGGREGATE-0005 looks for changelog topic: "app-AGGREGATE-0005-changelog" ──► NOT FOUND
    ──► Creates new changelog topic
    ──► Full state rebuild from scratch (minutes to hours for large state)
    ──► Old changelog topic "app-AGGREGATE-0004-changelog" orphaned (wasting disk)

Always enforce explicit naming:

rawOrders
    .filter((k, v) -> v.isValid(), Named.as("filter-valid-orders"))
    .mapValues(v -> v.normalize(), Named.as("normalize-order"))
    .groupByKey(Grouped.as("group-by-customer"))
    .aggregate(
        OrderSummary::new,
        (key, order, summary) -> summary.add(order),
        Named.as("aggregate-order-summary"),
        Materialized.<String, OrderSummary, KeyValueStore<Bytes, byte[]>>as("order-summary-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(orderSummarySerde)
    );

---

5. Internal Execution Model: Tasks & Threads

Tasks — The Unit of Parallelism

Kafka Streams Execution Model: Tasks, Threads & Processing Loop

Scale Instances:

Topic "orders-raw" (6 Partitions) ➔ 6 Stream TasksMax Parallelism = 6 (1 task per partition)

Instance A2 Tasks

Task 0_0 (Partition 0)

• Dedicated RocksDB store
• Consumer Offset #0

Task 0_3 (Partition 3)

• Dedicated RocksDB store
• Consumer Offset #3

Instance B2 Tasks

Task 0_1 (Partition 1)

• Dedicated RocksDB store
• Consumer Offset #1

Task 0_4 (Partition 4)

• Dedicated RocksDB store
• Consumer Offset #4

Instance C2 Tasks

Task 0_2 (Partition 2)

• Dedicated RocksDB store
• Consumer Offset #2

Task 0_5 (Partition 5)

• Dedicated RocksDB store
• Consumer Offset #5

PRODUCTION ARCHITECTURE INSIGHT

Unit of Parallelism: In Kafka Streams, tasks (not threads or pods) are the atomic unit of work. Adding more instances than source topic partitions creates idle pods. Setting `num.stream.threads > 1` allows a single pod with multi-core CPUs to process multiple tasks concurrently with independent RocksDB directories.

Kafka Streams divides a topology into tasks, one per source partition. Each task is an independent, isolated processing unit with its own consumer offset tracking, state store instance (RocksDB directory), and in-memory record buffer.

Topic "orders-raw" has 6 partitions:
  Partition 0 ──► Task 0_0
  Partition 1 ──► Task 0_1
  Partition 2 ──► Task 0_2
  Partition 3 ──► Task 0_3
  Partition 4 ──► Task 0_4
  Partition 5 ──► Task 0_5

Stream Threads — Concurrency Within an Instance

Each application instance runs one or more stream threads (num.stream.threads). Each thread runs its own continuous event loop: $\text{poll}(100\text{ms}) \longrightarrow \text{process records} \longrightarrow \text{commit offsets}$

---

6. Scaling & Sizing Math

$\text{Max Active Tasks} = \sum_{s \in \text{Sub-topologies}} \left( \max_{t \in \text{SourceTopics}(s)} \text{Partitions}(t) \right)$

• Thread Sizing Rule of Thumb: Allocate 1 Stream Thread per dedicated vCPU core, up to the number of tasks assigned to that instance.
• Thread Idling Trap: If your cluster has 12 total tasks and you deploy 3 pods each configured with num.stream.threads = 8 ($3 \times 8 = 24\text{ threads}$), exactly 12 threads will process tasks while the remaining 12 threads sit completely idle.
• Instance Saturation Limit: You cannot scale beyond the total task count. If total active tasks = 12, deploying 16 Kubernetes pods leaves 4 pods with 0 active tasks (idle unless hosting standby tasks).

---

7. Rebalancing Deep Dive: Cooperative Sticky Assignor

┌────────────────────────────────────────────────────────────────────────┐
│ Cooperative Sticky Rebalance (Modern Incremental Handoff):             │
│ Pod 1: [ Task 0 ] CONTINUES RUNNING ──► Only Task 1 revoked            │
│ Pod 2: [ Task 2 ] CONTINUES RUNNING ──► Only Task 3 revoked            │
│ Pod 3: (New instance joins) ──► Rebuilt state warm in background        │
│ ✅ Unaffected tasks NEVER stop processing; sub-second handoff!          │
└────────────────────────────────────────────────────────────────────────┘

The Cooperative Sticky Assignor uses incremental rebalancing:

1. Incremental Revocation: Only tasks migrating to another node are paused. All other tasks continue processing real-time events without interruption.
2. Probing Rebalances (probing.rebalance.interval.ms): When a new instance joins, the assignor assigns existing active tasks to their current owners, but assigns warm standby tasks to the new instance. Every 10 minutes (configurable), a lightweight probing rebalance checks whether the standby task's changelog lag is within acceptable.recovery.lag (default 10,000 records). Once caught up, an instant sub-second hot swap occurs.

---

8. Sub-Topology Reordering Disaster & Production Migration Runbook

Kafka Streams Topology: DAG Compilation, Auto-Naming & Sub-Topology Ordering

Sub-Topology 0Task ID: 0_0, 0_1...

Pipeline: Orders Stream (Original Index 0)

Consumes: topic "orders-raw"

Sub-Topology 1Task ID: 1_0, 1_1...

Pipeline: Payments Stream (Original Index 1)

Consumes: topic "payments-raw"

HOMOGENEOUS CLUSTER STATE (NORMAL)NORMAL PROCESSING

All pods agree on Sub-topology IDs: Task 0_0 processes orders-raw-0 and Task 1_0 processes payments-raw-0. No assignment conflicts occur during group rebalances.

Sub-Topology Reordering & Perpetual Rebalance Storm

Overview: Demonstrates why changing the order of independent sub-topologies in code breaks rolling deployments and halts all processing.

• Kafka Streams numbers sub-topologies sequentially (0, 1, 2...) based on declaration order in StreamsBuilder.
• Task IDs are computed as TaskId(subTopologyId, partitionId) — e.g. Task 0_0.
• During a rolling update, a v2 leader assigns Task 0_0 (Payments) to a v1 follower whose local topology expects Task 0_0 to process Orders.
• The follower throws TaskAssignmentException and rejoins the group, triggering an endless cluster-wide rebalance loop (0 msg/sec).

If you reorder sub-topologies during a rolling update, Task 0_0's topic assignment clashes between v1 and v2 pods, triggering an infinite rebalance storm with zero records processed:

Interactive Simulator: Rebalance Storm Duration & Outage Timeline

DEPLOYMENT PARAMETERS & WORKLOAD SIZE

Pod Count (N_pods):10 Pods

Deployment Strategy:

RocksDB State Size:20 GB / Store

🚨 ESTIMATED TOTAL OUTAGE DURATION

13m 25s

Phase 1 (Rebalance Storm): 5m 50s

Phase 2 (State Restore): 7m 35s

OUTAGE TIMELINE PHASES (CLICK TO INSPECT)

Phase 1: Storm (5m 50s)

Phase 2: RocksDB Replay (7m 35s)

PHASE 1: THE ROLLING REBALANCE STORM (0 MSG/SEC)

Active Mixed Cluster Conflict (10 Pods Rolling)

While at least one v1 pod and one v2 pod coexist, the leader's assignment clashes with follower topology definitions. Follower throws TaskAssignmentException, revoking partitions and triggering repeated rebalances.

• Throughput: 0 records/sec • Consumer Lag: Accumulating rapidly on input topics.

Safe Production Migration Runbook

Interactive Topology Migration & Decommissioning Runbook

EXECUTION RUNBOOK STEPSGold Standard (0ms Downtime)

Step 1: Increment application.id

Set application.id = "order-service-v2". Creates independent consumer group and separate changelog topics.

Step 2: Dual Run & Catch-Up

Deploy v2 pods. v2 builds RocksDB state in background while v1 serves active live production traffic.

Step 3: Verify Lag = 0

Monitor consumer group lag on v2 until catch-up is complete (offset lag < 100 records).

Step 4: Switch Traffic & Terminate v1

Route API / downstream consumers to v2 output topics. Scale v1 deployment to 0 replicas.

Step 5: Purge Orphaned v1 Topics

Run kafka-topics.sh --delete on old v1 changelog and repartition internal topics.

STEP 1 DEEP DIVE

Step 1: Increment application.id

Set application.id = "order-service-v2". Creates independent consumer group and separate changelog topics.

ARCHITECTURAL BENEFIT:

Guarantees zero dropped records and zero downtime. If any issue occurs in v2, traffic can be instantly rolled back to v1.

1. Strategy 1: Blue-Green / Application ID Versioning (Gold Standard): Increment application.id (e.g. order-service-v2), deploy alongside v1, wait for changelogs to catch up, switch downstream consumers, decommission v1.
2. Strategy 2: Cold Maintenance Window & Application Reset: Scale replicas to 0, run kafka-streams-application-reset, wipe local state, deploy v2.
3. Strategy 3: Architectural Decoupling: Split independent pipelines into separate microservices with isolated application.ids.
4. Strategy 4: Append-Only Evolution: Never delete or reorder sub-topologies at indices 0..N-1. Append new pipelines only at the very end (index N), or replace decommissioned pipelines with a dummy no-op stub (filter((k, v) -> false)).

Keeping Services Alive: Decoupled Probes

Interactive Fault Tolerance & Service Uptime Guardrails

SCENARIO CONFIGURATION

PRODUCTION ARCHITECTURAL IMPACT

How Decoupled Probes Save Rolling Updates

By separating Liveness (JVM alive = 200 OK) from Readiness (Streams running = 200 OK), Kubernetes stops external HTTP traffic during rebalances but NEVER kills the pod. This allows rolling updates to finish cleanly.

• Liveness Probe (/health/live): Checks JVM process health only (always returns 200 OK while JVM is up).
• Readiness Probe (/health/ready): Checks kafkaStreams.state() == RUNNING (returns 503 Service Unavailable during rebalances so traffic is paused, but container is NEVER killed).