Introduction
A studio has retained its voting history while its original display is paused. You need a new count without rewinding that display or changing the log. A poll can briefly return no message even though a new consumer has not yet received the frozen batch.
You will build the native replay loop, capture a separate end position for every partition, and publish only after the complete finite batch has been read. The logical-identity function from the preceding lab is already supplied in this independent VM. You need the Python and Kafka foundations used there. Use the Terminal. The author's 50–60 minute estimate includes reading and ordinary errors; it is not a timed beginner trial.
This is a frozen-batch reconstruction. There is no ongoing publication to these topics during the exercise, no transaction implementation to build, and no claim of reliable crash recovery or end-to-end exactly-once processing.
Capture Every Partition's Boundary and Read to It
In this step, you will implement a complete independent native read inside a prepared outer program. Start in the Terminal. Edit replay.py and, if needed, reducer.py or Python helpers under helpers/; write derived files under output/. Preserve the supplied tools/data, original topics and paused display groups:
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
The read-only description shows the real studio-votes partition IDs, retained low, current frozen end and replicas. A partition is an ordered log, not a separate VM. The end is the next position after the retained batch, not a record to count. Predict which partitions need any records: an empty partition has low == high and is complete already.
An offset span is not a record count. Kafka can skip internal/control or removed positions. This source includes a native skipped position; there is no missing vote to manufacture. Inspect the real envelopes and compare their actual positions with the description:
bigdata-python -B vote_tool.py inspect main
Before rebuilding, read the paused display's saved positions with the consumer-group tool from the prerequisite course:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Keep each CURRENT-OFFSET for the later comparison. No active members is normal: the display is paused but has saved progress. LAG is a difference between log positions here, not a count of logical votes. Your independent consumer must not move this group's progress.
Read the Prepared Outer Program
cat replay.py
Imports, arguments(), client configuration, error reporting and finally/close are already supplied. The core deliberately raises NotImplementedError; it does not read partitions or publish a sample answer. The correct identity function from the preceding lab is prepared separately in reducer.py.
The client uses the selected bootstrap.servers address and an independent group.id. allow.auto.create.topics is false. Automatic commit and automatic offset storage are disabled. assign will give it explicit starting positions without dynamic group allocation; this example makes no explicit commit call. An equivalent independent-group implementation may save its own progress, but must never join, commit to, reset or delete an original display group.
The prepared vote_support.py supplies arguments() for topic/catalog/output/broker/group and finish(...) for complete-result writing. It does not discover partitions or fetch votes. finish calls the identity rule before writing/replacing the result. finally closes the consumer on success or failure; keep that outer structure.
Implement the Native Core in Short Connected Parts
Use the editor workflow from the preceding lab:
nano replay.py
Keep the imports, argument parsing, Consumer configuration, try:, except and finally sections. Inside try:, remove only the two explanatory comments and the incomplete-core raise NotImplementedError(...). Use the arrow keys to position and Backspace/Delete to edit; Ctrl+K can cut the entire current line, so use it only when you are on a line you intend to replace.
Insert the following connected parts in that gap. They form one core: do not run individual blocks at the shell or save/run a half-built loop. Complete all parts, keep the supplied outer error/close framework, then save and run. Four spaces put an instruction inside try:; another four put it inside a loop or condition. The relative indentation shown is part of Python's meaning.
First discover the complete real partition set. Calling list_topics without a requested topic avoids asking metadata for an unknown name. Check availability before using the selected entry. Sorting the discovered IDs gives this example a stable assignment list; it does not create a global event order:
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
For every discovered partition, request get_watermark_offsets(..., timeout=10) once. TopicPartition identifies the topic and partition for that request. Keep both retained low and frozen high, and initialize the diagnostic next position at low. Do not refresh high inside the polling loop: chasing a growing end would no longer be a finite recount. This exercise forbids appending to or changing the frozen source:
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
Build pending only from partitions with retained positions to read. assign receives a list of explicit starting offsets, including lows for empty partitions. Initialize a fresh physical-record list for each run. time.monotonic() measures elapsed time independently of wall-clock adjustments; the deadline bounds an incomplete read:
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
Now start the loop. poll(0.2) can return a message, an error/event-bearing message or None on timeout. Predict why None cannot mean finished: the client may simply not have received a frozen record yet. Keep waiting within the overall deadline. Partition EOF events are enabled in the configuration; _PARTITION_EOF is the expected end event, while other errors fail the run. This allows native progress past a final internal position:
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
Continue the same if message is not None block with the following elif, aligned with if message.error(): at twelve spaces. Only a normal message in a pending partition, below its fixed high, becomes an envelope. Retain physical retries too; summarize handles logical identity later. Decode UTF-8 key/value bytes, then json.loads turns the value into a dictionary:
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
After handling this poll, query consumer.position(...) for every remaining partition. This is the consumer's next position, distinct from any group's committed progress. Compare each with its captured high, not a guessed record count. Remove only completed partitions from pending. The for position line is eight spaces, inside while but outside the message condition, so even an idle poll can be followed by a position check. Finally, call finish at four spaces, after the loop has completed:
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
Every partition now has a fixed boundary. Empty ones needed no vote; skipped positions did not require invented records; None did not stop the read. Only after all pending next positions reach their highs is a complete snapshot published.
Save with Ctrl+O, Enter to confirm replay.py, then Ctrl+X. Inspect the saved program:
cat replay.py
Check that all six parts are inside the supplied try, envelope handling stays inside the normal-message branch, position checking stays inside the loop, and finish is outside it. Run immediately against the selected native batch:
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
Expect eight physical records, five logical votes and cedar=3, maple=2, willow=0. The public entry first checks the identity function on disposable examples, then records broker data received by the actual native program/helper tree. A separate complete business comparison checks the actual result. Snapshot-reported bounds/positions are diagnostics, not independent proof of native processing.
Each successful invocation must actually write its result, directly or through a written temporary file renamed to the selected output. Identical bytes on repeat are valid, and helper children may write. Merely retaining or updating the timestamp of an old file, or writing a completion marker, is not publication. The provided finish uses the taught temporary-file/replace method. With that prepared writer, loop or validation errors before publication leave the earlier snapshot available; the entry does not roll back arbitrary in-place writes.
For a syntax/name/indentation error, use the reported line to reopen nano replay.py, fix that part and preserve the working pieces. Save/exit, inspect with cat replay.py, then rerun the public entry. For an incomplete-replay timeout, inspect the selected topic argument and the pending, EOF and position handling; it is not a reason to rewind the original group. Use pwd and ls replay.py reducer.py to locate a mistyped path. A local program error does not require rerunning setup or changing source records.
Rebuild from Scratch and Use a Different Batch
In this step, you will prove that another run reconstructs the same finite result, then use a new catalog and partition distribution. Remain in the Terminal at /home/labex/project.
The prepared writer orders physical envelopes by partition/offset and logical votes by ID before saving JSON. This makes the same completed snapshot stable without claiming a global event order. sha256sum computes a content fingerprint. > saves that fingerprint under your output directory; -c later compares the actual named file with the saved fingerprint:
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
Expect output/main-snapshot.json: OK. The program starts with empty records, seen and counts on each invocation. It replaces a complete snapshot, rather than adding the history's votes to the old totals again. Consumption did not delete those records or rewind the paused display.
Read the original display positions again after your repeat:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Every CURRENT-OFFSET must match the earlier paused-display table. Your independent reconstruction did not commit to or rewind that group, even though it read the retained records. The no-active-members message is still normal.
List the check batch and inspect its own booth catalog:
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
This batch has different booth names and three partitions instead of four. Keep the same program files. The parser selects its actual topic and catalog, while discovery finds its complete real partition set:
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
Expect six physical records, four logical votes and amber=2, violet=2, moss=0. Hardcoding the first catalog or stopping at the first topic's record count would not describe this batch.
Return to the first real file:
bigdata-python -B read_snapshot.py output/main-snapshot.json
It still contains the studio's result. The reader opens the requested file directly and supplies nothing from Kafka. If it reports a missing file, compare the filename with vote_tool.py list; correct the path. Rebuild the corresponding batch only when its result actually needs rebuilding.
Refuse a Conflict While Keeping the Valid Snapshot
In this step, you will observe a data error without replacing a useful result. Remain in the Terminal at /home/labex/project. The listed conflict topic deliberately contains one ID with different valid booth content. Its output destination is the existing main snapshot. There is no approved winner: successful publication must be refused.
The fingerprint from the previous step describes the valid main file. Run the same program on the conflict batch. The next command is expected to fail:
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
Expect a nonzero exit status and Conflicting event_id identifying the actual ID. The rule runs before the prepared writer creates/replaces the result. The .pending/replace writer is a complete-result publication technique here; it is not a crash-recovery or durability guarantee.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
The valid main file must still match, with the original studio counts. The failed conflict is recorded separately from the successful main/check runs. Do not edit the conflict source or choose whichever event arrived first. A reverse or sequential partition traversal must still reject the same conflict.
For an ordinary mistyped batch name, the entry prints the available names before starting a replay. Recover with vote_tool.py list and rerun the correct name. For a local program error, repair the file and rerun the affected valid batch. After any source-program edit, rerun both valid batches so their results belong to the current program files; then repeat the conflict self-check.
Keep the original display groups paused: no join, commit, reset or delete is allowed. Resetting and restoring one to the same final offset is still outside independent replay. The functional checks combine native program-tree receipt and actual complete outputs; they do not prove computation causality or resistance to an administrator falsifying the environment.
Summary
You captured actual per-partition boundaries and read the complete finite logs through transient idle polls, empty partitions and skipped positions. The same native program rebuilt two catalogs without accumulating old totals, and a conflicting ID failed while keeping the useful main snapshot and original consumer progress.
