Capstone: ShopStream

Part 3: The solution, step by step

How to use this part

This part walks you through one complete solution. It follows the same build order as Part 2.

• Each step says what you build, why, and how you check it.
• You will see short pieces of code. They show the important ideas. They are not the full classes.
• The full, working project is in a separate download: java-shopstream-solution.zip. Use it to compare with your own code, or to unblock yourself.

A good way to work: read one step, close this file, write the code yourself, run it, then compare.

Everything here is run from IntelliJ, the same way you ran every class in the producer and consumer chapters: open the class, click the green triangle next to main, and IntelliJ creates a run configuration you can reuse and edit (for program arguments, say).

---

Step 0: Get ready

Start your three brokers. Then open the project in IntelliJ (File → Open, choose the folder with pom.xml) and click Load Maven Changes if asked.

Run com.kafkacourse.setup.SetupTopics once. You should see:

created: shop.orders
created: shop.orders.dlq

If the topics already exist, you will see already exists: instead. That is fine.

To start again from zero at any time, add --reset as a program argument (Run → Edit Configurations…) and run it again.

---

Step 1: One place for settings

What: a small class, Config, with the settings every other class shares.

javaCopy
public final class Config {
    public static final String BOOTSTRAP_SERVERS =
            System.getenv().getOrDefault("KAFKA_BOOTSTRAP", "localhost:9092,localhost:9094,localhost:9095");

    public static final String ORDERS_TOPIC = "shop.orders";
    public static final String DLQ_TOPIC = "shop.orders.dlq";

    public static final String ORDERS_DB = "orders.db";
    public static final String SALES_DB = "sales.db";
}

Why: if a topic name or an address changes, you change it in one place.

The zones themselves are not a list of strings here — they are their own enum, which is Step 3. The starter kit already gives you Config.

---

Step 2: Check an event

What: a class, EventValidator, with one static method, parse(byte[] raw). It returns a clean OrderEvent, or throws InvalidEventException with a reason.

Why: both consumers need the same rules. Writing them once keeps them identical.

First, the event itself is a record, not a Map. Once the bytes have passed every check, there is exactly one legal shape for them to have:

javaCopy
public record OrderEvent(
        String eventId, EventType eventType, String orderId,
        Zone zone, String customerId, Long amount, String currency) {
}

A small checked exception carries the reason code:

javaCopy
public class InvalidEventException extends Exception {
    private final String reason;

    public InvalidEventException(String reason) {
        super(reason);
        this.reason = reason;
    }

    public String reason() {
        return reason;
    }
}

EventValidator.parse() reads the bytes as a Jackson JsonNode tree — not straight into OrderEvent — because which fields are required depends on the event type, and a broken line may not even be a JSON object:

javaCopy
public static OrderEvent parse(byte[] raw) throws InvalidEventException {
    if (raw == null || raw.length == 0) {
        throw new InvalidEventException("EMPTY_MESSAGE");
    }
    JsonNode node;
    try {
        node = MAPPER.readTree(raw);
    } catch (Exception e) {
        throw new InvalidEventException("INVALID_JSON");
    }
    if (node == null || !node.isObject()) {
        throw new InvalidEventException("NOT_AN_OBJECT");
    }
    // ... then event_id, event_type, order_id, zone, and only for ORDER_CREATED,
    // customer_id and amount.
}

The checks run cheapest first, exactly as the scope document lists them.

A small trap, same as in any language that leans on a loose type system: Jackson's JsonNode.isNumber() already returns false for a JSON boolean, so you might think there is nothing to guard against here. Check it anyway, and explicitly, because the reason is instructive:

javaCopy
private static Long positiveAmountOrNull(JsonNode amountNode) {
    if (amountNode == null || amountNode.isBoolean() || !amountNode.isNumber()) {
        return null;
    }
    double value = amountNode.asDouble();
    if (value <= 0) {
        return null;
    }
    return (long) value;
}

Check it: this needs no cluster at all — write a JUnit test.

javaCopy
assertThrows(InvalidEventException.class, () -> EventValidator.parse("not json".getBytes()));
assertThrows(InvalidEventException.class,
        () -> EventValidator.parse("{\"event_id\":\"E1\"}".getBytes()));

