Introduction
The successful consumer wrote a file and then committed reading progress. Those are two separate operations. This lab makes two failure cases visible: a malformed JSON record must stop processing before progress moves past it, and a process that exits after writing but before committing can receive the same record again.
You will first establish a normal delivery, then inspect the exact position of a bad record, and finally use a provided training switch to stop between flush and commit. Restarting the same group will reveal repeated delivery in the file through the same event ID and Kafka position. You will observe that behavior without implementing deduplication or a retry framework.
This independent environment prepares three one-partition topics and the complete consumer familiar from the preceding lab. The additional controlled-exit switch is already implemented. You need the earlier Python JSON, file, and manual-commit concepts; no broker reconfiguration or new recovery architecture is required. Use the Terminal.
Establish the Successful Processing Path
In this step, you will confirm that the application works on valid input before diagnosing a defect. Each topic has one partition so the record positions are direct. orders-normal contains two valid records. The consumer uses the same explicit write, flush, synchronous message commit, and close sequence as before.
Inspect the provided application, including its controlled-exit switch, without changing it:
cd /home/labex/project
cat /home/labex/project/consumer.py
The switch is off by default. Run the normal two-record path with its own group and file:
bigdata-python /home/labex/project/consumer.py --topic orders-normal --group normal-reader --output /home/labex/project/output/normal-delivery.jsonl --max-messages 2 --timeout-seconds 20 > /home/labex/project/output/normal.log 2>&1
cat /home/labex/project/output/normal.log
cat /home/labex/project/output/normal-delivery.jsonl
Expect normal1 and normal2 at offsets 0 and 1, with next committed offset 2 after both writes. max-messages is the number requested in this invocation; the append-only file can retain observations from multiple invocations in later steps.
This successful baseline separates an application-wide connection problem from a record-specific failure. You will use different topics, groups, and files for the deliberate failures so the evidence stays clear.
Stop at Malformed JSON Without Committing Past It
In this step, you will inspect a valid prefix, an invalid JSON record, and a valid later record. Read the local evidence:
cat /home/labex/project/data/bad-events.txt
In orders-bad partition 0, the valid prefix is at offset 0, the malformed text at offset 1, and the later valid event at offset 2. The policy is fail-fast: report the bad position and exit. Do not catch the error and continue to commit later records from that partition.
Request the three-record delivery. The next command is expected to fail.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-first.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first.log
cat /home/labex/project/output/bad-prefix.jsonl
Expect a nonzero status and Decode failure identifying topic=orders-bad partition=0 offset=1. The file should contain only bad-prefix at offset 0. Its successful write was committed, so the group's next offset is 1; the bad record and the later record remain unhandled.
Restart the same group and append to the same file. The next command is expected to fail.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-retry.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry.log
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group bad-reader --describe
The same offset 1 should fail again. The valid prefix should not be written a second time because its next position 1 was already committed. The later valid event must not appear in the file. This makes the policy visible: an operator must choose how to address the bad data, while this minimal application neither skips it silently nor commits past it.
Exit After a Successful Write but Before Its Commit
In this step, you will create a different failure window using the prepared orders-replay topic. It contains only repeat1 at offset 0. The --exit-before-commit training switch exits with status 73 immediately after the file flush and before the explicit commit. It records the actual position and still closes the consumer with automatic commits disabled.
Use a fresh group and a separate file. The next command is expected to fail.
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 --exit-before-commit > /home/labex/project/output/before-commit-exit.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit.log
cat /home/labex/project/output/redelivery.jsonl
Expect status 73 and one saved envelope for repeat1 at partition 0, offset 0. The log identifies CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT. This is a provided fault demonstration, not a change you need to implement.
The file now has the record, while the group has no committed position for it. Earlier starting-policy rules therefore apply when that same group starts again. This separation is exactly the window the next step will observe.
Restart and Identify the Repeated Kafka Record
In this step, you will restart replay-reader without the training switch. Its missing committed position and earliest policy let it read the retained record again. Keep append mode and the same output file so both deliveries remain visible:
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 > /home/labex/project/output/after-restart.log 2>&1
cat /home/labex/project/output/after-restart.log
cat /home/labex/project/output/redelivery.jsonl
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group replay-reader --describe
Expect two envelopes with the same event_id, key, value, topic, partition, and offset. There is still only one original broker record. The second invocation successfully commits next offset 1, while the file retains evidence of both deliveries.
Use the following evidence to distinguish common situations:
| Observation | First useful evidence |
|---|---|
| No new events | Valid topic plus group progress at the end; a bounded poll can return no records |
| Wrong topic name | Compare the supplied name with topic list/describe; this client disables accidental topic creation |
| No valid committed position | Inspect group progress and the configured earliest starting policy; it can read retained history |
| Decode failure | Read the reported topic/partition/offset and the value evidence; the group stops before that bad position |
Writing a file and committing a Kafka position are not one transaction. An interruption between them can cause repeated delivery, as this actual record position demonstrates. This exercise keeps both deliveries and does not implement deduplication, a dead-letter queue, or a transactional design. The final challenge applies the successful path with an explicit clean-run retry workflow.
Summary
You established a normal delivery, stopped malformed JSON without committing past it, and observed a flushed-but-uncommitted record delivered again after restart. Matching event content and Kafka position explained the duplicate file observations without inventing a second broker event. The final challenge now combines successful publication, bounded consumption, file output, and write-before-commit behavior on new data.
