Build a Java consumer that writes each processed event ID to a local file, then commits the poll result’s next offsets. Test how it behaves during group changes, a crash, and a poll timeout. Finally, add a topic through a pattern subscription and reset saved offsets to replay records.

The fixed contract makes the results checkable: every order event carries a unique eventId, the consumer writes one line per processed record, and it commits only after the consumer writes the whole poll result. The file is the external effect and the commit is a separate operation, so a crash after the file write but before the commit can cause repeated event IDs after restart.

Use one terminal per consumer instance plus one for Kafka commands, and run every java command from the project directory.

Create:

Start from the consumer project the building-a-consumer chapter set up. Copy its pom.xml into the lab directory unchanged; you do not need to change dependencies. Build and launch the same way, with mvn compile dependency:copy-dependencies -DoutputDirectory=target/lib followed by java -cp 'target/classes:target/lib/*'.

If no kafka-local container exists, create it (if it exists but is stopped, run docker start kafka-local instead):

Create the order topic with three partitions:

Save this as publish.sh. It generates keyed order events with consecutive event ids so each phase can add a fresh batch:

Publish the first batch, evt-7001 through evt-7009 with keys ord-1042 through ord-1050:

Never publish a range twice; a second run appends the same event ids at new offsets and contaminates the Phase 3 evidence.

Save this as src/main/java/com/example/kafka/OrderConsumerLab.java. One thread owns the consumer, appends each record to processed/<instance>.log, and commits the batch's next offsets only after it writes every record. System properties select the instance name, an optional crash point, an optional per-record sleep, and the poll settings.

The crash point does not exit on its own. It writes the file, prints its process id, and waits for up to 120 seconds before the commit so you can kill the process from another terminal. kill -9 bypasses the shutdown hook, so nothing runs between the file write and the process ending.

Build, then start the first instance:

Once the output stops, inspect the group in the command terminal:

What you should see: one ASSIGNED line listing orders.placed-0, orders.placed-1, and orders.placed-2, nine processed lines, and at least one COMMITTED line. processed/c1.log holds nine distinct event ids. In the describe output, CURRENT-OFFSET equals LOG-END-OFFSET for all three partitions and LAG is 0. Record which partition each key landed on for Phase 3.

Keep c1 running. In a second terminal, start c2 with -Dlab.instance=c2. Copy the REVOKED and ASSIGNED lines from both terminals into observations/assignment-log.md, then inspect the members:

Start c3 in a third terminal and record the new lines and member output. Then press Ctrl-C in the c2 terminal and record again.

What you should see: with the default RangeAssignor, two members split the three partitions two and one, three members own one each, and after c2 leaves the survivors return to two and one. Every REVOKED line on an existing member lists all of its partitions, including ones it receives back in the next ASSIGNED line; that is the eager handoff. The stopping instance prints its own REVOKED during close(). Build a table with one row per event and one column per instance.

Stop every instance. Publish the second batch and start one instance with a crash point on evt-7012:

When the CRASH POINT line appears, save the most recent BEFORE BATCH line, then run the printed kill -9 command within the 120-second pause. Describe the group until it reports no active members; the default session timeout is about 45 seconds, but verify the state. Save that output. Restart c1 without the crash property, wait until its backlog reaches zero, and describe the group again.

Extract the duplicates:

What you should see: after the kill, CURRENT-OFFSET for the partition holding evt-7012 matches the BEFORE BATCH committed value for the crash-point poll, LAG is greater than zero, and the group has no active members. After the restart, uniq -d prints evt-7012 and every other event id the loop had already written from the same poll result, because the batch commit never happened. Each duplicate pair shows the same topic, partition, and offset: the consumer reread them; the producer did not publish them again. Save the uniq -d output, the matching file lines, and both describe outputs in observations/duplicate-evidence.txt.

Stop the restarted Phase 3 consumer and describe the group. Save its per-partition offsets as the third batch's starting positions in observations/poll-violation.txt. Publish the third batch, then start the slow instance with a six-second delay per record and a fifteen-second poll interval:

