Filter and map operations in Kafka - Step-by-Step Execution

Process Flow - Filter and map operations

Start with stream of messages

↓

Apply filter condition

↓

Keep message

↓

Apply map function

↓

Output transformed messages

Messages flow through a filter to keep only those matching a condition, then a map transforms each kept message.

Execution Sample

Kafka

stream.filter((key, value) -> value.toLowerCase().contains("error"))
      .mapValues(value -> value.toUpperCase())
      .foreach((key, value) -> System.out.println(value));

This code filters messages containing 'error' and converts their values to uppercase before printing.

Process Table

Step	Message Value	Filter Condition (contains 'error')	Filter Result	Map Operation (toUpperCase)	Output
1	System started	false	Discard	-	-
2	Error detected	true	Keep	ERROR DETECTED	ERROR DETECTED
3	User login	false	Discard	-	-
4	Critical error	true	Keep	CRITICAL ERROR	CRITICAL ERROR
5	Shutdown	false	Discard	-	-

💡 All messages processed; only those with 'error' passed filter and were mapped.

Status Tracker

Variable	Start	After 1	After 2	After 3	After 4	After 5	Final
Current Message	-	System started	Error detected	User login	Critical error	Shutdown	-
Filter Result	-	false	true	false	true	false	-
Mapped Value	-	-	ERROR DETECTED	-	CRITICAL ERROR	-	-
Output	-	-	ERROR DETECTED	-	CRITICAL ERROR	-	-

Key Moments - 2 Insights

Why are some messages not transformed or printed?

Messages failing the filter condition are discarded early (see rows 1,3,5 in execution_table), so map and output steps do not run for them.

Does the map operation change the original message?

No, map creates a new transformed value (uppercase string) without modifying the original message, as shown in the 'Mapped Value' column.

Visual Quiz - 3 Questions

Look at the execution table, what is the output at step 4?

A Critical error

B CRITICAL ERROR

C critical error

D -

💡 Hint

Check the 'Output' column at step 4 in the execution_table.

At which step does the filter condition become false for the first time?

A Step 3

B Step 2

C Step 1

D Step 5

💡 Hint

Look at the 'Filter Result' column in execution_table for the earliest 'false'.

If the filter condition changed to check for 'warning' instead of 'error', which message would be kept at step 2?

A -

B System started

C User login

D Error detected

💡 Hint

Refer to the original message values and see if 'warning' appears in step 2 message.

Concept Snapshot

Filter and map in Kafka streams:
- filter(condition): keeps messages matching condition
- map(function): transforms each kept message
- Order matters: filter first, then map
- Messages failing filter are dropped
- Map creates new transformed output

Full Transcript

In Kafka streams, messages flow through operations like filter and map. First, filter checks each message's value to keep only those containing 'error'. Messages not matching are discarded immediately. Then, map transforms each kept message by converting its value to uppercase. The output shows only transformed messages that passed the filter. This step-by-step trace shows how messages are processed, filtered, transformed, and output. Understanding this flow helps beginners see why some messages are dropped and how transformations apply only after filtering.

Practice

1. What does the filter operation do in Kafka Streams?

easy

A. Deletes all messages in the stream

B. Changes messages to a new form

C. Keeps only messages that meet a condition

D. Sends messages to a different topic

Solution

1. Step 1: Understand filter purpose

   The filter operation selects messages that satisfy a given condition, removing others.

2. Step 2: Compare with other operations

   Unlike map, which changes message content, filter only keeps or drops messages.

3. Final Answer:

   Keeps only messages that meet a condition -> Option C

4. Quick Check:

   Filter = Keep messages by condition [OK]

Hint: Filter keeps or drops messages based on a rule [OK]

Common Mistakes:

• Confusing filter with map operation
• Thinking filter changes message content
• Assuming filter deletes all messages

2. Which of the following is the correct syntax to apply a map operation on a Kafka stream named stream to convert messages to uppercase?