The solution's EventValidatorTest has one parameterized test per reason code. Run it (right-click the class, Run 'EventValidatorTest') before you write a single line of Kafka code — that is the whole point of keeping this class Kafka-free.

---

Step 3: The zone rule

What: an enum, not a tuple of strings.

javaCopy
public enum Zone {
    NORTH, SOUTH, EAST, WEST;

    public static Zone fromText(String text) {
        for (Zone zone : values()) {
            if (zone.name().equals(text)) {
                return zone;
            }
        }
        return null;
    }
}

Why an enum, and not a lookup against a List<String>? Zone.NORTH.ordinal() already is the position in the declared order — 0 for NORTH, 3 for WEST — so there is nothing left to compute by hand, and the compiler stops you from ever typing a zone name that does not exist. This is the same reason OrderStatus (Step 6) and EventType are enums too.

The rule itself is one line, wherever you need it:

javaCopy
zone.ordinal() % numPartitions

Why % numPartitions, and not just ordinal()? It keeps the answer inside the number of partitions, even if the topic is ever recreated with a different count.

Why does an unknown zone need no "return null, let Kafka decide" case here, the way the Python version's zone_partition() needed to return None? Because in Java this rule does not live in a free function the producer calls by hand — it lives inside a real Partitioner, and that class, not this one, is what decides what happens when the zone is unknown. That is Step 4.

Check it:

javaCopy
assertEquals(0, Zone.NORTH.ordinal());
assertEquals(3, Zone.WEST.ordinal());
assertNull(Zone.fromText("CENTRAL"));

---

Step 4: The shared producer code

What: a class, OrderFeed, with everything both producers need: the producer settings, the partitioning rule as a real Partitioner, and the send loop.

4a. The partitioner

This is the one piece of ShopStream that most rewards building the Java way rather than translating the Python one. The producer chapter already taught you the extension point: a class that implements Partitioner, registered with partitioner.class. ZonePartitioner is that same shape, with a zone-based rule instead of PremiumPartitioner's customer-tier rule:

javaCopy
public class ZonePartitioner implements Partitioner {

    private final AtomicInteger roundRobin = new AtomicInteger();

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes,
                          Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);

        Zone zone = peekZone(valueBytes);
        if (zone != null) {
            return zone.ordinal() % numPartitions;
        }
        if (keyBytes != null) {
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
        return Math.floorMod(roundRobin.getAndIncrement(), numPartitions);
    }
    // configure() and close(): nothing to do in either one.
}

Notice what the fallback is: not an arbitrary choice, but the same Murmur2 hash-of-the-key (or round-robin for a null key) that Kafka's own default partitioner uses — the algorithm you verified by hand in the producer chapter, now called directly from org.apache.kafka.common.utils.Utils. An unknown zone does not need a "let the caller decide" signal, because there is no caller to hand it back to; the partitioner is the last stop.

peekZone() is package-private on purpose, so ZonePartitionerTest can check it with no cluster running:

javaCopy
static Zone peekZone(byte[] valueBytes) {
    try {
        JsonNode node = MAPPER.readTree(valueBytes);
        JsonNode zoneNode = node.get("zone");
        return zoneNode == null ? null : Zone.fromText(zoneNode.asText(null));
    } catch (Exception e) {
        return null; // not even valid JSON: EventValidator will reject it properly, later
    }
}

4b. Build the producer

javaCopy
public static KafkaProducer<String, String> buildProducer(String clientId) {
    Properties config = new Properties();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Config.BOOTSTRAP_SERVERS);
    config.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    config.put(ProducerConfig.ACKS_CONFIG, "all");
    config.put(ProducerConfig.LINGER_MS_CONFIG, 50);
    config.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);
    config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
    config.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, ZonePartitioner.class.getName());
    return new KafkaProducer<>(config);
}

Every setting here matches a row in the Part 2 table. Registering ZonePartitioner is the only line that decides partitioning — nothing downstream ever sets a partition number on a record by hand.

4c. Prepare one line

A line becomes a key and a value. A good JSON object gets an event_time stamp (if it does not already have one) and its order_id as the key. Anything else — not JSON, or JSON that is not an object — goes out unchanged, with no key:

