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

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

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

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

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.

Practice

1. What is the main purpose of error handling in Kafka streams?

easy

A. To delete all bad data permanently

B. To keep the stream running smoothly even if some data is bad

C. To stop the stream immediately when an error occurs

D. To ignore errors and continue processing without any checks

Solution

1. Step 1: Understand error handling goal

   Error handling aims to manage problems without stopping the whole process.

2. Step 2: Identify correct purpose in Kafka streams

   Kafka streams use error handling to keep processing even if some data is bad.

3. Final Answer:

   To keep the stream running smoothly even if some data is bad -> Option B

4. Quick Check:

   Error handling = keep stream running [OK]

Hint: Error handling means keep running despite bad data [OK]

Common Mistakes:

• Thinking error handling stops the stream
• Assuming bad data is deleted automatically
• Ignoring the need to handle errors

2. Which syntax correctly shows how to catch exceptions in a Kafka Streams processor in Java?

easy

A. catch (Exception e) { processRecord(); } try { handleError(e); }

B. processRecord() catch Exception e handleError(e)

C. try { processRecord(); } catch (Exception e) { handleError(e); }

D. try processRecord() catch Exception handleError()

Solution

1. Step 1: Recall Java try-catch syntax

   Java uses try { } catch (Exception e) { } to handle exceptions.

2. Step 2: Match correct syntax for Kafka Streams processor

   try { processRecord(); } catch (Exception e) { handleError(e); } correctly wraps processRecord() in try and catches Exception e to handle errors.

3. Final Answer:

   try { processRecord(); } catch (Exception e) { handleError(e); } -> Option C

4. Quick Check:

   Java try-catch syntax = try { processRecord(); } catch (Exception e) { handleError(e); } [OK]

Hint: Remember Java try-catch blocks use braces and catch after try [OK]

Common Mistakes:

• Placing catch before try
• Missing braces {} around try or catch blocks
• Using incorrect keywords or order

3. Given this Kafka Streams code snippet, what happens when a bad record causes an exception?

streams.foreach((key, value) -> {
  try {
    process(value);
  } catch (Exception e) {
    deadLetterTopic.send(key, value);
  }
});

medium

A. The exception crashes the whole application

B. The stream stops processing immediately

C. The bad record is ignored and lost

D. The bad record is sent to a dead-letter topic for later review

Solution

1. Step 1: Analyze try-catch block in code

   The code tries to process each record; if an exception occurs, it catches it.

2. Step 2: Check catch block action

   In catch, the bad record is sent to deadLetterTopic for later review.

3. Final Answer:

   The bad record is sent to a dead-letter topic for later review -> Option D

4. Quick Check:

   Exception caught = send to dead-letter topic [OK]

Hint: Catch exceptions and send bad data to dead-letter topic [OK]

Common Mistakes:

• Assuming stream stops on exception
• Thinking bad data is lost without handling
• Believing exception crashes app without catch

4. Identify the error in this Kafka Streams error handling code:

streams.foreach((key, value) -> {
  try {
    process(value);
  } catch Exception e {
    log.error(e.getMessage());
  }
});

medium

A. Missing parentheses around Exception e in catch block

B. Using foreach instead of map for error handling

C. Logging error message is incorrect

D. Try block should be outside the foreach

Solution

1. Step 1: Check catch block syntax

   In Java, catch must have parentheses around the exception type and variable.

2. Step 2: Identify missing parentheses

   The code has 'catch Exception e' missing parentheses; correct is 'catch (Exception e)'.

3. Final Answer:

   Missing parentheses around Exception e in catch block -> Option A

4. Quick Check:

   Catch syntax requires parentheses [OK]

Hint: Catch blocks always need parentheses around exception [OK]

Common Mistakes:

• Omitting parentheses in catch
• Confusing foreach and map usage
• Thinking logging method is wrong

5. You want to build a Kafka Streams application that processes user events but sends any malformed events to a separate topic without stopping the main stream. Which approach best achieves this?

hard

A. Use a try-catch inside the stream processor to send bad events to a dead-letter topic

B. Stop the stream when a malformed event is detected to fix data immediately

C. Ignore malformed events and continue processing without logging

D. Filter out malformed events before processing without saving them

Solution

1. Step 1: Understand requirement to keep stream running

   The stream must continue processing even if some events are bad.

2. Step 2: Choose method to handle bad events

   Using try-catch to catch errors and send bad events to a dead-letter topic allows this.

3. Final Answer:

   Use a try-catch inside the stream processor to send bad events to a dead-letter topic -> Option A

4. Quick Check:

   Try-catch + dead-letter topic = best error handling [OK]

Hint: Try-catch plus dead-letter topic keeps stream healthy [OK]

Common Mistakes:

• Stopping stream on error instead of handling
• Ignoring bad data without saving for review
• Filtering bad data without storing it