KafkaIQ4.1 Stateless Operations
Module Overview
Goal: Define pure transformations clearly — operations that don't require maintaining state.
What You'll Learn:
- What makes an operation "stateless"
- map / mapValues
- filter / filterNot
- flatMap / flatMapValues
- peek
- Branch and merge operations
- Stateless processing guarantees
What's NOT Covered Here:
- Aggregations (covered in Module 5.1)
- Windows (covered in Module 5.2)
- State stores (covered in Module 6.1)
- Joins (covered in Module 8.1)
Key Insight: Stateless operations process each record independently without remembering previous records.
1. What is Stateless Processing?
Definition
A stateless operation processes each record independently without needing to remember or store information about previous records.
Stateless:
Record 1 → [Process] → Output 1
Record 2 → [Process] → Output 2
Record 3 → [Process] → Output 3
Each processed independently
No memory between records
No state stored
Stateless vs Stateful
STATELESS OPERATIONS:
├─ filter → keep/discard based on record alone
├─ map → transform record alone
├─ flatMap → expand/transform record alone
└─ No state stored, no memory
STATEFUL OPERATIONS:
├─ count → need to remember previous counts
├─ aggregate → need to maintain running total
├─ join → need to keep records for matching
└─ State stored, memory required
(Stateful operations covered in Module 5)
Characteristics
STATELESS OPERATIONS:
1. Independent Processing
└─ Each record processed in isolation
2. No State Storage
└─ No RocksDB, no memory overhead
3. High Throughput
└─ Fast, no I/O for state
4. Order Doesn't Matter (Usually)
└─ Result same regardless of order
5. Easy to Scale
└─ No state to migrate
2. map: Transform Both Key and Value
Syntax
KStream<K2, V2> map(KeyValueMapper<K1, V1, KeyValue<K2, V2>> mapper)
What it does: Transform each record (key AND value) into a new record.
Example: Change Key and Value
KStream<String, String> input = builder.stream("users");
// Input: Key="123", Value="Alice"
// Output: Key="user-123", Value="ALICE"
KStream<String, String> transformed = input.map((key, value) ->
KeyValue.pair("user-" + key, value.toUpperCase())
);
Use case: When you need to change BOTH key and value.
Example: Swap Key and Value
// Input: Key="user-id", Value="username"
// Output: Key="username", Value="user-id"
KStream<String, String> swapped = input.map((key, value) ->
KeyValue.pair(value, key)
);
Important: map Can Trigger Repartitioning
If key changes:
└─ Data may need to be repartitioned
└─ Expensive operation (writes to internal topic)
└─ Covered in Module 7.1
Prefer mapValues when possible (doesn't change key)
3. mapValues: Transform Only Value
Syntax
KStream<K, V2> mapValues(ValueMapper<V1, V2> mapper)
What it does: Transform value, keep key unchanged.
Example: Uppercase Values
// Input: Key="123", Value="alice"
// Output: Key="123", Value="ALICE"
KStream<String, String> uppercased = input.mapValues(value ->
value.toUpperCase()
);
Example: Parse JSON
KStream<String, String> jsonStream = builder.stream("json-topic");
// Parse JSON string to object
KStream<String, Order> orders = jsonStream.mapValues(json -> {
return parseJson(json, Order.class);
});
Why mapValues is Preferred
mapValues benefits:
✅ Doesn't change key
✅ No repartitioning needed
✅ Faster than map
✅ Use whenever you only need to transform value
map drawbacks:
⚠️ Can change key
⚠️ May trigger repartitioning
⚠️ Use only when key must change
4. filter: Keep or Discard Records
Syntax
KStream<K, V> filter(Predicate<K, V> predicate)
What it does: Keep records where predicate returns true, discard others.
Example: Filter by Value
KStream<String, Order> orders = builder.stream("orders");
// Keep only high-value orders
KStream<String, Order> highValue = orders.filter((key, order) ->
order.getAmount() > 1000
);
// Input: 100 orders
// Output: ~10 orders (those > 1000)
Example: Filter by Key
// Keep only specific user
KStream<String, Event> userEvents = allEvents.filter((userId, event) ->
userId.equals("user-123")
);
Example: Filter Nulls
// Remove records with null values
KStream<String, String> nonNull = stream.filter((key, value) ->
value != null
);
5. filterNot: Inverse of filter
Syntax
KStream<K, V> filterNot(Predicate<K, V> predicate)
What it does: Discard records where predicate returns true.
Example
// Discard low-value orders
KStream<String, Order> notLowValue = orders.filterNot((key, order) ->
order.getAmount() < 100
);
// Equivalent to:
// orders.filter((key, order) -> order.getAmount() >= 100)
6. flatMap: One-to-Many Transformation
Syntax
KStream<K2, V2> flatMap(KeyValueMapper<K1, V1, Iterable<KeyValue<K2, V2>>> mapper)
What it does: Transform each record into zero or more output records.
Example: Split String into Words
// Input: Key="1", Value="hello world"
// Output: Key="1", Value="hello"
// Key="1", Value="world"
KStream<String, String> words = sentences.flatMap((key, sentence) -> {
List<KeyValue<String, String>> result = new ArrayList<>();
for (String word : sentence.split(" ")) {
result.add(KeyValue.pair(key, word));
}
return result;
});
Example: Expand Nested Array
// Input: Order with multiple items
// Key="order-1", Value={items: ["item1", "item2", "item3"]}
// Output: One record per item
// Key="order-1", Value="item1"
// Key="order-1", Value="item2"
// Key="order-1", Value="item3"
KStream<String, String> items = orders.flatMap((orderId, order) -> {
return order.getItems().stream()
.map(item -> KeyValue.pair(orderId, item))
.collect(Collectors.toList());
});
Example: Filter with flatMap (Zero Outputs)
// flatMap can return empty list to filter
KStream<String, String> filtered = stream.flatMap((key, value) -> {
if (value.length() > 10) {
return Collections.singletonList(KeyValue.pair(key, value));
} else {
return Collections.emptyList(); // Discard
}
});
7. flatMapValues: One-to-Many, Value Only
Syntax
KStream<K, V2> flatMapValues(ValueMapper<V1, Iterable<V2>> mapper)
What it does: Like flatMap, but key unchanged.
Example
// Input: Key="1", Value="hello world"
// Output: Key="1", Value="hello"
// Key="1", Value="world"
KStream<String, String> words = sentences.flatMapValues(sentence ->
Arrays.asList(sentence.split(" "))
);
Prefer flatMapValues over flatMap when key doesn't change (avoids repartitioning).
8. peek: Side Effects (Debugging/Logging)
Syntax
KStream<K, V> peek(ForeachAction<K, V> action)
What it does: Perform action on each record, pass through unchanged.
Example: Logging
KStream<String, Order> orders = builder.stream("orders");
// Log each order (doesn't modify stream)
orders
.peek((key, order) ->
System.out.println("Processing order: " + order.getId())
)
.mapValues(order -> processOrder(order))
.to("processed-orders");
Use Cases
✅ Debugging
└─ Print records during development
✅ Logging
└─ Log events for monitoring
✅ Metrics
└─ Increment counters
❌ Don't use for:
└─ Business logic
└─ State updates
└─ Anything that requires reliability
Important: peek is for side effects. It doesn't modify the stream.
9. Branch: Split Stream into Multiple Branches
Syntax
KStream<K, V>[] branch(Predicate<K, V>... predicates)
What it does: Split stream into multiple streams based on predicates.
Example: Split by Order Value
KStream<String, Order> orders = builder.stream("orders");
KStream<String, Order>[] branches = orders.branch(
(key, order) -> order.getAmount() > 1000, // Branch 0: High value
(key, order) -> order.getAmount() > 100, // Branch 1: Medium value
(key, order) -> true // Branch 2: Low value (catch-all)
);
KStream<String, Order> highValue = branches[0];
KStream<String, Order> mediumValue = branches[1];
KStream<String, Order> lowValue = branches[2];
// Write to different topics
highValue.to("high-value-orders");
mediumValue.to("medium-value-orders");
lowValue.to("low-value-orders");
Complete Multi-Way Routing Example
Here's a more realistic scenario showing different processing logic for each branch:
public class OrderRouter {
public static void main(String[] args) {
StreamsBuilder builder = new StreamsBuilder();
// Source: all orders
KStream<String, Order> orders = builder.stream("orders",
Consumed.with(Serdes.String(), orderSerde));
// Branch orders into 3 categories
KStream<String, Order>[] orderBranches = orders.branch(
(orderId, order) -> order.getAmount() > 1000.0, // High-value
(orderId, order) -> order.getAmount() > 100.0, // Medium-value
(orderId, order) -> true // Low-value (catch-all)
);
// HIGH-VALUE orders: Manual review + priority shipping
orderBranches[0]
.mapValues(order -> order.withStatus("REQUIRES_MANUAL_APPROVAL"))
.mapValues(order -> order.withShipping("PRIORITY"))
.peek((k, v) -> log.info("High-value order: {} → ${}", k, v.getAmount()))
.to("high-value-orders", Produced.with(Serdes.String(), orderSerde));
// MEDIUM-VALUE orders: Auto-approve + standard shipping
orderBranches[1]
.mapValues(order -> order.withStatus("AUTO_APPROVED"))
.mapValues(order -> order.withShipping("STANDARD"))
.to("medium-value-orders", Produced.with(Serdes.String(), orderSerde));
// LOW-VALUE orders: Auto-approve + economy shipping
orderBranches[2]
.filter((orderId, order) -> order.getAmount() > 0) // Reject $0 orders
.mapValues(order -> order.withStatus("AUTO_APPROVED"))
.mapValues(order -> order.withShipping("ECONOMY"))
.to("low-value-orders", Produced.with(Serdes.String(), orderSerde));
// Build and start
KafkaStreams streams = new KafkaStreams(builder.build(), getProperties());
streams.start();
}
}
Data Flow:
Input Topic: "orders"
│
├─────────────────────────────────────────┐
│ │
▼ ▼
amount > 1000? amount > 100?
│ YES │ YES
▼ ▼
[High-Value Branch] [Medium-Value Branch]
│ │
├─ Add: "REQUIRES_MANUAL_APPROVAL" ├─ Add: "AUTO_APPROVED"
├─ Add: "PRIORITY" shipping ├─ Add: "STANDARD" shipping
▼ ▼
"high-value-orders" topic "medium-value-orders" topic
amount ≤ 100
│
▼
[Low-Value Branch]
│
├─ Filter: amount > 0
├─ Add: "AUTO_APPROVED"
├─ Add: "ECONOMY" shipping
▼
"low-value-orders" topic
Branch Behavior
Records tested against predicates in order:
├─ First matching predicate wins
├─ Record goes to that branch only
└─ If no match, discarded (unless catch-all)
Always add catch-all predicate (key, value) -> true
to avoid losing records
10. Merge: Combine Multiple Streams
Syntax
KStream<K, V> merge(KStream<K, V> stream)
What it does: Combine two streams into one.
Example: Merge Multiple Sources
KStream<String, Event> stream1 = builder.stream("source1");
KStream<String, Event> stream2 = builder.stream("source2");
KStream<String, Event> stream3 = builder.stream("source3");
// Merge all into one stream
KStream<String, Event> merged = stream1
.merge(stream2)
.merge(stream3);
// Now process unified stream
merged.to("unified-events");
Merge vs Join
MERGE:
├─ Combines streams into one
├─ No matching/pairing
├─ Just union of records
└─ Stateless
JOIN:
├─ Matches records by key
├─ Combines matching pairs
├─ Requires buffering
└─ Stateful (covered in Module 8.1)
11. Chaining Operations
Fluent API
All stateless operations return KStream, so you can chain them:
builder.stream("input")
.filter((key, value) -> value != null)
.mapValues(value -> value.toUpperCase())
.filter((key, value) -> value.length() > 5)
.peek((key, value) -> System.out.println("Processing: " + value))
.to("output");
Processing flow:
Input → filter(not null) → mapValues(uppercase) → filter(length > 5) → peek(log) → Output
Example: Complete Pipeline
KStream<String, String> orders = builder.stream("raw-orders");
orders
// 1. Parse JSON
.mapValues(json -> parseJson(json, Order.class))
// 2. Filter invalid orders
.filter((key, order) -> order.isValid())
// 3. Enrich with customer name
.mapValues(order -> enrichWithCustomer(order))
// 4. Log for debugging
.peek((key, order) -> log.info("Processed order: {}", order.getId()))
// 5. Convert back to JSON
.mapValues(order -> toJson(order))
// 6. Write to output
.to("processed-orders");
12. Performance Characteristics
Stateless = Fast
Stateless operations:
✅ No disk I/O
✅ No state management overhead
✅ Low latency
✅ High throughput
✅ Easy to scale
Typical latency: < 1ms per record
mapValues vs map
BENCHMARK (1 million records):
mapValues: 500 ms
map: 2000 ms (4x slower if repartitioning occurs)
Use mapValues whenever possible!
13. Common Patterns
Pattern 1: ETL Pipeline
// Extract, Transform, Load
builder.stream("raw-data")
.mapValues(this::extract) // Parse
.filter((k, v) -> this.validate(v)) // Validate
.mapValues(this::transform) // Transform
.to("clean-data"); // Load
Pattern 2: Routing
// Route to different topics based on criteria
KStream<String, Event> events = builder.stream("events");
KStream<String, Event>[] branches = events.branch(
(k, v) -> v.getType().equals("CRITICAL"),
(k, v) -> v.getType().equals("WARNING"),
(k, v) -> true
);
branches[0].to("critical-events");
branches[1].to("warning-events");
branches[2].to("info-events");
Pattern 3: Enrichment
// Add computed fields
orders.mapValues(order -> {
order.setTotalWithTax(order.getTotal() * 1.10);
order.setProcessedAt(System.currentTimeMillis());
return order;
});
Pattern 4: Dead Letter Queue
KStream<String, String>[] branches = input.branch(
(k, v) -> isValid(v), // Valid records
(k, v) -> true // Invalid records
);
KStream<String, String> valid = branches[0];
KStream<String, String> invalid = branches[1];
valid.to("processed");
invalid.to("dead-letter-queue");
14. Best Practices
1. Prefer mapValues over map
// Good
stream.mapValues(value -> value.toUpperCase())
// Avoid if possible
stream.map((key, value) -> KeyValue.pair(key, value.toUpperCase()))
2. Use peek for Debugging Only
// Good (debugging)
stream.peek((k, v) -> log.debug("Record: {}", v))
// Bad (business logic)
stream.peek((k, v) -> database.save(v)) // Don't do this!
3. Add Catch-All in branch
// Good
stream.branch(
predicate1,
predicate2,
(k, v) -> true // Catch-all
)
// Bad (may lose records)
stream.branch(
predicate1,
predicate2
// Missing catch-all!
)
4. Keep Transformations Simple
// Good (simple, readable)
stream.mapValues(value -> value.toUpperCase())
// Bad (complex, hard to debug)
stream.mapValues(value -> {
// 50 lines of complex logic
// Hard to test, hard to maintain
})
// Better: extract to method
stream.mapValues(this::complexTransformation)
15. Mental Model Summary
Stateless Operations
TRANSFORMATIONS (1-to-1):
├─ map → transform key + value
├─ mapValues → transform value only
└─ Use for: changing format, enriching
FILTERS (1-to-0-or-1):
├─ filter → keep matching records
├─ filterNot → discard matching records
└─ Use for: validation, routing
EXPANSIONS (1-to-many):
├─ flatMap → expand to multiple records
├─ flatMapValues → expand values only
└─ Use for: splitting, unnesting
UTILITIES:
├─ peek → side effects (logging)
├─ branch → split into multiple streams
├─ merge → combine multiple streams
└─ Use for: routing, combining sources
16. Questions You Should Be Able to Answer
After this module, you should be able to answer:
✅ What makes an operation "stateless"?
✅ When should I use mapValues instead of map?
✅ What's the difference between filter and filterNot?
✅ When would I use flatMap?
✅ What is peek used for?
✅ How do I split a stream into multiple branches?
✅ How do I combine multiple streams?
✅ Why are stateless operations fast?
17. What's Next?
Now that you understand stateless operations, the next modules will teach you stateful processing:
- Module 5.1: Stateful operations — aggregations that require state
- Module 5.2: Windowing — time-based processing
- Module 6.1: State stores — how state is managed
- Module 8.1: Joins — combining streams with state
Summary
You learned:
- ✅ Stateless operations process each record independently without storing state
- ✅ map/mapValues transform records (prefer mapValues when key unchanged)
- ✅ filter/filterNot keep or discard records based on predicates
- ✅ flatMap/flatMapValues expand single records into multiple records
- ✅ peek performs side effects without modifying the stream
- ✅ branch splits streams, merge combines them
- ✅ Stateless operations are fast, scalable, and easy to reason about
Next up: Module 5.1 — Stateful Operations — Learn how to aggregate and maintain state.