7.1 Topology Design & Internal Topics

Module Overview

Goal: Understand Kafka Streams topology structure and internal topics (repartition, changelog).

What You'll Learn:

• What a topology is
• Sub-topologies and parallelism
• Repartition topics (when/why created)
• Internal topic naming and configuration
• Topology visualization and debugging
• Best practices for topology design

What's NOT Covered Here:

• Changelog topics (covered in Module 6.2)
• Specific operations (covered in Modules 4, 5, 8)
• EOS internals (covered in Module 9.1)

Key Insight: The topology is the computational graph — understanding its structure is key to performance, parallelism, and troubleshooting.

1. What is a Topology?

Definition

Topology = Directed Acyclic Graph (DAG) of processing nodes

Components:
├─ Source nodes (read from Kafka topics)
├─ Processor nodes (transform data)
└─ Sink nodes (write to Kafka topics)

Edges represent data flow

Simple Example

// Code
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> input = builder.stream("input-topic");
input.filter((k, v) -> v.length() > 5)
     .to("output-topic");

// Topology structure
Source (input-topic)
   ↓
Processor (filter)
   ↓
Sink (output-topic)

Complex Example

// Code
KStream<String, String> input = builder.stream("orders");
KTable<String, Long> counts = input
    .groupByKey()
    .count();

input.join(counts, ...)
     .to("enriched-orders");

// Topology structure (simplified)
Source (orders)
   ↓
   ├──→ Group → Count → KTable (counts)
   │                        ↓
   └──→ Join ←──────────────┘
        ↓
     Sink (enriched-orders)

2. Sub-Topologies

What are Sub-Topologies?

Sub-topology = independent processing sub-graph

Created when:
├─ No direct data flow between parts
└─ Reading from different source topics

Each sub-topology can scale independently

Example: Two Independent Streams

// Stream 1: orders → process → output1
builder.stream("orders")
    .mapValues(v -> v.toUpperCase())
    .to("output1");

// Stream 2: payments → process → output2
builder.stream("payments")
    .filter((k, v) -> v.contains("USD"))
    .to("output2");

// Result: 2 sub-topologies
// They share no data flow

Visualizing Sub-Topologies

Sub-topology 0:
  Source (orders) → Map → Sink (output1)

Sub-topology 1:
  Source (payments) → Filter → Sink (output2)

Each can have different parallelism

Why Sub-Topologies Matter

PARALLELISM:
├─ Each sub-topology processes independently
├─ Can scale sub-topologies separately
└─ Max parallelism = partitions per sub-topology

RESOURCE ISOLATION:
├─ One sub-topology slow ≠ other slow
└─ Independent processing rates

3. Topology Inspection

Describing the Topology

Topology topology = builder.build();
System.out.println(topology.describe());

Example Output

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])
      --> KSTREAM-FILTER-0000000001
    Processor: KSTREAM-FILTER-0000000001 (stores: [])
      --> KSTREAM-SINK-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Sink: KSTREAM-SINK-0000000002 (topic: output-topic)
      <-- KSTREAM-FILTER-0000000001

Interpreting Topology Description

Each node shows:
├─ Type (Source, Processor, Sink)
├─ Name (auto-generated)
├─ Inputs (arrows: -->)
├─ Outputs (arrows: <--)
└─ State stores (if any)

Useful for:
├─ Understanding data flow
├─ Debugging processing issues
└─ Identifying repartitions

4. Internal Topics

Overview

Internal topics = topics created/managed by Kafka Streams

Two types:
├─ Repartition topics
└─ Changelog topics

Naming pattern:
{application.id}-{operation}-{type}

Repartition Topics

Purpose: Re-distribute data by new key

Created when:
├─ groupBy() changes key
├─ Join requires co-partitioning
└─ Aggregation after key change

Example:
my-app-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition

Changelog Topics

Purpose: Backup state stores

Created for:
├─ Every stateful operation (count, aggregate, etc.)
└─ Ensures state durability

Example:
my-app-user-counts-changelog

(Covered in detail in Module 6.2)

5. Repartition Topics Deep Dive

When Repartitioning Happens

// DOES NOT require repartition (key unchanged)
stream.groupByKey()
      .count();

// REQUIRES repartition (key changed)
stream.groupBy((k, v) -> v.userId())  // New key!
      .count();

// REQUIRES repartition (key changed)
stream.selectKey((k, v) -> v.customerId())
      .groupByKey()
      .count();

Why Repartitioning is Needed

Problem:
├─ Kafka guarantees order within partition
├─ Stateful operations need same-key records together
└─ Key change → records may be on wrong partition

