Concurrency Practice: Multi-threaded Data Processor

What You'll Build

In this exercise you will build a multi-threaded data processor: a program that takes a large batch of work items, cricket match records to score and analyse, and processes them concurrently across multiple threads, safely aggregating the results. It brings together everything from Module 5: the Executor framework to run tasks, Callable and Future (or virtual threads) to produce results, thread-safe aggregation with atomics and concurrent collections, and CompletableFuture to compose the processing pipeline.

Rather than practising each concurrency tool in isolation, you will combine them into a realistic processing engine where correctness under concurrency is the whole point: many threads compute partial results simultaneously, and those results must be merged without races. By the end you will have a processor that turns a large workload into results far faster than sequential code while remaining correct, the core pattern behind every concurrent batch and request-processing system, consolidating Module 5 before the course moves into modern language features.

Prerequisites

• Completion of lessons 21 through 24: threads and executors, synchronization and locks, CompletableFuture, and virtual threads.
• A JDK installed, version 21 recommended so virtual threads are available; verify with java -version.
• Comfort submitting Callable tasks to an ExecutorService and retrieving results via Future from lesson 21.
• Understanding of why shared mutable state races and how atomics and concurrent collections make aggregation safe, from lesson 22.
• Familiarity with composing asynchronous stages and handling failures with CompletableFuture from lesson 23, and the thread-per-task virtual-thread model from lesson 24.

Setup & Project Structure

The processor is organised into a work item (the data to process), a worker that computes a partial result for a chunk of items, a thread-safe aggregator that merges partial results, and a driver that splits the workload, dispatches the workers concurrently, and collects the final result. Separating the unit of work, the parallel computation, and the safe aggregation keeps each concern clear, which matters doubly in concurrent code where tangled responsibilities breed races.

Lay out the pieces so the data flow is explicit, raw items in, partial results computed in parallel, merged result out, and build incrementally: first a correct sequential version to validate the logic, then the concurrent version, so you can confirm the concurrent results match the sequential ones. That comparison is the single most valuable test for concurrent code, since it catches races that only manifest under parallelism.

# Project layout for the multi-threaded data processor.
mkdir data-processor && cd data-processor

# Pieces (one public type each):
#   MatchRecord.java     -- an immutable work item (a match to process)
#   ScoreWorker.java     -- a Callable that processes a CHUNK and returns a partial result
#   ResultAggregator.java-- thread-safe merge of partial results (atomics / ConcurrentHashMap)
#   Processor.java       -- main: split work, dispatch concurrently, collect final result

# Build strategy:
#   1) Write a SEQUENTIAL processor first and record its output (the source of truth).
#   2) Write the CONCURRENT processor; assert its result EQUALS the sequential result.
#      (Mismatches under concurrency reveal races -- the key test.)

# Compile and run:
#   javac *.java   then   java Processor

Step 1 — Foundation

Step 1 builds the immutable work item and a correct sequential processor as the baseline. The MatchRecord is immutable, processing must never mutate the input, which is exactly what makes it safe to share across threads later. The sequential processor computes whatever aggregate the exercise targets (say, total runs and a count per team) by iterating all records in one thread.

Establishing the sequential version first matters because it is the ground truth: it is simple enough to be obviously correct, and it produces the exact result the concurrent version must reproduce. This is the disciplined way to approach concurrency, get a correct single-threaded answer, then parallelise while verifying the answer is unchanged, so any discrepancy points straight at a concurrency bug rather than a logic error you cannot distinguish from a race.

// MatchRecord.java + a sequential baseline -- the ground truth for correctness.
import java.util.*;

public record MatchRecord(String team, int runs, int wickets) {}

class SequentialProcessor {
    // Baseline: total runs per team, computed in ONE thread. Obviously correct.
    static Map<String, Integer> totalRunsByTeam(List<MatchRecord> records) {
        Map<String, Integer> totals = new HashMap<>();
        for (MatchRecord r : records) {
            totals.merge(r.team(), r.runs(), Integer::sum);
        }
        return totals;
    }

    static long grandTotal(List<MatchRecord> records) {
        long total = 0;
        for (MatchRecord r : records) total += r.runs();
        return total;
    }
}

Step 2 — Core Logic

Step 2 implements the worker and the thread-safe aggregator, the heart of the concurrent design. A ScoreWorker is a Callable that takes a chunk of records and returns a partial result computed entirely from its own chunk, with no shared mutable state, which is the safest possible concurrent task: it reads immutable input and returns a value. The aggregator then merges these partial results, and here is where thread-safety is decided.

The key design choice is how to aggregate without races. Two clean approaches: have each worker return its own partial map and merge them on a single thread afterward (no shared state during computation at all), or have workers update a shared ConcurrentHashMap or AtomicLong directly using thread-safe operations. The first is simplest and race-free by construction; the second uses the concurrent tools from lesson 22. Either way, the principle is the same: never let two threads perform an unsynchronised read-modify-write on shared data, exactly the race the module warned about.

// ScoreWorker.java + ResultAggregator.java -- Step 2: race-free partial results + safe merge.
import java.util.*;
import java.util.concurrent.*;

// A Callable that processes ITS OWN chunk and returns a partial map -- no shared state.
class ScoreWorker implements Callable<Map<String, Integer>> {
    private final List<MatchRecord> chunk;
    ScoreWorker(List<MatchRecord> chunk) { this.chunk = chunk; }

