Introduction
A consumer that displays an event has not necessarily saved a useful handoff. This application will decode each value, preserve its key and Kafka position in a local JSONL file, and commit progress only after its corresponding write has succeeded and been flushed.
You will map the existing group-and-offset model to Python: subscribe chooses the topic, poll requests records, and an explicit synchronous commit saves the next offset for the handled message. Automatic commit and automatic offset storage are disabled so they do not advance progress independently of your code. File writing and Kafka commits remain separate operations; this is not an atomic cross-system transaction.
This independent lab prepares the six initial order events familiar from the producer lab in a new topic. No earlier VM state is required. You need Python functions, loops, dictionaries, JSON, and running scripts. The parser and deadline structure are supplied. Use the Terminal and the installed bigdata-python environment.
Subscribe, Poll, and Decode One Real Record
In this step, you will inspect one received record without committing preview progress. Read the prepared event values first:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
A Consumer is configured with a group identity and an initial earliest policy. subscribe registers the desired topic; poll returns either a message, an error-bearing message, or None while no record is available, including during assignment. The loop keeps polling within an overall deadline.
The supplied control-flow structure keeps this read bounded. time.monotonic() measures elapsed time without depending on the wall-clock date. continue starts the next poll when no message is ready; break leaves the loop after a record is displayed. The while loop's else runs only if the loop ends without that break, which here means no record arrived before the deadline.
finally closes the consumer whether the read succeeds or raises an error. Read this framework as waiting and cleanup around the Kafka calls; you do not need to design a retry framework.
This preview uses a separate group, file-preview, with both automatic settings disabled. The decoded value is a Python dictionary; the wrapper also retains topic, partition, offset, and a decoded key. No commit call is made:
cat > /home/labex/project/preview_consumer.py <<'PYTHON'
import json,time
from confluent_kafka import Consumer
consumer = Consumer({"bootstrap.servers":"localhost:9092", "group.id":"file-preview", "auto.offset.reset":"earliest", "enable.auto.commit":False, "enable.auto.offset.store":False, "allow.auto.create.topics":False})
deadline = time.monotonic() + 20
try:
consumer.subscribe(["orders-file"])
while time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
if message.error() is not None:
raise RuntimeError(str(message.error()))
value = json.loads(message.value().decode("utf-8"))
record = {"topic":message.topic(), "partition":message.partition(), "offset":message.offset(), "key":message.key().decode("utf-8"), "value":value}
print(json.dumps(record, ensure_ascii=False))
break
else:
raise SystemExit("No preview record arrived before the deadline.")
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/preview_consumer.py > /home/labex/project/output/one-event.json
cat /home/labex/project/output/one-event.json
Expect one JSON envelope whose value contains one of py1 through py6 and whose key matches that event's order_id. Across three partitions, the first observed event is not fixed. The partition and offset identify its actual broker record.
Closing this preview does not commit progress because automatic commits are disabled and the script made no explicit commit. The next application uses its own group for the file handoff.
Write and Flush Each Envelope Before Committing Its Message
In this step, you will build the bounded file consumer. Read its processing order carefully: decode and validate the JSON object and key, write one JSONL envelope, flush the file, synchronously commit that message, then count it as completed. If a write, decode, or commit fails, the application exits rather than skipping the error and later pretending the batch succeeded.
The with statement closes the local file when its block exits. Path identifies the local filename. The output opens in append mode so a restart does not silently truncate earlier delivery evidence. The clean first run starts with no file. max-messages is the target for this invocation; the overall deadline makes an incomplete run exit nonzero. JSONL stores one complete JSON object per line, without a surrounding array. flush() transfers Python's buffered file data to the operating system before the commit request. It does not guarantee survival of a power or disk failure, and it does not make the file and Kafka commit one atomic transaction. This exercise observes process interruption and saved progress in the running VM.
commit(message=..., asynchronous=False) commits the handled message's next offset for its partition. The returned per-partition errors are checked. enable.auto.offset.store=false prevents unrelated automatic progress storage, while the explicit message argument identifies exactly which handled position is being committed.
Save the complete supplied parser/loop structure with this processing sequence:
cat > /home/labex/project/consumer.py <<'PYTHON'
import argparse,json,sys,time
from pathlib import Path
from confluent_kafka import Consumer
parser = argparse.ArgumentParser(description="Save a finite JSON event delivery and commit only written records.")
parser.add_argument("--topic", required=True)
parser.add_argument("--group", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--max-messages", type=int, default=6)
parser.add_argument("--timeout-seconds", type=float, default=30)
parser.add_argument("--bootstrap-server", default="localhost:9092")
args = parser.parse_args()
if args.max_messages <= 0 or args.timeout_seconds <= 0:
parser.error("Message count and timeout must be positive.")
consumer = Consumer({
"bootstrap.servers": args.bootstrap_server,
"group.id": args.group,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
})
received = 0
deadline = time.monotonic() + args.timeout_seconds
try:
consumer.subscribe([args.topic])
with Path(args.output).open("a", encoding="utf-8") as sink:
while received < args.max_messages and time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
position = f"topic={message.topic()} partition={message.partition()} offset={message.offset()}"
if message.error() is not None:
raise RuntimeError(f"Kafka error {position}: {message.error()}")
try:
key = message.key().decode("utf-8") if message.key() is not None else None
value = json.loads(message.value().decode("utf-8"))
if not isinstance(value, dict) or not isinstance(value.get("event_id"), str) or not isinstance(value.get("order_id"), str):
raise ValueError("Expected an object with string event_id and order_id.")
if key != value["order_id"]:
raise ValueError("Record key does not match order_id.")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
print(f"Decode failure {position}: {error}", file=sys.stderr)
raise SystemExit(1)
record = {"topic": message.topic(), "partition": message.partition(), "offset": message.offset(), "key": key, "value": value}
sink.write(json.dumps(record, ensure_ascii=False) + "\n")
sink.flush()
committed = consumer.commit(message=message, asynchronous=False)
if not committed or any(item.error is not None for item in committed):
raise RuntimeError(f"Commit failed {position}: {committed}")
received += 1
print(f"SAVED_AND_COMMITTED event_id={value['event_id']} {position} next_offset={message.offset()+1}")
if received < args.max_messages:
print(f"Only received {received} of {args.max_messages} requested records before the deadline.", file=sys.stderr)
raise SystemExit(2)
print("Saved and committed records:", received)
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/received-events.jsonl --max-messages 6 --timeout-seconds 30 > /home/labex/project/output/consumer.log 2>&1
cat /home/labex/project/output/consumer.log
cat /home/labex/project/output/received-events.jsonl
Expect Saved and committed records: 6 and six JSONL envelopes. Every envelope should preserve the full value, matching key, and actual record position. A callback or console log is not a substitute for the saved file; the next step compares file evidence with actual committed progress.
Connect the File to Per-Partition Committed Next Offsets
In this step, you will inspect coordinator progress using the command-line tool you already know:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group file-writer --describe > /home/labex/project/output/file-writer-position.txt
cat /home/labex/project/output/file-writer-position.txt
For each partition containing delivered records, CURRENT-OFFSET should reach that partition's LOG-END-OFFSET, with lag zero. Exact partition assignments and distribution are not fixed by this lesson. A partition with no records may have no committed position yet; there is no handled message there to commit.
Compare a saved envelope's offset with its partition's committed next offset. If a partition's last handled record is at offset n, the next committed position is n+1. That next position is not a global count across all partitions.
The consumer has closed, so a no-active-members message is normal. The stored file and saved group progress remain. You have separate evidence of processing output and coordinator state, without claiming they were written in one transaction.
Restart the Same Group and Observe No New Delivery
In this step, you will test the saved position instead of assuming earliest always rewinds. The six records are still retained, but file-writer has valid committed end positions. Predict another startup with that same group: it should find no new records, even though auto.offset.reset is earliest.
Use a separate output path to retain the original six-record file. Request one record with a short overall deadline. Because the requested count cannot be reached, this application deliberately reports a nonzero incomplete-batch status. The next command is expected to fail.
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/no-new-events.jsonl --max-messages 1 --timeout-seconds 5 > /home/labex/project/output/same-group-restart.log 2>&1
Capture its status immediately:
printf '%s\n' "$?" > /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-restart.log
Expect a nonzero status and Only received 0 of 1 requested records before the deadline. The new file should be empty; the original received-events.jsonl still contains six envelopes. The same group's valid committed offsets were not replaced by its earliest starting default.
To intentionally read retained history again, use the distinct-group or inactive-group reset workflow already taught. If a process stops after writing but before committing, the relationship is different: the same group may receive the record again. The next lab will make that failure window visible.
Summary
You subscribed and polled with Python, decoded JSON/UTF-8 records, and preserved keys and Kafka positions in JSONL. Successful flushes preceded explicit synchronous commits, and actual coordinator state matched the completed file. A same-group restart found no new records without deleting the retained topic history. Next, decoding failure and a write-before-commit interruption will show why records can be delivered again.