javaCopy
static PreparedLine prepare(String line) {
    JsonNode node;
    try {
        node = MAPPER.readTree(line);
    } catch (Exception e) {
        return new PreparedLine(null, line);
    }
    if (node == null || !node.isObject()) {
        return new PreparedLine(null, line);
    }

    ObjectNode event = (ObjectNode) node;
    if (!event.hasNonNull("event_time")) {
        event.put("event_time", nowIso());
    }

    JsonNode orderIdNode = event.get("order_id");
    String key = (orderIdNode != null && orderIdNode.isTextual() && !orderIdNode.asText().isEmpty())
            ? orderIdNode.asText()
            : null;

    return new PreparedLine(key, event.toString());
}

Notice there is no zone extracted or passed anywhere here. Unlike the Python version, where prepare() had to hand the zone to the send loop so it could compute a partition number, the Java ZonePartitioner reads the zone itself, straight out of the value it is already given. Cleaning is still the consumer's job, not the producer's — a broken line is still sent, not dropped.

4d. The send loop and the delivery callback

javaCopy
try {
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            failed.add(exception.toString());
        } else {
            delivered.computeIfAbsent(metadata.partition(), p -> new AtomicInteger()).incrementAndGet();
        }
    });
} catch (TimeoutException e) {
    failed.add(e.toString());
}

Where is the retry-on-a-full-buffer loop? There is not one. producer.send() already blocks internally, up to max.block.ms, whenever the local buffer is full — this is the behaviour you read about in the producer chapter's buffer-and-backpressure lecture, and it is why librdkafka's immediate BufferError has no equivalent to catch here. The only thing left for us to catch is a TimeoutException, thrown if the buffer is still full once that internal wait runs out. One try/catch, no while (true).

4e. The end

javaCopy
producer.flush();
System.out.printf("[%s] delivered per partition: %s%n", label, sortedCounts(delivered));
if (!failed.isEmpty()) {
    return 1;
}
return 0;

---

Step 5: The two producers

What: two tiny classes. Each reads its own file and calls OrderFeed.

javaCopy
public class OrderProducer {
    public static void main(String[] args) throws IOException {
        Path file = Path.of("data", "orders.jsonl");
        double rate = 20;
        int limit = 0;
        // ... parse --file / --rate / --limit from args ...
        System.exit(OrderFeed.sendFile(file, "order-service", rate, limit));
    }
}

FulfilmentProducer is the same shape, with fulfilment.jsonl and the client label fulfilment-service.

Check it: run OrderProducer with program argument --rate 0 (full speed). You will see something like:

[order-service] topic shop.orders has 4 partitions
[order-service] queued 232 messages
[order-service] delivered per partition: {0=50, 1=81, 2=42, 3=59}

The four numbers add up to 232, the number of lines in the file.

Look at the numbers. Partition 1 (SOUTH) has the most messages. This is the busy-zone cost from Part 2.

Now run FulfilmentProducer with --rate 0. The numbers add up to 139.

Both producers use partitions 0 to 3 by zone, through the same ZonePartitioner. The few bad lines fall back to the partitioner's Murmur2/round-robin rule, so your exact counts may differ slightly.

---

Step 6: The database code

What: two small classes, OrderRepository and SalesRepository, each wrapping its own SQLite file through plain JDBC.

6a. The orders database

Two tables: the orders, and the event IDs already seen — created once in the constructor.