Any poll result with three or more records now outlasts the interval. Check the POLL records= lines: poll boundaries vary, so a smaller result may finish in time. If the consumer processes all nine records without a violation, stop it, reset the third batch's positions to their pre-run values, and repeat. Record the observed batch size rather than claiming that every poll contains nine records. After a violation makes the process exit, describe the group.

What you should see: while the loop is still sleeping, the client logs a warning that the consumer poll timeout has expired and the member is leaving the group. Processing continues, because nothing interrupts the owner thread. When the loop reaches commitSync(), it throws CommitFailedException stating that the consumer is no longer part of an active group, and the process exits. The describe output shows no active members, and the failed poll has not advanced CURRENT-OFFSET beyond its BEFORE BATCH values, even though processed/slow.log contains that poll's event ids. Earlier successful polls may already have advanced other positions.

Once the slow instance has exited, export the current offsets while the group is inactive. These positions define the same unread work for both fixes:

Copy the reset file into the container:

First run with -Dlab.max.poll.records=1, keeping the six-second delay. Wait until lag reaches zero, then stop the instance and confirm that the group has no active members. Restore the exported offsets before testing the second fix:

Describe the group and compare the restored positions with poll-start.csv. Then run with -Dlab.max.poll.interval.ms=120000 and the default record limit. Keep the six-second delay. Once lag reaches zero, stop the instance again. Save both runs and the reset output in observations/poll-violation.txt.

Both runs finish the batch and commit. The report's Q4 asks which fix fits which scenario.

Start one instance with the pattern subscription. The pattern ^orders\.[a-z]+$ selects the orders.* family while keeping the chapter's orders.placed.consumer-demo topic out if it still exists:

After the ASSIGNED line appears, note the time and create the cancellation topic:

Wait for the next ASSIGNED line, note the time again, then publish three cancellations:

What you should see: the second ASSIGNED line includes orders.cancelled-0 through orders.cancelled-2 alongside the placed partitions, and it arrives without a restart, after a metadata refresh and assignment; record the observed delay rather than treating metadata.max.age.ms as a deadline. The three cancellation events appear in processed/c1.log with topic orders.cancelled, and the describe output lists six partitions. Record both timestamps, the assignment lines, and the describe output in observations/pattern-discovery.txt.

Stop every instance, describe the group, and save that output as the "before" state in observations/reset.txt. Preview a reset of the whole group to the earliest offsets; without --execute the tool only reports what it would change:

Check that the preview names both topics and all six partitions, then run the same command with --execute appended and describe the group again. Start a fresh instance named replay:

Wait until the group description shows zero lag for both topics. Then stop it and count the lines in processed/replay.log. Before shifting, check that orders.placed-0 contains at least three records. If it does not, choose another placed partition that does and substitute its number in the command and comparison below. Shift that partition backward by three records and replay once more:

What you should see: after the first reset, CURRENT-OFFSET is 0 for all six partitions, since Kafka has deleted nothing from these fresh topics. processed/replay.log contains thirty lines (twenty-seven placed, three cancelled) with no repeated event ids. After the shift, only orders.placed-0 shows a committed offset three below its log end, and the second replay appends exactly that partition's last three records. Save the preview, both execute outputs, and the line counts.

Submit:

The submission passes when:

Press Ctrl-C in every consumer terminal and confirm no OrderConsumerLab process remains:

Remove the broker and its stored data:

Verify with docker ps -a, which should list no kafka-local container. Delete target/; keep processed/, observations/, and report.md as your submission.

The consumer you built keeps three positions apart: the records the client has returned, the work the application has finished, and the offset the group has saved. Phase 3 showed that a commit placed after processing still leaves a window in which the external effect exists and the saved offset does not yet include that work, and the restart repeated exactly the records from that window. Phase 6 produced the same replay by moving the saved offset deliberately, so replay-safe processing matters whether the cause is a crash or an operator.

Group membership added its own evidence. The REVOKED and ASSIGNED lines exposed the eager handoff of the default assignor, the pattern subscription showed a topic joining the assignment through a metadata refresh, and the poll-interval violation showed the client leaving the group while its heartbeats were still healthy. Ownership changes determined which process could commit, and CommitFailedException was the group protocol enforcing that boundary.