Solution:
├─ Write to repartition topic with new key
├─ Kafka partitions by new key
└─ Read from repartition topic for processing

This ensures co-partitioning

Repartition Flow

Original stream:
Key: orderId, Value: Order (contains customerId)

After groupBy(customerId):
┌────────────────────────────────────────┐
│ Source (orders)                        │
│   ↓                                    │
│ SelectKey (customerId)                 │
│   ↓                                    │
│ Sink (repartition topic)               │  ← Write with new key
└────────────────────────────────────────┘
           ↓ (Kafka re-partitions)
┌────────────────────────────────────────┐
│ Source (repartition topic)             │  ← Read from repartition
│   ↓                                    │
│ GroupBy → Count                        │
│   ↓                                    │
│ Sink (output)                          │
└────────────────────────────────────────┘

Repartition Topic Configuration

// Control repartition behavior
stream.groupBy((k, v) -> v.userId())
      .count(Materialized.as("user-counts"));

// Repartition topic created automatically with:
├─ Partitions: same as input topic (default)
├─ Replication factor: same as input
├─ Cleanup policy: delete (not compacted)
├─ Retention: 1 day (default)

Custom Repartition Config

// Use `repartition()` for explicit control
stream.repartition(Repartitioned.<String, Order>as("custom-repartition")
          .withNumberOfPartitions(10)
          .withKeySerde(Serdes.String())
          .withValueSerde(orderSerde))
      .groupByKey()
      .count();

6. Co-Partitioning

What is Co-Partitioning?

Co-partitioning = records with same key on same partition

Required for:
├─ Joins (stream-stream, stream-table)
├─ Stateful operations
└─ Windowed operations

Ensures same-key records processed together

Co-Partitioning Requirements

For two topics to be co-partitioned:

1. Same number of partitions
2. Same partitioning strategy
3. Same key type

If not co-partitioned:
└─ Kafka Streams automatically repartitions

Example: Join Without Repartition

// orders and customers both have 10 partitions
// Both partitioned by customerId

KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");

orders.join(customers, ...)  // No repartition needed

Example: Join With Repartition

// orders: 10 partitions, key = orderId
// customers: 10 partitions, key = customerId

// Must change orders key to customerId
orders.selectKey((k, v) -> v.customerId())  // Key changed
      .join(customers, ...)  // Repartition happens automatically

7. Parallelism and Tasks

Tasks

Task = unit of parallelism in Kafka Streams

One task per input partition:
├─ 10 partitions → 10 tasks
└─ Each task processes one partition

Tasks distributed across instances

Example: 3 Instances, 9 Partitions

Instance 1: Tasks 0, 1, 2
Instance 2: Tasks 3, 4, 5
Instance 3: Tasks 6, 7, 8

Max parallelism = 9 (number of partitions)

Sub-Topology Parallelism

Each sub-topology has independent tasks:

Sub-topology 0 (input: orders, 10 partitions):
├─ 10 tasks

Sub-topology 1 (input: payments, 5 partitions):
├─ 5 tasks

Total: 15 tasks across 2 sub-topologies

Over-Partitioning for Future Scale

Strategy: Start with more partitions than current load

Example:
├─ Current load: 3 instances
├─ Partitions: 30 (10× instances)
└─ Future: Can scale to 30 instances without repartition

Trade-off:
✅ Easy scaling
❌ More tasks per instance initially
❌ More metadata overhead

8. Topology Design Patterns

Pattern 1: Linear Pipeline

// Simple linear flow
builder.stream("input")
    .filter(...)
    .map(...)
    .to("output");

Characteristics:
├─ One sub-topology
├─ Easy to understand
└─ Limited parallelism

Pattern 2: Branching

// Split stream into multiple outputs
KStream<String, Order> orders = builder.stream("orders");

orders.filter((k, v) -> v.amount() > 1000)
      .to("large-orders");

orders.filter((k, v) -> v.amount() <= 1000)
      .to("small-orders");

Characteristics:
├─ One source, multiple sinks
├─ No repartition needed
└─ Still one sub-topology

Pattern 3: Join and Enrich

// Join stream with table
KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");

orders.selectKey((k, v) -> v.customerId())
      .join(customers, ...)
      .to("enriched-orders");

Characteristics:
├─ Repartition likely needed
├─ State for table
└─ Two sub-topologies (orders, customers)

Pattern 4: Fan-In

// Merge multiple streams
KStream<String, Event> stream1 = builder.stream("topic1");
KStream<String, Event> stream2 = builder.stream("topic2");

KStream<String, Event> merged = stream1.merge(stream2);
merged.to("output");

Characteristics:
├─ Multiple sources, one output
├─ No repartition needed (if keys compatible)
└─ One or multiple sub-topologies

