3.1 Building a Kafka Streams Application

Module Overview

Goal: Teach application structure, not processing logic.

What You'll Learn:

• StreamsBuilder for topology creation
• Topology structure (high level)
• Essential properties and bootstrap configs
• Serdes (Serializers/Deserializers) conceptual understanding
• Application lifecycle: starting and shutting down
• Basic application template

What's NOT Covered Here:

• Stateless/stateful operators (covered in Modules 4 and 5)
• Performance tuning (covered in Module 13.1)
• Error handling (covered in Module 10.1)
• State stores (covered in Module 6.1)

Key Insight: This module teaches you the scaffolding of a Streams app. The processing logic comes in later modules.

1. Project Setup & Dependencies

Maven Dependency

Before writing any code, add the Kafka Streams dependency to your project:

pom.xml (Maven):

<dependencies>
    <!-- Kafka Streams -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>3.6.0</version>
    </dependency>
    
    <!-- SLF4J for logging -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>2.0.9</version>
    </dependency>
    
    <!-- Optional: JSON Serialization with Jackson -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
        <version>2.15.2</version>
    </dependency>
</dependencies>

build.gradle (Gradle):

dependencies {
    implementation 'org.apache.kafka:kafka-streams:3.6.0'
    implementation 'org.slf4j:slf4j-simple:2.0.9'
    
    // Optional: JSON
    implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'
}

Important: Use the same Kafka version across all clients (producer, consumer, streams) to avoid compatibility issues.

CLI Testing: kafka-console-producer & kafka-console-consumer

For quick testing during development, use Kafka's CLI tools:

Produce test data:

# Send test messages to input topic
kafka-console-producer --bootstrap-server localhost:9092 --topic input-topic \
  --property "key.separator=:" \
  --property "parse.key=true"

# Type messages (key:value format):
user1:{"action":"login","timestamp":1234567890}
user2:{"action":"click","timestamp":1234567891}

Consume output:

# Read from output topic
kafka-console-consumer --bootstrap-server localhost:9092 --topic output-topic \
  --from-beginning \
  --property "print.key=true" \
  --property "key.separator= => "

# You should see:
# user1 => {"action":"login","timestamp":1234567890}
# user2 => {"action":"click","timestamp":1234567891}

Debugging Internal Topics:

# List all topics (including internal Kafka Streams topics)
kafka-topics --bootstrap-server localhost:9092 --list

# Expected internal topics:
# <application-id>-<source-changelog>-changelog
# <application-id>-KSTREAM-AGGREGATE-STATE-STORE-0000000001-repartition

2. The Basic Structure

Five Components

Every Kafka Streams application has five core components:

1. PROPERTIES
   └─ Configure connection, behavior

2. STREAMSBUILDER
   └─ Define processing topology

3. TOPOLOGY
   └─ Compiled processing graph

4. KAFKASTREAMS
   └─ Runtime engine

5. LIFECYCLE MANAGEMENT
   └─ Start, stop, error handling

Minimal Application Template

public class MinimalStreamsApp {
    
    public static void main(String[] args) {
        // 1. Properties
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        
        // 2. StreamsBuilder
        StreamsBuilder builder = new StreamsBuilder();
        
        // Define topology (processing logic)
        KStream<String, String> input = builder.stream("input-topic");
        input.to("output-topic");
        
        // 3. Topology
        Topology topology = builder.build();
        
        // 4. KafkaStreams
        KafkaStreams streams = new KafkaStreams(topology, props);
        
        // 5. Lifecycle
        streams.start();
        
        // Graceful shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

This is the foundation. Everything else adds processing logic between lines 11-12.

2. Properties Configuration

Essential Properties

Three properties are required:

Properties props = new Properties();

// 1. Application ID (required)
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");

// 2. Bootstrap servers (required)
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

// 3. Default serdes (required for most cases)
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, 
    Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, 
    Serdes.String().getClass());

Application ID

props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");

What it does:

• Identifies your Streams application
• Used as Kafka consumer group ID
• Prefixes internal topics (changelog, repartition)
• Must be unique per application

Example:

application.id = "order-processor"
├─ Consumer group: "order-processor"
├─ Changelog topic: "order-processor-KSTREAM-AGGREGATE-STATE-STORE-0000000001-changelog"
└─ Repartition topic: "order-processor-KSTREAM-AGGREGATE-STATE-STORE-0000000001-repartition"

Important: Changing application.id = new application (new state, new offsets).

Bootstrap Servers

props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");

What it does:

• Initial connection to Kafka cluster
• Same as consumer/producer bootstrap

Best practices:

• Provide multiple brokers for high availability
• Use DNS or load balancer in production

Default Serdes

props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

What are Serdes?

• Serializer + Deserializer = Serde
• Converts between objects and bytes
• Required for reading from and writing to Kafka

More on serdes in Section 4.

Common Optional Properties

// Processing guarantee
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
    StreamsConfig.EXACTLY_ONCE_V2);

