Error handling in streams in Kafka - Mini Project: Build & Apply

Error handling in streams

📖 Scenario: You are building a Kafka stream application that processes user events. Sometimes, the events might have missing or incorrect data. You want to handle these errors gracefully so the stream keeps running without crashing.

🎯 Goal: Build a Kafka Streams application that reads from an input topic, checks for errors in the data, sends bad records to a dead-letter topic, and processes good records normally.

📋 What You'll Learn

Create a Kafka Streams topology with an input topic named user-events

Create a dead-letter topic named user-events-dead-letter

Add a configuration variable maxRetries to limit retry attempts

Implement error handling logic to detect invalid events

Send invalid events to the dead-letter topic

Print valid events to the console

💡 Why This Matters

🌍 Real World

In real applications, data streams can have bad or incomplete data. Handling errors without stopping the stream is important to keep the system reliable.

💼 Career

Kafka Streams developers and data engineers often need to build robust pipelines that handle errors gracefully and keep data flowing smoothly.

1

Set up the Kafka Streams input topic and basic stream

Create a Kafka Streams StreamsBuilder instance called builder. Use it to create a stream from the input topic named user-events and assign it to a variable called userEventsStream.

Kafka

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
from kafka import KafkaAdminClient
from kafka.admin import NewTopic
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka import KafkaStreams

# Your code here

Need a hint?

Use StreamsBuilder() to create the builder and builder.stream('user-events') to create the stream.

2

Add a configuration variable for maximum retries

Create an integer variable called maxRetries and set it to 3 to limit the number of retry attempts for processing events.

Kafka

from kafka import KafkaStreams

builder = KafkaStreams.StreamsBuilder()
userEventsStream = builder.stream('user-events')

# Your code here

Need a hint?

Just create a variable named maxRetries and assign it the value 3.

3

Implement error handling logic in the stream

Use userEventsStream and apply a map function with variables key and value. Inside the function, check if value contains the key "userId". If it does not, return a tuple with key and None to mark it as invalid. Otherwise, return the original key and value.

Kafka

from kafka import KafkaStreams

builder = KafkaStreams.StreamsBuilder()
userEventsStream = builder.stream('user-events')

maxRetries = 3

# Your code here

Need a hint?

Define a function validate_event that checks if value has 'userId'. Use map on userEventsStream with this function.

4

Send invalid events to dead-letter topic and print valid events

Use validatedStream and branch it into two streams: one where value is None (invalid events) and one where value is not None (valid events). Send invalid events to the topic user-events-dead-letter using to(). Print valid events using foreach() with a function that prints key and value.

Kafka

from kafka import KafkaStreams

builder = KafkaStreams.StreamsBuilder()
userEventsStream = builder.stream('user-events')

maxRetries = 3

def validate_event(key, value):
    if 'userId' not in value:
        return (key, None)
    return (key, value)

validatedStream = userEventsStream.map(validate_event)

# Your code here

Need a hint?

Use branch() to split the stream by checking if value is None. Send invalid events to user-events-dead-letter with to(). Use foreach() on valid events to print them.