sqlCopy
CREATE TABLE IF NOT EXISTS orders (
    order_id    TEXT PRIMARY KEY,
    zone        TEXT NOT NULL,
    customer_id TEXT,
    amount      INTEGER,
    status      TEXT NOT NULL,
    status_rank INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS processed_events (
    event_id TEXT PRIMARY KEY
);

status_rank is OrderStatus.ordinal() — the enum already carries its own rank, so nothing needs a separate lookup table.

6b. Save one event in one transaction

JDBC has no with conn: sugar, so the transaction is written out by hand with setAutoCommit:

javaCopy
public boolean applyEvent(OrderEvent event) throws SQLException {
    conn.setAutoCommit(false);
    try {
        try (PreparedStatement insert = conn.prepareStatement(
                "INSERT OR IGNORE INTO processed_events (event_id) VALUES (?)")) {
            insert.setString(1, event.eventId());
            if (insert.executeUpdate() == 0) {
                conn.commit();
                return false; // we have seen this event before: do nothing
            }
        }
        try (PreparedStatement upsert = conn.prepareStatement(UPSERT_ORDER)) {
            // ... bind order_id, zone, customer_id, amount, status, status_rank ...
            upsert.executeUpdate();
        }
        conn.commit();
        return true;
    } catch (SQLException e) {
        conn.rollback();
        throw e;
    } finally {
        conn.setAutoCommit(true);
    }
}

Read it slowly:

1. INSERT OR IGNORE tries to add the event ID. If it was there already, executeUpdate() returns 0, and we commit (nothing changed) and stop.
2. If it is new, we update the order.
3. Both statements run under one setAutoCommit(false)/commit() pair. A failure anywhere rolls the whole thing back, so a crash between the two steps is impossible.

6c. The update rule

The SQL itself needs no translation — ON CONFLICT ... DO UPDATE, COALESCE, and the rank comparison are exactly as capable in SQLite from Java as from Python:

sqlCopy
INSERT INTO orders (order_id, zone, customer_id, amount, status, status_rank)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(order_id) DO UPDATE SET
    customer_id = COALESCE(excluded.customer_id, orders.customer_id),
    amount      = COALESCE(excluded.amount, orders.amount),
    status      = CASE WHEN excluded.status_rank > orders.status_rank
                       THEN excluded.status ELSE orders.status END,
    status_rank = MAX(excluded.status_rank, orders.status_rank)

• ON CONFLICT ... DO UPDATE: update the existing row, or insert a new one.
• COALESCE(new, old): keep the old value when the new event does not carry one. A SHIPPED event has no amount; it must not erase an amount already saved. A late ORDER_CREATED fills in the amount a stub row did not have.
• CASE WHEN new rank > old rank: the status only moves forward. This is the one place event.eventType().status().ordinal() — the enum's rank — actually earns its keep.

6d. The sales database

Same idea, a different table and a different key (zone and minute, not order ID):

sqlCopy
CREATE TABLE IF NOT EXISTS sales_by_minute (
    zone TEXT NOT NULL, minute TEXT NOT NULL,
    orders INTEGER NOT NULL, revenue INTEGER NOT NULL,
    PRIMARY KEY (zone, minute)
);

To turn a Kafka timestamp (milliseconds since the epoch) into a minute, Java's own time API replaces Python's datetime.fromtimestamp:

javaCopy
private static final DateTimeFormatter MINUTE_FORMAT =
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm").withZone(ZoneOffset.UTC);

public static String minuteOf(long timestampMs) {
    return MINUTE_FORMAT.format(Instant.ofEpochMilli(timestampMs));
}

Check it: SalesRepositoryTest and OrderRepositoryTest run against a real temporary SQLite file created fresh in @BeforeEach (Java has no convenient shared :memory: database across separate DriverManager.getConnection() calls the way Python's sqlite3 does, so a temp file stands in for it and is deleted in @AfterEach). Run them: save the same event twice and count the rows — you must get one row, not two. Save a DELIVERED event and then a late SHIPPED for the same order — the status must stay DELIVERED.

---

Step 7: The Order Loader

What: the first consumer, and the biggest step — but every piece of it is a pattern you already built in the consumer chapter.

7a. The consumer settings

javaCopy
Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Config.BOOTSTRAP_SERVERS);
config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-loader");
config.put(ConsumerConfig.CLIENT_ID_CONFIG, name);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getName());

Match each line with a requirement:

• group.id: requirement B1.
• auto.offset.reset=earliest: a new group starts from the beginning.
• enable.auto.commit=false: we mark our own progress, with commitSync() — requirement B10.
• CooperativeStickyAssignor: requirement B2.

The --name program argument lets you start two copies with different names, so their log lines are easy to tell apart.

Where is enable.auto.offset.store=false? It does not exist in the Java client — there is no separately-tracked "stored offset" to turn off storing. Java's auto-commit, when it is on, simply commits the position poll() has already advanced to, which is exactly what we must not let happen before a batch is fully handled. Turning auto-commit off entirely, and calling commitSync() ourselves once a batch is done, is the Java way to get the same guarantee — and it is precisely the SafeConsumerWithDlq pattern from the consumer chapter, not a new idea invented for this capstone.

7b. A producer for the dead-letter topic

The loader needs a producer too, built with the same safe-delivery settings as OrderFeed's, but no custom partitioner — the DLQ does not need zone-based partitioning:

javaCopy
Properties dlqConfig = new Properties();
dlqConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Config.BOOTSTRAP_SERVERS);
dlqConfig.put(ProducerConfig.CLIENT_ID_CONFIG, name + "-dlq");
dlqConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
dlqConfig.put(ProducerConfig.ACKS_CONFIG, "all");

7c. Sending a bad message to the DLQ

javaCopy
private void sendToDlq(ConsumerRecord<byte[], byte[]> record, String reason) {
    String source = record.topic() + "[" + record.partition() + "]@" + record.offset();
    List<Header> headers = List.of(
            new RecordHeader("error", reason.getBytes(StandardCharsets.UTF_8)),
            new RecordHeader("source", source.getBytes(StandardCharsets.UTF_8)),
            new RecordHeader("failed_at", Instant.now().toString().getBytes(StandardCharsets.UTF_8)));

    ProducerRecord<byte[], byte[]> dlqRecord =
            new ProducerRecord<>(Config.DLQ_TOPIC, null, record.key(), record.value(), headers);

    try {
        dlqProducer.send(dlqRecord).get(10, TimeUnit.SECONDS);
    } catch (ExecutionException | TimeoutException e) {
        throw new RuntimeException("could not write to the dead-letter topic", e);
    }
}

Three things to notice:

1. The original key and value are copied without change, byte[] in, byte[] out.
2. The three headers make the message easy to understand later.
3. .get(10, TimeUnit.SECONDS) — the Future that send() returns — blocks until Kafka confirms, or throws. If the copy fails, the exception propagates out of the poll loop and the batch's commitSync() never runs. That is the safe order, enforced the Java way: by not catching the exception, rather than by Python's explicit raise.

Waiting for each message is slow, but bad messages are rare. The simple way is the right way here, exactly as in the Python version.

7d. Know when partitions move

javaCopy
consumer.subscribe(List.of(Config.ORDERS_TOPIC), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log("assigned: " + partitions.stream().map(TopicPartition::partition).sorted().toList());
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log("revoked:  " + partitions.stream().map(TopicPartition::partition).sorted().toList());
        try {
            consumer.commitSync(); // save our progress before giving these partitions away
        } catch (Exception e) {
            log("could not commit on revoke: " + e);
        }
    }
});

Unlike confluent_kafka's commit(asynchronous=False), Java's commitSync() simply does nothing (rather than raising a "no offset" error) when there is nothing yet to commit, so there is no special exception to swallow here — the try/catch only guards against a genuine commit failure.

7e. The loop

This is the heart of the program:

javaCopy
while (true) {
    ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofSeconds(1));

    for (ConsumerRecord<byte[], byte[]> record : records) {
        try {
            OrderEvent event = EventValidator.parse(record.value());
            if (orders.applyEvent(event)) {
                counters.saved++;
            } else {
                counters.duplicate++;
            }
        } catch (InvalidEventException bad) {
            sendToDlq(record, bad.reason());
            counters.dlq++;
        } catch (SQLException e) {
            throw new RuntimeException("could not save to " + Config.ORDERS_DB, e);
        }
    }

    if (!records.isEmpty()) {
        consumer.commitSync();
    }
}

Follow one record through. It is checked. It is saved (database or dead-letter topic). Only once every record in the whole batch has been handled is commitSync() called. There is no store_offsets() call sprinkled inside the loop — in Java, the batch is committed as a single unit, after the for loop ends, not record by record.

7f. Stopping cleanly

A shutdown hook replaces Python's signal.signal(SIGINT, ...):

javaCopy
Thread mainThread = Thread.currentThread();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    consumer.wakeup();
    try {
        mainThread.join();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}));

consumer.wakeup() makes the blocked poll() throw WakeupException, which the loop catches as its normal, expected way out:

javaCopy
try {
    while (true) { ... }
} catch (WakeupException e) {
    // Expected: this is how wakeup() ends a poll() that is waiting.
} finally {
    log("summary: saved=" + counters.saved + " duplicate=" + counters.duplicate + " dlq=" + counters.dlq);
    consumer.close();
    dlqProducer.close();
    orders.close();
}

consumer.close() leaves the group and triggers one last partition-revoke (and therefore one last commit) on the way out.