// Number of threads per instance
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);

// State directory (where local state stored)
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/kafka-streams");

// Commit interval (how often to commit offsets)
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30000); // 30 seconds

// Cache size (for aggregations)
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024); // 10 MB

Note: Performance tuning is covered in Module 13.1.

3. StreamsBuilder: Defining the Topology

What is a Topology?

A topology is the processing graph: sources → processors → sinks.

Source → Processor → Processor → Sink
 (read)   (transform)  (aggregate)  (write)

StreamsBuilder is the high-level DSL for building topologies.

Creating a StreamsBuilder

StreamsBuilder builder = new StreamsBuilder();

StreamsBuilder provides:

• builder.stream() — create KStream from topic
• builder.table() — create KTable from topic
• builder.globalTable() — create GlobalKTable from topic

Reading from Topics

// Read KStream from topic
KStream<String, String> orders = builder.stream("orders");

// Read KTable from topic
KTable<String, String> users = builder.table("users");

// Read GlobalKTable from topic
GlobalKTable<String, String> products = builder.globalTable("products");

Key difference:

• stream() — every record is independent event
• table() — latest value per key (changelog)
• globalTable() — fully replicated table

(Covered in detail in Module 2.1)

Simple Transformation Example

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> input = builder.stream("input-topic");

// Transform: uppercase all values
KStream<String, String> uppercased = input.mapValues(value -> value.toUpperCase());

// Write to output topic
uppercased.to("output-topic");

Processing flow:

"input-topic" → mapValues(uppercase) → "output-topic"

Building the Topology

// Define processing with StreamsBuilder
StreamsBuilder builder = new StreamsBuilder();
builder.stream("input").to("output");

// Build topology
Topology topology = builder.build();

// Optional: print topology for debugging
System.out.println(topology.describe());

Topology output:

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

4. Serdes: Serialization and Deserialization

What are Serdes?

SERIALIZATION:
Java Object → Bytes → Kafka Topic

DESERIALIZATION:
Kafka Topic → Bytes → Java Object

SERDE:
Handles both directions

Analogy: Think of Serdes like writing a letter:

• Serialization = Writing your thoughts (Java object) into text on paper (bytes) so it can be mailed (sent to Kafka)
• Deserialization = Reading the letter (bytes from Kafka) and understanding it (converting back to Java object)
• Serde = Your ability to both write and read the letter

Without Serdes, Kafka only sees raw bytes — meaningless without instructions on how to interpret them.

Why Serdes Matter

Problem: Kafka topics store bytes, not objects.

Your Application:
Order order = new Order("123", 99.50);  // Java object

Kafka Topic:
[0x7B, 0x22, 0x69, 0x64, ...]  // Just bytes!

Another Application:
??? (How do I read this?)

Solution: Serdes provide the contract:

Producer (Serializer):
Order → {"id":"123","amount":99.50} → [bytes] → Kafka

Consumer (Deserializer):
Kafka → [bytes] → {"id":"123","amount":99.50} → Order

Critical: Producer and consumer must use compatible serdes, or deserialization will fail!

Built-In Serdes

Kafka Streams provides serdes for common types:

import org.apache.kafka.common.serialization.Serdes;

// String
Serdes.String()

// Integer
Serdes.Integer()

// Long
Serdes.Long()

// Double
Serdes.Double()

// Bytes
Serdes.ByteArray()

// UUID
Serdes.UUID()

Configuring Serdes

Default Serdes (Global)

// Set once for entire application
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, 
    Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, 
    Serdes.String().getClass());

All streams/tables use these by default.

Per-Stream Serdes (Override Default)

// Override default for specific stream
KStream<String, Long> stream = builder.stream(
    "my-topic",
    Consumed.with(Serdes.String(), Serdes.Long())
);

Per-Operation Serdes

// Override for specific operation
stream.to(
    "output-topic",
    Produced.with(Serdes.String(), Serdes.Long())
);

Custom Serdes (JSON Example)

For custom objects, you'll often use JSON:

Option 1: Jackson JSON Serde (Simple, Lightweight)

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;

public class JsonSerde<T> implements Serializer<T>, Deserializer<T> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    private Class<T> clazz;
    
    public JsonSerde(Class<T> clazz) {
        this.clazz = clazz;
    }
    
    @Override
    public byte[] serialize(String topic, T data) {
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Failed to serialize", e);
        }
    }
    
    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            return objectMapper.readValue(data, clazz);
        } catch (Exception e) {
            throw new RuntimeException("Failed to deserialize", e);
        }
    }
}

