Filter and map operations in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Kafka Streams: What is the output of this filter and map operation?

Consider a Kafka Streams application that processes a stream of integers. It filters out even numbers and then maps each remaining number to its square. What will be the output for the input stream [1, 2, 3, 4, 5]?

Kafka

KStream<Integer, Integer> input = builder.stream("input-topic");
KStream<Integer, Integer> processed = input
    .filter((key, value) -> value % 2 != 0)
    .mapValues(value -> value * value);
processed.to("output-topic");

A [1, 4, 9, 16, 25]

B [1, 9, 25]

C [2, 4]

D [1, 3, 5]

💡 Hint

Remember that filter removes even numbers, and mapValues squares the remaining numbers.

The filter keeps only odd numbers: 1, 3, 5. Then mapValues squares each: 1*1=1, 3*3=9, 5*5=25. So the output is [1, 9, 25].

❓ Predict Output

intermediate

Kafka Streams: Result of chaining filter and map with keys

Given a Kafka Streams pipeline that filters records with keys less than 3 and maps values by adding 10, what is the output for input records with keys and values: (1, 5), (2, 7), (3, 9), (4, 11)?

Kafka

KStream<Integer, Integer> input = builder.stream("input-topic");
KStream<Integer, Integer> result = input
    .filter((key, value) -> key < 3)
    .mapValues(value -> value + 10);
result.to("output-topic");

A [(1, 15), (2, 17)]

B [(3, 19), (4, 21)]

C [(1, 5), (2, 7)]

D [(1, 15), (2, 17), (3, 19), (4, 21)]

💡 Hint

Filter keeps keys less than 3, then values get 10 added.

Only records with keys 1 and 2 pass the filter. Their values 5 and 7 become 15 and 17 after adding 10.

🔧 Debug

advanced

Why does this Kafka Streams filter-map code cause a runtime error?

Examine the following Kafka Streams code snippet. It attempts to filter and map values but causes a runtime NullPointerException. What is the cause?

Kafka

KStream<Integer, String> input = builder.stream("input-topic");
KStream<Integer, String> result = input
    .filter((key, value) -> value.length() > 3)
    .mapValues(value -> value.toUpperCase());
result.to("output-topic");

A The stream is missing a call to selectKey before filter, causing runtime failure.

B The mapValues function is incorrectly used and causes a syntax error.

C The filter predicate should use key instead of value, causing a type error.

D Some values in the stream are null, causing value.length() to throw NullPointerException.

💡 Hint

Think about what happens if value is null when calling value.length().

If any record has a null value, calling value.length() causes a NullPointerException at runtime.

📝 Syntax

advanced

Which Kafka Streams filter-map code snippet is syntactically correct?

Choose the syntactically correct Kafka Streams code that filters values greater than 10 and maps values to their double.

A input.filter((k, v) -> { v > 10 }).mapValues(v -> v * 2);

B input.filter(k, v -> v > 10).mapValues(v -> v * 2);

C input.filter((k, v) -> v > 10).mapValues(v -> v * 2);

D input.filter((k, v) => v > 10).mapValues(v -> v * 2);

💡 Hint

Check the syntax of lambda expressions in Java.

Option C uses correct Java lambda syntax. Option C misses parentheses around parameters. Option C uses JavaScript arrow syntax. Option C uses braces without return statement causing no boolean result.

🚀 Application

expert

How many records are in the output after this Kafka Streams filter and map chain?

A Kafka Streams application reads from a topic with 6 records having values: [5, 12, 7, 20, 3, 15]. It filters values greater than 10 and then maps values to value minus 5. How many records are in the output topic?

Kafka

KStream<Integer, Integer> input = builder.stream("input-topic");
KStream<Integer, Integer> output = input
    .filter((k, v) -> v > 10)
    .mapValues(v -> v - 5);
output.to("output-topic");

A 3

B 4

C 2

D 6

💡 Hint

Count how many values are greater than 10 before mapping.

Values greater than 10 are 12, 20, and 15. So 3 records pass the filter and appear in output.

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