Check it: run OrderLoader with program argument --name loader-1.

[loader-1] started. Press Ctrl+C to stop.
[loader-1] assigned: [0, 1, 2, 3]
[loader-1] bad message shop.orders[3]@33 -> DLQ (MISSING_ZONE)
...

Because the events are already in the topic, it starts working at once. Let it run until the messages stop, then stop it (the red square in IntelliJ sends the same interrupt a terminal Ctrl+C would). You will see a summary line:

[loader-1] summary: saved=358 duplicate=3 dlq=10

Count them: 358 + 3 + 10 = 371. That is 232 + 139, every line from both files.

---

Step 8: The Sales Aggregator

What: the second consumer, SalesAggregator. It looks very much like OrderLoader. Only a few things differ.

1. A different group:

javaCopy
config.put(ConsumerConfig.GROUP_ID_CONFIG, "sales-aggregator");

2. It counts only ORDER_CREATED. Everything else is ignored, bad messages too — and bad messages are not sent to the DLQ here, because OrderLoader already owns that job:

javaCopy
try {
    OrderEvent event = EventValidator.parse(record.value());
    if (event.eventType() != EventType.ORDER_CREATED) {
        ignored++;
    } else if (sales.addSale(event, record.timestamp())) {
        added++;
    } else {
        duplicate++;
    }
} catch (InvalidEventException bad) {
    ignored++; // the Order Loader sends these to the DLQ, not us
}

3. The clock. ConsumerRecord.timestamp() is a single long, the time Kafka itself stored on the message — there is no two-part tuple to unpack the way confluent_kafka's msg.timestamp() needed.

4. The same safe order. addSale() first, then commitSync() once the batch is done.

5. A summary at the end: SalesReport.print() prints a small table of orders and revenue per zone, built from Zone.values() so every zone appears even with zero sales.

Check it: run SalesAggregator. When it has caught up, stop it. You will see:

Sales so far
ZONE      ORDERS   REVENUE (INR)
NORTH         28         110,172
SOUTH         40         109,210
EAST          21          56,429
WEST          31         148,769

These numbers are the ones in expected_results.json. Your aggregator and loader read the same topic without disturbing each other — two consumer groups at work.

---

Step 9: Run everything and check

Start again from zero so you see the whole flow: run SetupTopics with --reset.

Open five run configurations (or five IntelliJ run windows):

Start 1, 2 and 3 first. Then start 4 and 5.

Watch the loaders. In the end, each one has two partitions, for example [0, 2] and [1, 3]. The numbers can differ on your computer. The idea is the same: four partitions, two readers, two partitions each.

If you started the loaders one after the other, the first may first show assigned: [0, 1, 2, 3]. When the second joins, the first shows revoked: for the two partitions it gives away. This is cooperative rebalancing. Only the partitions that must move, move.

When the producers finish and the consumers go quiet, stop 1, 2 and 3. Then run VerifyResults.

You should see ALL CHECKS PASSED. Well done!.

If a check fails

---

Step 10: The failure tests

Now prove the pipeline is strong. Reset first (SetupTopics --reset).

Test 1: Stop one loader

Start two loaders, the aggregator and both producers. While data flows, stop loader 2. In loader 1's output you will see loader 2's partitions arrive, logged as only the new partitions:

[loader-1] assigned: [0, 2]

Loader 1 did not stop and restart. With cooperative rebalancing, it kept its own partitions and only received the extra ones.

Test 2: Crash one loader