    @Override public Map<String, Integer> call() {
        Map<String, Integer> partial = new HashMap<>();   // LOCAL: nothing shared
        for (MatchRecord r : chunk) partial.merge(r.team(), r.runs(), Integer::sum);
        return partial;                                    // result flows back via Future
    }
}

class ResultAggregator {
    // APPROACH A: merge partial maps on ONE thread -> race-free by construction.
    static Map<String, Integer> mergeAll(List<Map<String, Integer>> partials) {
        Map<String, Integer> total = new HashMap<>();
        for (Map<String, Integer> p : partials) {
            p.forEach((team, runs) -> total.merge(team, runs, Integer::sum));
        }
        return total;
    }
    // APPROACH B (alternative): workers update a shared ConcurrentHashMap directly,
    // using merge(...) which is atomic per key -> also race-free, via lesson 22's tools.
}

Step 3 — Integration & Enhancement

Step 3 wires it together in the driver: split the workload into chunks, submit a ScoreWorker per chunk to an executor, collect the partial results from the Futures, and merge them into the final answer. This is where the Executor framework, Callables, and Futures from lesson 21 combine with the safe aggregation from Step 2. You can use a fixed thread pool sized to the cores for this CPU-bound aggregation, or, to practise lesson 24, a virtual-thread-per-task executor.

The enhancement is to express the whole pipeline with CompletableFuture: launch each chunk asynchronously with supplyAsync, combine all the partial-result futures with allOf, and then merge, with exceptionally handling any chunk that fails, so the processor degrades gracefully rather than losing the whole batch to one bad chunk. This demonstrates the progression of the module: from raw executors and Futures to composed asynchronous pipelines, all over the same race-free partial-result design.

// Processor.java -- Step 3: split, dispatch concurrently, collect, and merge.
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class Processor {
    static <T> List<List<T>> chunk(List<T> items, int parts) {
        int size = (int) Math.ceil(items.size() / (double) parts);
        List<List<T>> chunks = new ArrayList<>();
        for (int i = 0; i < items.size(); i += size)
            chunks.add(items.subList(i, Math.min(i + size, items.size())));
        return chunks;
    }

    public static void main(String[] args) throws Exception {
        // Build a large workload.
        List<MatchRecord> records = IntStream.range(0, 100_000)
            .mapToObj(i -> new MatchRecord(i % 4 == 0 ? "IND" : "AUS", i % 200, i % 11))
            .collect(Collectors.toList());

        Map<String,Integer> expected = SequentialProcessor.totalRunsByTeam(records); // ground truth

        // CONCURRENT: one async worker per chunk, combine with allOf, then merge.
        try (ExecutorService pool = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors())) {
            List<CompletableFuture<Map<String,Integer>>> futures = chunk(records, 8).stream()
                .map(c -> CompletableFuture.supplyAsync(() -> new ScoreWorker(c).call(), pool)
                            .exceptionally(ex -> Map.of()))   // a bad chunk -> empty, not fatal
                .collect(Collectors.toList());

            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            List<Map<String,Integer>> partials = futures.stream().map(CompletableFuture::join)
                                                         .collect(Collectors.toList());
            Map<String,Integer> result = ResultAggregator.mergeAll(partials);

            // THE KEY TEST: concurrent result must EQUAL the sequential ground truth.
            System.out.println("matches sequential? " + result.equals(expected));
            System.out.println("totals: " + result);
        }
    }
}

Step 4 — Testing & Verification

The decisive test for concurrent code is that the concurrent result exactly equals the sequential ground truth, run it repeatedly, since races are intermittent, and confirm it matches every time. Also verify graceful degradation: inject a chunk that throws and confirm the processor still completes with the other chunks rather than failing entirely, and confirm the executor is shut down so the program exits. If you used a shared aggregator instead of partial-result merging, this comparison is exactly what exposes any race in it.

# Compile and run, then verify correctness and resilience.
javac *.java && java Processor

# Expected:
#   matches sequential? true        <- concurrent result EQUALS the single-threaded result
#   totals: {IND=..., AUS=...}
#
# Verification checklist:
#   [ ] Run it 10+ times: 'matches sequential? true' EVERY time (races are intermittent).
#   [ ] Concurrent run is faster than the sequential baseline on a multi-core machine.
#   [ ] Injecting a throwing chunk still yields a completed result (exceptionally handles it).
#   [ ] The program EXITS promptly -> the executor was shut down (try-with-resources).
#
# If 'matches sequential?' is EVER false, you have a race: a worker is sharing/mutating
# state it should not, or aggregation does an unsynchronised read-modify-write.
# Re-run many times under load to surface intermittent races before trusting the result.

• The processor combines all of Module 5: an Executor running Callable workers, results via Future/CompletableFuture, and thread-safe aggregation of partial results.
• Immutable work items are safe to share across threads because they are only read, never mutated, the safest foundation for concurrent processing.
• Each worker computes a partial result from its own chunk with no shared mutable state, the most race-free design; merging happens afterward or via concurrent collections.
• The driver splits the workload, dispatches workers concurrently, collects results, and merges them, optionally composing the pipeline with CompletableFuture and allOf.
• exceptionally lets a failed chunk degrade to an empty partial result rather than losing the whole batch, and the executor is shut down via try-with-resources so the program exits.
• The decisive test is that the concurrent result exactly equals a sequential ground truth, run many times, since races are intermittent and never proven absent by one passing run.