Pattern 5: Aggregation Pipeline

// Multi-stage aggregation
builder.stream("events")
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count()
    .toStream()
    .to("windowed-counts");

Characteristics:
├─ Stateful (state store + changelog)
├─ Repartition if key changed
└─ One sub-topology

9. Internal Topic Naming

Naming Convention

Repartition topic:
{application.id}-{operation}-repartition

Changelog topic:
{application.id}-{store-name}-changelog

Example:
Application ID: my-streaming-app

Topics created:
├─ my-streaming-app-user-counts-changelog
├─ my-streaming-app-KSTREAM-AGGREGATE-0000000003-repartition
└─ my-streaming-app-windowed-store-changelog

Why Naming Matters

DEBUGGING:
├─ Identify which app created topic
└─ Understand purpose (repartition vs changelog)

CLEANUP:
├─ Delete app-specific topics on reset
└─ Avoid orphaned topics

MONITORING:
├─ Track internal topic lag
└─ Measure repartition overhead

10. Managing Internal Topics

Auto-Created Topics

By default:
├─ Kafka Streams creates internal topics automatically
├─ Partitions: match input topic
├─ Replication factor: match input topic
└─ Retention: varies (repartition: delete, changelog: infinite)

Custom Configuration

// Configure internal topics
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);

// All internal topics will use replication factor 3

Explicit Topic Configuration

// Configure specific internal topic
Map<String, String> topicConfig = new HashMap<>();
topicConfig.put("retention.ms", "86400000");  // 1 day

// Apply to repartition topic
stream.repartition(Repartitioned.<String, Order>as("my-repartition")
          .withNumberOfPartitions(10)
          .withStreamPartitioner(...)
      );

Deleting Internal Topics

# List internal topics
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep my-app

# Delete repartition topic (safe, recreated on restart)
kafka-topics.sh --bootstrap-server localhost:9092 \
  --delete --topic my-app-...-repartition

# ⚠️ Delete changelog topic (NOT RECOMMENDED)
# State lost if no local backup
kafka-topics.sh --bootstrap-server localhost:9092 \
  --delete --topic my-app-user-counts-changelog

11. Topology Optimization

Minimize Repartitions

// BAD: Multiple repartitions
stream.groupBy((k, v) -> v.field1()).count()
      .toStream()
      .groupBy((k, v) -> k.field2()).count();  // 2 repartitions

// GOOD: Single repartition
stream.selectKey((k, v) -> v.field1())
      .groupByKey().count();  // 1 repartition

Use groupByKey() When Possible

// BAD: Unnecessary repartition
stream.groupBy((k, v) -> k).count();  // Repartitions even though key unchanged

// GOOD: No repartition
stream.groupByKey().count();  // Recognizes key unchanged

Pre-Partition Input Topics

Strategy: Partition input topics by processing key

Example:
├─ Input: orders topic, key = customerId
├─ Processing: aggregate by customerId
└─ No repartition needed (already partitioned correctly)

Pre-partitioning reduces overhead

Reduce State Store Count

// BAD: Many small state stores
stream.groupByKey().count(Materialized.as("count1"));
stream.groupByKey().aggregate(..., Materialized.as("agg1"));
stream.groupByKey().reduce(..., Materialized.as("reduce1"));
// 3 state stores, 3 changelogs

// GOOD: Combine operations
stream.groupByKey()
      .aggregate(
          () -> new Stats(),
          (k, v, agg) -> {
              agg.count++;
              agg.sum += v;
              return agg;
          },
          Materialized.as("combined-stats")
      );
// 1 state store, 1 changelog

12. Debugging Topologies

Enable Topology Logging

// Print topology on startup
Topology topology = builder.build();
System.out.println(topology.describe());

KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();

Visualize Topology

# Use Kafka Streams Topology Visualizer
# https://zz85.github.io/kafka-streams-viz/

# Paste topology.describe() output
# Get visual graph representation

Check Internal Topics

# List all topics for your app
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep my-app

# Identify internal topics
# Repartition: *-repartition
# Changelog: *-changelog

Monitor Repartition Lag

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

# Look for repartition topics
# High lag = repartition bottleneck

Inspect Internal Topic Data

# Peek at repartition topic
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-app-...-repartition \
  --from-beginning \
  --max-messages 10

# Verify keys/values correct

13. Topology Best Practices

1. Design Before Coding

Planning checklist:
├─ What are the inputs (topics)?
├─ What are the outputs?
├─ What transformations needed?
├─ Where is state required?
├─ Where will repartitioning occur?
└─ How many sub-topologies?

Sketch topology on paper first

2. Name State Stores Explicitly