Do the same, but stop the loader with force — in a terminal, kill -9 <process id> on the JVM process IntelliJ started (IntelliJ's stop button is a clean shutdown; a real crash needs the terminal). Start it again.

Do not worry if it waits. Kafka does not know the loader is gone until its session times out — the default is 45 seconds. After that, the group rebalances and work continues.

When all is quiet, run VerifyResults. It must pass. Look at the summary lines of the loaders — you will probably see higher duplicate counts than before. Those are messages that arrived a second time after the crash. The event_id check in processed_events ignored them. Nothing was lost, and nothing was counted twice.

This is the safe order paying for itself.

Test 3: Stop a broker

Start the producers with a slow rate, so you have time — program argument --rate 5.

While it runs, stop one of the three broker containers, the same way you did in the failure labs of the course. The producer keeps going. It may print a short delay while Kafka picks a new leader for some partitions. When it ends, check that it reports no failed messages. Start the broker again.

Why does this work? Replication factor 3, min.insync.replicas=2, acks=all and an idempotent producer. Two brokers are enough to accept a write. The producer retries internally and safely — no code of yours needs to notice.

---

Step 11: The replay tool

What: ReplaySales. It rebuilds sales for a time window, straight from Kafka. This is ReplayByTimestamp from the consumer chapter, with one addition: an end timestamp, so a window has both edges.

11a. Read the times

javaCopy
private static long parseUtcMinute(String text) {
    return LocalDateTime.parse(text, MINUTE).toInstant(ZoneOffset.UTC).toEpochMilli();
}

11b. A consumer that cannot disturb anything

javaCopy
Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Config.BOOTSTRAP_SERVERS);
config.put(ConsumerConfig.CLIENT_ID_CONFIG, "replay-tool");
// No group.id at all.

Python's replay tool used a throw-away random group name — a convention that relies on nobody else happening to pick the same name. Leaving group.id out entirely in Java is a stronger guarantee than a convention: a consumer with no group cannot join one, and cannot commit, even by accident. That is what requirement D3 is really asking for.

11c. Find where to start

javaCopy
Map<TopicPartition, Long> timestamps = new HashMap<>();
for (TopicPartition partition : partitions) {
    timestamps.put(partition, startMs);
}
Map<TopicPartition, OffsetAndTimestamp> resolved = consumer.offsetsForTimes(timestamps);
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

offsetsForTimes() answers with an OffsetAndTimestamp per partition, or null if nothing in that partition is at or after startMs. endOffsets() is recorded once, up front — the same "freeze the finish line before you start running" idea ReplayByTimestamp already used.

11d. Read the window

javaCopy
consumer.assign(startOffsets.keySet());
for (Map.Entry<TopicPartition, Long> start : startOffsets.entrySet()) {
    consumer.seek(start.getKey(), start.getValue());
}

assign() points the consumer straight at these partitions; there is no group and no rebalance to wait for.

Inside the loop, a partition can finish for either of two reasons:

javaCopy
if (record.timestamp() >= endMs) {
    // We reached --to on this partition. Stop reading it; do not count this record.
    pending.remove(partition);
    consumer.pause(List.of(partition));
    continue;
}
...
// Reached the end offset that existed when we started? Done too, even if no
// record ever reached --to (the window end is beyond anything written).
for (TopicPartition partition : new ArrayList<>(pending)) {
    if (consumer.position(partition) >= endOffsets.get(partition)) {
        pending.remove(partition);
        consumer.pause(List.of(partition));
    }
}

The first check is new, for --to. The second is exactly ReplayByTimestamp's own stopping rule, kept as a safety net for a window whose end lies past everything ever written. The loop ends when every partition is done.

11e. Compare

javaCopy
Map<String, long[]> saved = salesDb.zoneTotals(
        SalesRepository.minuteOf(startMs), SalesRepository.minuteOf(endMs));
System.out.println(sameTotals(saved, totals) ? "MATCH" : "DIFFERENT: sales.db does not agree with Kafka");

Check it: note the time when you started the producers, in UTC. For example, if you started at 14:05 UTC and finished by 14:08:

Run ReplaySales with program arguments --from "2026-11-06 14:05" --to "2026-11-06 14:08".

Read 77 messages between 2026-11-06 14:05 and 2026-11-06 14:08 (UTC)

Rebuilt from Kafka
ZONE      ORDERS   REVENUE (INR)
...

Saved in sales.db for the same window
ZONE      ORDERS   REVENUE (INR)
...

MATCH

Use a window that covers a time when your producers were running. If your producers sent everything in one minute, use that minute.

If it says DIFFERENT, you have found a bug in the aggregator or in the replay tool. Fix it — this is exactly what the tool is for.

---

Step 12: Clean up

When you are done:

• Stop the consumers.
• Delete orders.db and sales.db if you want a fresh start — SetupTopics --reset does this for you.

---

What you practised

---

Common problems

---

Well done

You built a small but real data pipeline. It takes events from two sources, keeps them in order where it matters, survives crashes, parks bad data safely, and can look back in time.

The same ideas run in large companies every day.