Process Each Event Once

An interviewer asks: "Your consumer crashed and restarted, and it received events 40 to 60 a second time. How do you keep the totals right?" Event streams such as Kafka deliver at least once. A restarted consumer reads again from the last place it saved, so whatever it handled after that place arrives twice. Every event carries an offset, its place in the stream (1, 2, 3…), and it keeps the same offset when it is sent again.

The trap: adding up whatever arrives. The consumer here reads events 1 to 60, restarts at 40 and reads on to 80. It counts 101 events and 4431 in revenue, but the stream holds 80 events worth 3711. Every report built on those totals overstates sales by 720, the revenue of events 40 to 60.

The fix is called exactly-once processing: the stream may repeat events, and your consumer makes the repeats harmless. Keep the committed offset, the offset of the last event already in your totals, and skip every event at or below it. With a committed offset of 5:

committed offset 5
event 4 arrives  →  skip it (4 <= 5)
event 5 arrives  →  skip it (5 <= 5)
event 6 arrives  →  add it, commit 6
event 7 arrives  →  add it, commit 7

Commit after writing, in the same save as the totals. If you commit first and crash before writing, the restart skips an event that never reached the totals, so it is lost. If you write and then crash before committing, the restart counts the event again. Kafka stores each consumer group's committed offsets for you, and its transactions can commit a job's output and its offsets together. The rule stays the same.

The hidden setup writes events.csv: the first 80 order lines of orders.csv as events, with offset (1 to 80), order_id, product and amount (quantity × unit price). In the starting code, load_state() reads state.json and save_state(state) writes it. It holds offset, events and revenue. consume(first, last) processes events first to last and saves after each one. The script starts from a blank state, consumes 1 to 60, then 40 to 80. The check prints the state, the true revenue, and the state after one more restart at 75:

After the restart: {'offset': 80, 'events': 80, 'revenue': 3711}
Revenue of the 80 events: 3711
After another restart at 75: {'offset': 80, 'events': 80, 'revenue': 3711}

Your task: in the loop of consume, read the event's offset as a whole number. Skip the event with continue when its offset is at or below state["offset"]. Then record the offset in state below the revenue line, so save_state commits it together with the totals.

Assignment
Follow the Coding Tutorial to make consume skip the events it has already counted and commit its offset with the totals.

Hint
Use <=, not <: the committed offset belongs to an event that is already in the totals. Set state["offset"] before the save_state(state) line, not in a separate save, so the offset and the totals are written together.