// Usage:
Serde<Order> orderSerde = Serdes.serdeFrom(
    new JsonSerde<>(Order.class), 
    new JsonSerde<>(Order.class)
);

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

Option 2: Confluent JSON Serde (Schema Registry)

import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerde;

// Define custom serde for Order class
Serde<Order> orderSerde = new KafkaJsonSchemaSerde<>();

// Configure it
Map<String, Object> serdeProps = new HashMap<>();
serdeProps.put("json.value.type", Order.class);
serdeProps.put("schema.registry.url", "http://localhost:8081");
orderSerde.configure(serdeProps, false); // false = value serde

// Use in stream
KStream<String, Order> orders = builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde)
);

Option 3: Avro Serde (Schema Evolution)

import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;

// Avro requires Schema Registry
Map<String, String> serdeConfig = Map.of(
    "schema.registry.url", "http://localhost:8081"
);

// Create Avro serde for generated Avro class
Serde<OrderAvro> orderSerde = new SpecificAvroSerde<>();
orderSerde.configure(serdeConfig, false);

KStream<String, OrderAvro> orders = builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde)
);

Comparison:

┌───────────┬───────────────┬─────────────────┬────────────────┐
│ Format    │ Schema        │ Backward        │ Performance    │
│           │ Evolution     │ Compatibility   │                │
├───────────┼───────────────┼─────────────────┼────────────────┤
│ String    │ None          │ Manual only     │ Fast           │
│ JSON      │ Flexible      │ Loose           │ Medium         │
│ Avro      │ Built-in      │ Strong          │ Fast (binary)  │
│ Protobuf  │ Built-in      │ Strong          │ Fast (binary)  │
└───────────┴───────────────┴─────────────────┴────────────────┘

Recommendation:

• Development/Testing: String or JSON (simple)
• Production: Avro or Protobuf (schema evolution + performance)

Note: Serialization is covered in more depth in Producer Module 3.1. Here, we just need the basics.

5. KafkaStreams: The Runtime Engine

Creating KafkaStreams Instance

Topology topology = builder.build();
Properties props = /* ... */;

// Create Kafka Streams instance
KafkaStreams streams = new KafkaStreams(topology, props);

KafkaStreams is the runtime that:

• Connects to Kafka
• Assigns tasks to threads
• Processes records
• Manages state
• Handles failures

State Listener (Optional but Useful)

streams.setStateListener((newState, oldState) -> {
    System.out.println("State transition: " + oldState + " -> " + newState);
});

States:

• CREATED — topology defined, not started
• REBALANCING — assigning tasks
• RUNNING — actively processing
• PENDING_SHUTDOWN — cleaning up
• NOT_RUNNING — stopped

Uncaught Exception Handler (Recommended)

streams.setUncaughtExceptionHandler((thread, exception) -> {
    System.err.println("Uncaught exception in thread " + thread.getName());
    exception.printStackTrace();
    return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
});

Response options:

• SHUTDOWN_APPLICATION — stop entire app
• SHUTDOWN_CLIENT — stop this instance
• REPLACE_THREAD — restart failed thread

6. Application Lifecycle

Starting the Application

streams.start();

What happens:

1. Connect to Kafka brokers
2. Join consumer group (using application.id)
3. Subscribe to input topics
4. Assign tasks to threads
5. Begin processing records

Note: start() is non-blocking. Processing happens in background threads.

Graceful Shutdown

// Add shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    System.out.println("Shutting down Kafka Streams...");
    streams.close();
    System.out.println("Shutdown complete");
}));

What close() does:

1. Stop processing new records
2. Finish processing in-flight records
3. Commit offsets
4. Flush state stores
5. Leave consumer group
6. Close resources

Timeout:

// Close with timeout
streams.close(Duration.ofSeconds(30));

Complete Lifecycle Example

public class StreamsApp {
    public static void main(String[] args) {
        Properties props = getProperties();
        Topology topology = buildTopology();
        
        KafkaStreams streams = new KafkaStreams(topology, props);
        
        // State listener
        streams.setStateListener((newState, oldState) -> {
            log.info("State change: {} -> {}", oldState, newState);
        });
        
        // Exception handler
        streams.setUncaughtExceptionHandler((thread, exception) -> {
            log.error("Uncaught exception", exception);
            return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
        });
        
        // Shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("Shutting down gracefully...");
            streams.close(Duration.ofSeconds(30));
        }));
        
        // Start
        streams.start();
        log.info("Streams application started");
    }
    
    private static Properties getProperties() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return props;
    }
    
    private static Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();
        builder.stream("input").to("output");
        return builder.build();
    }
}