// BAD: Auto-generated names
stream.groupByKey().count();

// GOOD: Explicit names
stream.groupByKey().count(Materialized.as("user-counts"));

Benefits:
├─ Easier debugging
├─ Predictable changelog topic names
└─ Interactive queries (Module 11.1)

3. Understand Repartition Cost

Repartition overhead:
├─ Network I/O (write + read)
├─ Disk I/O (broker storage)
├─ Latency (additional hop)
└─ Topic management

Minimize when possible

4. Use Appropriate Partition Count

Too few partitions:
├─ Limited parallelism
└─ Can't scale

Too many partitions:
├─ High metadata overhead
├─ More tasks per instance
└─ Slower rebalance

Rule of thumb: 2-10× current instance count

5. Monitor Internal Topics

Watch for:
├─ Repartition topic lag (bottleneck)
├─ Changelog topic growth (state size)
├─ Orphaned topics (app deleted but topics remain)
└─ Disk usage on brokers

14. Common Topology Pitfalls

Pitfall 1: Unexpected Repartition

// Looks innocent but repartitions
stream.groupBy((k, v) -> k).count();  // Repartitions!

// Use this instead
stream.groupByKey().count();  // No repartition

Pitfall 2: Too Many State Stores

// BAD: Separate aggregations
stream.groupByKey().count();
stream.groupByKey().reduce(...);
stream.groupByKey().aggregate(...);
// 3 state stores, 3 changelogs

// GOOD: Combined aggregation
stream.groupByKey().aggregate(...);  // 1 state store

Pitfall 3: Ignoring Co-Partitioning

// BAD: Join without checking partitions
// orders: 10 partitions
// customers: 8 partitions

orders.join(customers, ...);  // Fails! Not co-partitioned

// GOOD: Ensure same partition count
// Or let Kafka Streams repartition

Pitfall 4: Over-Partitioning Repartition Topics

// BAD: Too many partitions for repartition
stream.repartition(Repartitioned.as("repartition")
          .withNumberOfPartitions(1000))  // Overkill
      .groupByKey().count();

// GOOD: Match input topic or slightly higher
stream.repartition(Repartitioned.as("repartition")
          .withNumberOfPartitions(10))
      .groupByKey().count();

15. Mental Model Summary

Topology as a Graph

┌────────────────────────────────────────┐
│  Topology = Processing Graph           │
│                                        │
│  Nodes:                                │
│  ├─ Source (read Kafka)                │
│  ├─ Processor (transform)              │
│  └─ Sink (write Kafka)                 │
│                                        │
│  Edges:                                │
│  └─ Data flow between nodes            │
└────────────────────────────────────────┘

Internal Topics Purpose

REPARTITION TOPICS:
├─ Re-distribute data by new key
├─ Ensure co-partitioning
└─ Created automatically when needed

CHANGELOG TOPICS:
├─ Backup state stores
├─ Enable fault tolerance
└─ One per state store

Both managed by Kafka Streams

The Complete Flow

Input Topic
   ↓
[Source Node]
   ↓
[Processor: change key]
   ↓
[Sink → Repartition Topic] ← Kafka re-partitions
   ↓
[Source ← Repartition Topic]
   ↓
[Processor: stateful operation]
   ├─ State Store (local)
   └─ Changelog Topic (Kafka) ← Durability
   ↓
[Sink → Output Topic]

16. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What is a Kafka Streams topology?
✅ What are sub-topologies?
✅ When are repartition topics created?
✅ What's the difference between repartition and changelog topics?
✅ How do I inspect my topology?
✅ What is co-partitioning and why does it matter?
✅ How can I minimize repartitions?
✅ How are tasks and parallelism related?

17. What's Next?

Now that you understand topology design, the next module will cover:

• Module 8.1: Joins in Kafka Streams — stream-stream, stream-table, table-table joins
• Module 5.2: Windowing — how windows interact with topology
• Module 12.1: Monitoring — tracking topology performance

Summary

You learned:

• ✅ Topology = DAG of source, processor, and sink nodes
• ✅ Sub-topologies = independent sub-graphs with separate parallelism
• ✅ Repartition topics created when key changes (groupBy, joins)
• ✅ Changelog topics back up state stores (covered in Module 6.2)
• ✅ Co-partitioning required for joins and stateful operations
• ✅ Tasks = one per input partition (unit of parallelism)
• ✅ Use groupByKey() to avoid unnecessary repartitions
• ✅ Name state stores explicitly for better debugging
• ✅ Monitor internal topics for performance bottlenecks

Next up: Module 8.1 — Joins in Kafka Streams — Learn how to combine multiple streams and tables with different join types.