easy

A. stream.map(value.toUpperCase)

B. stream.filter(value -> value.toUpperCase())

C. stream.map(value => value.toUpperCase())

D. stream.map(value -> value.toUpperCase())

Solution

1. Step 1: Identify correct lambda syntax in Kafka Streams

   Kafka Streams in Java uses map(value -> ...) syntax for mapping values.

2. Step 2: Check method call correctness

   value.toUpperCase() correctly converts string to uppercase.

3. Final Answer:

   stream.map(value -> value.toUpperCase()) -> Option D

4. Quick Check:

   Correct lambda and method call = stream.map(value -> value.toUpperCase()) [OK]

Hint: Use arrow -> for lambda in Kafka Streams Java [OK]

Common Mistakes:

• Using filter instead of map
• Using JavaScript arrow => instead of Java ->
• Missing parentheses in method call

3. Given the following Kafka Streams code, what will be the output messages after processing?

stream.filter(value -> value.length() > 3)
      .map(value -> value.substring(0, 3))

Input messages: ["apple", "cat", "banana", "dog"]

medium

A. ["cat", "dog"]

B. ["app", "ban"]

C. ["apple", "banana"]

D. ["app", "cat", "ban", "dog"]

Solution

1. Step 1: Apply filter condition

   Keep messages with length > 3: "apple" (5), "banana" (6). "cat" and "dog" are dropped.

2. Step 2: Apply map operation

   Take first 3 characters: "apple" -> "app", "banana" -> "ban".

3. Final Answer:

   ["app", "ban"] -> Option B

4. Quick Check:

   Filter then map = ["app", "ban"] [OK]

Hint: Filter first, then map to transform filtered messages [OK]

Common Mistakes:

• Not filtering before mapping
• Mapping before filtering
• Including messages with length ≤ 3

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

stream.filter(value -> value.startsWith("a"))
      .map(value -> value.toUpperCase())
      .filter(value -> value.length() > 5)

medium

A. No error; code is correct

B. Using two filter operations is not allowed

C. The second filter uses the mapped uppercase value correctly

D. The lambda syntax is incorrect

Solution

1. Step 1: Check filter and map chaining

   Chaining multiple filters and maps is allowed in Kafka Streams.

2. Step 2: Verify lambda syntax and logic

   Lambdas use correct syntax and logic: first filter checks start, map uppercases, second filter checks length.

3. Final Answer:

   No error; code is correct -> Option A

4. Quick Check:

   Multiple filters and maps allowed = No error; code is correct [OK]

Hint: Multiple filters and maps can be chained without error [OK]

Common Mistakes:

• Thinking multiple filters are disallowed
• Confusing lambda syntax with other languages
• Assuming map changes variable type incorrectly

5. You want to process a Kafka stream of user names to keep only those starting with 'J' and convert them to lowercase. Which code snippet correctly does this?

hard

A. stream.filter(name -> name.startsWith("J"))
.map(name -> name.toLowerCase())

B. stream.map(name -> name.toLowerCase())
.filter(name -> name.startsWith("J"))

C. stream.filter(name -> name.toLowerCase().startsWith("j"))
.map(name -> name.toUpperCase())

D. stream.map(name -> name.startsWith("J"))
.filter(name -> name.toLowerCase())

Solution

1. Step 1: Filter names starting with 'J'

   Use filter(name -> name.startsWith("J")) to keep only names starting with uppercase 'J'.

2. Step 2: Map filtered names to lowercase

   Apply map(name -> name.toLowerCase()) to convert names to lowercase.

3. Final Answer:

   stream.filter(name -> name.startsWith("J")) .map(name -> name.toLowerCase()) -> Option A

4. Quick Check:

   Filter first, then map to lowercase [OK]

Hint: Filter first by condition, then map to transform [OK]

Common Mistakes:

• Mapping before filtering causing wrong results
• Using incorrect lambda syntax
• Filtering after changing case causing misses