7. Running Your Application

As Standalone JAR

# Package as JAR
mvn clean package

# Run
java -jar my-streams-app.jar

With Docker

FROM openjdk:11-jre-slim
COPY target/my-streams-app.jar /app/app.jar
ENTRYPOINT ["java", "-jar", "/app/app.jar"]

docker build -t my-streams-app .
docker run -e BOOTSTRAP_SERVERS=kafka:9092 my-streams-app

With Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: streams-app
spec:
  replicas: 3  # Scale out
  selector:
    matchLabels:
      app: streams-app
  template:
    metadata:
      labels:
        app: streams-app
    spec:
      containers:
      - name: streams-app
        image: my-streams-app:latest
        env:
        - name: BOOTSTRAP_SERVERS
          value: "kafka:9092"
        - name: APPLICATION_ID
          value: "my-streams-app"

Key point: Streams apps are stateless deployments (state is in Kafka).

8. Scaling Your Application

Horizontal Scaling

More instances = more parallelism (up to partition count)

Topic with 6 partitions:

1 instance:  All 6 partitions
2 instances: 3 partitions each
3 instances: 2 partitions each
6 instances: 1 partition each
7 instances: 1 idle (max = partition count)

Adding Instances

# Instance 1
java -jar streams-app.jar

# Instance 2 (same application.id!)
java -jar streams-app.jar

# Instance 3
java -jar streams-app.jar

# Kafka Streams automatically rebalances tasks

Requirements:

• Same application.id
• Same topology
• Access to Kafka cluster

Scaling Best Practices

1. Partition count = max parallelism
   └─ 6 partitions → max 6 instances

2. Over-partition topics
   └─ More partitions → more scaling headroom
   └─ Example: 30 partitions for future growth

3. Monitor task distribution
   └─ Ensure even load across instances

4. Use stateful scaling carefully
   └─ State migration takes time (covered in Module 6.3)

9. Development Tips

Print Topology for Debugging

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

Output shows:

• Sources (input topics)
• Processors (transformations)
• Sinks (output topics)
• Internal topics

Local Development Configuration

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "dev-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

// Useful for development
props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); // Disable caching
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); // Commit every second

Logging

<!-- log4j2.xml -->
<Configuration>
    <Loggers>
        <Logger name="org.apache.kafka.streams" level="INFO"/>
        <Logger name="org.apache.kafka.clients" level="WARN"/>
        <Root level="INFO">
            <AppenderRef ref="Console"/>
        </Root>
    </Loggers>
</Configuration>

10. Common Patterns

Pattern 1: Simple Pass-Through

builder.stream("input").to("output");

Use case: Topic-to-topic copy, testing

Pattern 2: Filter and Forward

builder.stream("orders")
    .filter((key, order) -> order.getAmount() > 1000)
    .to("high-value-orders");

Use case: Routing, conditional forwarding

Pattern 3: Transform and Write

builder.stream("raw-data")
    .mapValues(value -> transform(value))
    .to("processed-data");

Use case: ETL, data transformation

Pattern 4: Multiple Outputs

KStream<String, Order> orders = builder.stream("orders");

// Output 1: High value
orders.filter((k, v) -> v.getAmount() > 1000)
    .to("high-value-orders");

// Output 2: Low value
orders.filter((k, v) -> v.getAmount() <= 1000)
    .to("low-value-orders");

Use case: Branching, routing to multiple destinations

11. Mental Model Summary

Application Structure

┌─────────────────────────────────────────┐
│  PROPERTIES                             │
│  ├─ application.id                      │
│  ├─ bootstrap.servers                   │
│  └─ default serdes                      │
└─────────────────────────────────────────┘
            ↓
┌─────────────────────────────────────────┐
│  STREAMSBUILDER                         │
│  ├─ Define sources                      │
│  ├─ Define processors                   │
│  └─ Define sinks                        │
└─────────────────────────────────────────┘
            ↓
┌─────────────────────────────────────────┐
│  TOPOLOGY                               │
│  └─ Compiled processing graph           │
└─────────────────────────────────────────┘
            ↓
┌─────────────────────────────────────────┐
│  KAFKASTREAMS                           │
│  └─ Runtime engine                      │
└─────────────────────────────────────────┘
            ↓
┌─────────────────────────────────────────┐
│  LIFECYCLE                              │
│  ├─ start()                             │
│  ├─ [processing...]                     │
│  └─ close()                             │
└─────────────────────────────────────────┘

