Introduction
A small gallery lets visitors vote for a favourite booth. A sender can retry one event, so a new Kafka record does not always mean another vote. You will first observe a normal two-vote sample, then a genuine retry at another stored position, and finally rebuild the gallery's vote counts.
You should already know terminal paths, Python functions/loops/dictionaries, JSON, and the Kafka key, partition, offset, consumer-group and Python-client basics from Apache Kafka for Beginners. This lab adds business event identity. A complete native reader is prepared so you can focus on the counting rule; the next guided lab builds that reader. Use the Terminal and the installed bigdata-python environment. The author's 40–50 minute estimate includes reading and ordinary mistakes, not a measured beginner trial.
Only a finite, already delivered batch is involved. There is no voter-eligibility system, continuous stream, crash-recovery guarantee or end-to-end exactly-once claim. Each lab starts independently.
Observe One Retry at Another Stored Position
In this step, you will compare two ordinary votes with a retried event. Start in the Terminal. The workspace contains prepared tools and catalogs. Edit only reducer.py, replay.py and optional Python helpers under helpers/; keep derived files under output/. Preserve the supplied tools/data, frozen source topics and paused display groups. The separate demo topic has one permitted action: the retry below.
cd /home/labex/project
-B keeps these Python commands from creating bytecode cache files. The read-only inspect tool prints actual broker envelopes: the topic, partition, offset, key and decoded value. Its final count is a count of stored application records. Inspect the small demo:
bigdata-python -B vote_tool.py inspect demo
Expect two records. Their values have this schema: event_id, booth and choice are nonempty strings; choice is exactly favourite; booth must appear in the selected catalog. No extra fields belong to this course's vote schema. The key is the booth name, used for routing. event_id identifies one business vote. Here demo-1 votes for oak, and demo-2 for fern: two records and two logical votes.
Before sending, predict both results: how many physical records will be stored, and what will the booth counts be if demo-1 is still one business vote? Then test that prediction.
The retry command sends the same complete demo-1 value to the real broker and waits for its delivery confirmation. It checks the existing demo before sending, so running it again after the expected retry does not append a fourth record:
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
There are now three records. demo-1 appears at two different offsets with identical value content. Those positions identify two stored records, while the repeated event ID still identifies one vote. The correct booth totals remain oak=1, fern=1, reed=0. Reading an existing record again and storing another retry are different situations; both require a business identity rule when building counts.
The frozen course batch uses explicit partition placement, so equal booth keys can appear in different partitions. Read the actual metadata when comparing records.
Within the selected course batch, an event ID is global across all partitions, not just within one partition or booth. Two different IDs count as two votes even when every other field is equal. The supplied business accepts those events; it does not decide how many votes one real person may cast.
If a tool reports an unknown batch, list the prepared names and use demo for this observation:
bigdata-python -B vote_tool.py list
A typo has not reset the broker. Correct the argument and repeat the read-only inspection. Do not resend arbitrary events to repair a display count.
Keep Logical Identities and Refuse Conflicting Content
In this step, you will implement the business rule inside the prepared Python function. Remain in the Terminal at /home/labex/project. Read the current file:
cat reducer.py
It initializes catalog counts, then increments for every input value. Predict what it would do with the three demo records: the second stored demo-1 would add another vote. The correct rule must count that ID once while keeping distinct IDs separate.
The prepared rule checker calls an exact copy of your function on disposable examples, including reordered values and different booth names. It does not change the live Kafka log. Run the current naive function; this command is expected to fail with a rule message:
bigdata-python -B check_rules.py
The shell stores the immediately preceding command's exit status in $?: zero means success and nonzero means failure. Display it immediately, before another command replaces it:
printf '%s\n' "$?"
summarize(values, booths) must return a dictionary containing votes and counts for valid input, and raise an ordinary exception for invalid input. Keep the module callable; do not terminate the Python interpreter from it. The checker must receive actual returned values, not just an interpreter exit status.
Edit the Existing Function in nano
Open the editable file in the Terminal's text editor:
nano reducer.py
The top bar identifies the filename. Use the arrow keys to move and Backspace/Delete to correct characters. Python uses indentation to group statements: use four spaces inside the function, eight inside its loop, and another four for each nested condition. Keep the existing def summarize(values, booths): line. Remove the old comment about intentionally counting records when you replace that logic.
The following short Python blocks are connected edits inside this one function, not commands to run at the shell and not separate programs. Complete all the edits before running the checker. Keep the valid parts of the starter instead of replacing the whole file.
First, keep the existing zero-count initialization and add seen immediately below it, before the loop. A dictionary indexed by event_id will hold the complete value for each accepted business vote:
counts = {booth: 0 for booth in booths}
seen = {}
Next, keep the single existing for value in values: line and replace its original increment with the validation below. The block shows that existing for line for context; do not add a second loop. set(value) collects field names; isinstance checks types; any(...) asks whether at least one field value is invalid. The first condition requires a dictionary with exactly three nonempty string fields and the approved choice. Check the catalog before trying to increment a booth. Keep the for line at four spaces and its validation at eight:
for value in values:
if (not isinstance(value, dict)
or set(value) != {"event_id", "booth", "choice"}
or any(not isinstance(item, str) or not item for item in value.values())
or value["choice"] != "favourite"):
raise ValueError("Invalid vote schema")
if value["booth"] not in counts:
raise ValueError("Unknown booth: " + value["booth"])
Now add identity handling immediately after that catalog check, still inside the loop. On a known ID, compare the complete decoded value. Different content must raise an error naming that ID; choosing the first or last value would make the answer depend on traversal order. An identical retry reaches continue, which moves to the next input without another increment. A first ID is saved and counted:
event_id = value["event_id"]
if event_id in seen:
if seen[event_id] != value:
raise ValueError("Conflicting event_id: " + event_id)
continue
seen[event_id] = dict(value)
counts[value["booth"]] += 1
dict(value) keeps the accepted value as a dictionary. JSON whitespace and field order are not business identity. The dictionary key is global across the selected batch's partitions; it is neither a booth name nor a partition/offset pair. Different IDs with the same booth and choice must each reach the first-ID increment.
Finally, replace the old return that used every input value. Return the accepted unique values and all catalog counts. This line has four spaces, outside the loop but inside the function:
return {"votes": list(seen.values()), "counts": counts}
Save and exit. In nano's bottom labels, ^ means hold Ctrl: press Ctrl+O, then Enter to confirm reducer.py; press Ctrl+X to return to the shell. These keys save the current file rather than run Python. Read back what you actually saved:
cat reducer.py
Check that seen is initialized once per function call, schema/catalog checks happen before counting, only a first ID increments, and the return is outside the loop. Then immediately test the completed function:
bigdata-python -B check_rules.py
Expect Rules pass with identities, distinct IDs, conflict, schema, zero booths and reordered input. A dictionary keyed only by booth would collapse distinct vote events; one keyed by partition/offset would retain all retries. Your dictionary represents business identity instead.
For an indentation, syntax or name error, read the reported filename and line, run nano reducer.py again, and use the arrow keys to correct the affected part. Save with Ctrl+O, Enter, exit with Ctrl+X, inspect with cat reducer.py, and rerun the rule check. Keep the parts already correct; you do not need to rewrite the whole function or reset the VM. A mismatched path can be found with these read-only commands:
pwd
ls reducer.py replay.py
Return with cd /home/labex/project if necessary. The next step uses the prepared native reader to apply your rule to real Kafka records.
Publish and Read a Snapshot from the Real Gallery Log
In this step, you will apply your function to the actual frozen gallery batch. Remain in the Terminal at /home/labex/project. List the source and output selected by the main batch, then inspect its catalog:
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main refers to gallery-votes. It has several partitions, repeated IDs at genuinely different stored positions, different IDs with equal other content, and an empty partition. replay.py is the prepared native reader; vote_support.py supplies argument parsing and snapshot writing. The reader passes every actual envelope's value to your summarize function. It does not fetch values from a local vote-input file.
Run the public entry below. It first checks the current identity function on disposable examples, then executes your current program and its helper children, retains native reading evidence for that process tree, and saves the successful result in the listed output path:
bigdata-python -B run_replay.py main
Expect a publication with records=8, logical_votes=5. The entry records real broker reception by this program tree, rather than trusting an unrelated consumer with the same client label. It does not prove internal calculation causality or protection from administrator forgery. The final business result is checked separately against the actual complete source.
Each successful run must actually publish the result during that invocation: write file data directly, or write a temporary file and rename it to the result. A helper child may perform the write. Repeating identical result bytes is valid; merely reading Kafka and retaining an old file is not a new publication.
Read the real result file. This reader never silently fetches, fills or repairs missing results:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Expect five logical vote IDs and these booth counts:
Booth Votes
fern 2
oak 3
reed 0
The snapshot also retains all eight physical envelopes, so you can distinguish its source records from its five logical votes. reed remains present with zero. With the prepared writer, loop or validation errors before publication leave the previous snapshot available. The entry does not automatically roll back other writing methods. After any program edit, run the public entry again before checking completion.
The original display is paused with saved consumer-group progress. You may describe that group, but must not join it, reset/delete it or commit to it. Do not edit/delete the source topic or add partitions. Rebuild only your own derived output. The check observes original topic identity, complete partition/replica topology, actual records and the display's saved progress; it does not claim arbitrary broker configuration protection.
You now have a useful booth snapshot, not just a delivery transcript. The next lab explains and builds the complete independent native reader, including why an empty poll is not a finished batch.
Summary
You observed a genuine stored retry, separated global business identity from Kafka record positions, and rejected conflicting same-ID content. Your real gallery snapshot retains the complete physical source while reporting logical votes and every booth, including zero votes. The source and paused display group remain available.
