Definition
- Side-effect-free: intermediate operations should not mutate external variables, collections, or fields.
- Parallel danger: multiple ForkJoinPool threads executing simultaneously corrupt shared non-thread-safe collections.
- collect(): handles accumulation internally with per-thread local lists merged at the end.
- reduce(): folds without shared state; the accumulator operates on intermediate values.
- JVM optimisation: side effects in filter/map may be reordered, skipped, or duplicated during optimisation.
- Acceptable: forEach at the pipeline end for fire-and-forget side effects like logging.
Imagine 10 clerks (parallel threads) at National Bank of India each adding entries to one shared ledger book (shared ArrayList). Clerk A writes line 5 while Clerk B writes line 5 at the same time: one entry is lost. The correct approach: each clerk has their own notepad (thread-local accumulation), and at the end a supervisor merges all notepads into the final ledger (collect/reduce).
Understanding the Concepts
Why should stream pipelines be side-effect-free?
Stream pipelines should not mutate external state inside filter, map, or forEach. Side effects in these operations cause race conditions in parallel streams and may be skipped or reordered by JVM optimisations.
What is the problem with shared mutable state?
package com.bank.core;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.IntStream;
public class SideEffectBug {
public static void main(String[] args) {
// WRONG: shared ArrayList in parallel forEach
List<Integer> unsafe = new ArrayList<>();
IntStream.range(0, 10_000).parallel()
.forEach(unsafe::add); // race condition!
System.out.println("Unsafe size: " + unsafe.size()); // often less than 10000
// CORRECT: collect handles parallelism internally
List<Integer> safe = IntStream.range(0, 10_000).parallel()
.boxed()
.toList();
System.out.println("Safe size: " + safe.size()); // always 10000
}
}
Unsafe size: 9847
Safe size: 10000
How does collect handle parallelism safely?
collect uses three functions: supplier (creates a new ArrayList per thread), accumulator (adds to the thread-local list), and combiner (merges thread-local lists). No shared mutable state during execution.
// collect internally:
// Thread-1 → local list [A, C]
// Thread-2 → local list [B, D]
// Thread-3 → local list [E]
// combiner merges: [A, C, B, D, E]
What about side effects in filter and map?
// WRONG: side effects inside filter
List<String> auditLog = new ArrayList<>();
accounts.stream()
.filter(a -> {
auditLog.add("checked: " + a.id()); // side effect in filter!
return a.active();
})
.toList();
// Problem: in parallel, auditLog is corrupted
// Problem: JVM might skip or reorder filter calls during optimisation
What is the correct approach?
Use collect to produce collections, reduce to produce values. Reserve forEach for fire-and-forget side effects at the end of the pipeline.
// CORRECT: collect for collections
List<String> ids = accounts.stream()
.filter(Account::active)
.map(Account::id)
.toList();
// CORRECT: reduce for values
BigDecimal total = accounts.stream()
.map(Account::balance)
.reduce(BigDecimal.ZERO, BigDecimal::add);
What are common mistakes?
// WRONG: mutating a counter in a lambda
int[] count = {0};
accounts.parallelStream()
.filter(Account::active)
.forEach(a -> count[0]++); // race condition on count[0]!
// CORRECT: use count() terminal operation
long count = accounts.parallelStream()
.filter(Account::active)
.count();
// WRONG: accumulating into a shared Map
Map<String, BigDecimal> totals = new HashMap<>();
trades.parallelStream()
.forEach(t -> totals.merge(t.symbol(), t.amount(), BigDecimal::add)); // RACE!
// CORRECT: use groupingBy collector
Map<String, BigDecimal> totals = trades.parallelStream()
.collect(Collectors.groupingBy(Trade::symbol,
Collectors.reducing(BigDecimal.ZERO, Trade::amount, BigDecimal::add)));
What about ConcurrentHashMap as a workaround?
Using ConcurrentHashMap with parallel forEach is thread-safe but still considered bad practice. collect with Collectors is the intended design pattern and is optimised for the ForkJoinPool.
When are side effects acceptable?
forEach at the end of the pipeline is acceptable for fire-and-forget actions like logging or sending notifications.
// Acceptable: forEach at the end for logging
accounts.stream()
.filter(Account::active)
.forEach(a -> logger.info("Active: {}", a.id()));
What should a senior engineer know?
- Streams are designed for functional-style programming where data flows through transformations and results are produced, not accumulated into shared state.
- If you must perform side effects, use forEach at the end and accept non-deterministic order in parallel.
- Use forEachOrdered if order matters, but it defeats parallelism benefit.
- In code reviews, flag any mutable state mutation inside filter, map, or peek as a bug.
Visual Overview
WRONG (shared state): CORRECT (collect): ┌─────────────────────┐ ┌─────────────────────┐ │ Thread 1 → add(A) │ │ Thread 1 → [A, C] │ │ Thread 2 → add(B) │ RACE! │ Thread 2 → [B, D] │ │ Thread 3 → add(C) │ │ Thread 3 → [E] │ │ Shared ArrayList │ │ Per-thread lists │ │ Lost elements! │ │ Merge at end: safe! │ └─────────────────────┘ └─────────────────────┘
Key Takeaway
Stream pipelines should be side-effect-free. Mutating shared state inside filter, map, or forEach causes race conditions in parallel streams. ArrayList.add from multiple threads loses elements silently.
The correct approach: use collect(Collectors.toList()) to produce collections and reduce() to produce values. collect handles parallelism internally with per-thread accumulators merged at the end.
Remember one sentence: use collect for collections and reduce for values; never mutate shared state inside a stream pipeline.
---