12. Common Pitfalls & Troubleshooting

Pitfall 1: Serialization Mismatch

Problem: Producer writes JSON, consumer expects Avro (or vice versa).

Error: org.apache.kafka.common.errors.SerializationException: 
  Error deserializing key/value for partition my-topic-0 at offset 42

Solution:

• Ensure producer and consumer use compatible serdes
• Use Schema Registry for enforced compatibility
• Log serialization exceptions and send to Dead Letter Topic (DLT)

// Wrong: Mismatch
// Producer: Serdes.String()
// Consumer: Serdes.Long()  ❌

// Correct: Match
// Producer: Serdes.String()
// Consumer: Serdes.String()  ✅

Pitfall 2: Application ID Collision

Problem: Two different Streams apps share the same application.id.

Result:
├─ Both apps share consumer group
├─ Both read from same internal topics
└─ State corruption and processing errors

Solution:

• Use unique application.id per logical app
• Never reuse application IDs across different apps
• Use semantic names: payment-processor, order-enrichment

// Wrong:
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");  // Generic!

// Correct:
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "payment-processor-v1");  // Specific!

Pitfall 3: Broker Connection Issues

Problem: Cannot connect to Kafka brokers.

Error: org.apache.kafka.common.errors.TimeoutException: 
  Failed to update metadata after 60000 ms

Solution:

• Verify bootstrap.servers is correct
• Check network connectivity: telnet localhost 9092
• Ensure Kafka brokers are running: kafka-topics --list
• Check firewall rules and security groups

# Test broker connectivity
nc -zv localhost 9092
# Connection to localhost port 9092 [tcp/*] succeeded!

Pitfall 4: State Directory Permissions

Problem: Kafka Streams cannot write to state directory.

Error: java.io.IOException: Failed to create state directory

Solution:

• Ensure application has write permissions to /tmp/kafka-streams (default)
• Or specify custom directory with proper permissions:

props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/kafka-streams");

# Fix permissions
sudo mkdir -p /var/kafka-streams
sudo chown myapp:myapp /var/kafka-streams
sudo chmod 755 /var/kafka-streams

Pitfall 5: Not Handling Exceptions

Problem: Streams app crashes on first exception.

Solution:

• Add UncaughtExceptionHandler for thread-level exceptions
• Add deserialization exception handler

streams.setUncaughtExceptionHandler((thread, exception) -> {
    log.error("Uncaught exception in thread {}", thread.getName(), exception);
    // Return REPLACE_THREAD to restart the failed thread
    return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
});

// Handle deserialization errors
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
    LogAndContinueExceptionHandler.class);

Pitfall 6: Ignoring State During Redeployment

Problem: Redeploying app with new application.id loses all state.

Solution:

• Keep the same application.id if logic hasn't fundamentally changed
• If you MUST change ID, use kafka-streams-application-reset tool to clean old state
• Or manually delete consumer group and internal topics

# Reset state (USE WITH CAUTION)
kafka-streams-application-reset --application-id payment-processor \
  --input-topics orders,payments \
  --bootstrap-servers localhost:9092

Debugging Checklist

✅ Are broker addresses correct?
✅ Are topics created? (kafka-topics --list)
✅ Are serdes compatible?
✅ Is application.id unique?
✅ Are there exceptions in logs?
✅ Is state directory writable?
✅ Are consumer offsets committed?
✅ Are internal topics created?
✅ Are there lags in consumer groups?

13. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What are the five core components of a Streams app?
✅ What is the application.id and why is it important?
✅ What are serdes and why do we need them?
✅ How do I start and stop a Streams application?
✅ How does horizontal scaling work?
✅ What is the difference between stream(), table(), and globalTable()?
✅ How do I gracefully shut down a Streams app?

13. What's Next?

Now that you understand application structure, the next modules will teach you processing logic:

• Module 4.1: Stateless operations — filter, map, transform
• Module 5: Stateful operations — aggregations, grouping
• Module 6: State stores — how state is managed
• Module 10.1: Error handling — making your app production-ready

Summary

You learned:

• ✅ A Streams app has five components: properties, builder, topology, streams instance, lifecycle
• ✅ Three required properties: application.id, bootstrap.servers, default serdes
• ✅ StreamsBuilder defines the processing topology using DSL
• ✅ Serdes handle serialization/deserialization between objects and bytes
• ✅ KafkaStreams is the runtime engine that processes records
• ✅ Lifecycle management: start with start(), stop with close()
• ✅ Scaling is horizontal: add more instances (up to partition count)

Next up: Module 4.1 — Stateless Operations — Learn how to transform streams without maintaining state.