Introduction
The console producer proved the event and key model. An application needs one more distinction: calling a send API can place a record in a local client queue before the broker confirms it. A program that exits at that point may report success without delivering its records.
You will use the installed confluent-kafka Python client with the prepared local Apache Kafka broker. The package is a client library; no Confluent Cloud account is needed. You will serialize dictionaries to JSON/UTF-8 bytes, send order keys, and inspect delivery callbacks and the final queue state. An unreachable local address provides a bounded failure test.
This independent lab prepares a three-partition topic and two input fixtures. The later course activities assume Python functions, loops, dictionaries, JSON, and running scripts; the Python path can help with those prerequisites. Kafka installation and service configuration remain prepared. Use the Terminal.
Turn a Dictionary into a Key and Value on the Wire
In this step, you will inspect the conversion before contacting Kafka. Start with the six-event input:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Each line is a complete JSON object. json.loads converts that text into a Python dictionary. json.dumps converts the dictionary into JSON text, and encode("utf-8") converts text into bytes suitable for the client API. The key is separately encoded from order_id; it is not the event_id.
Save and run this small preview. The quoted here-document preserves the Python source literally:
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
Find the dictionary, JSON text, and byte representations. A Python bytes display begins with b; that display notation is not part of the transmitted value. The first key should encode order-4001. JSON preserves the named event fields, while UTF-8 provides the character encoding.
No event has been sent yet. The next step supplies these bytes to a producer and checks when the broker confirms them.
Publish and Confirm Every Actual Delivery
In this step, you will run a complete producer with supplied argument handling and a delivery callback. Producer configures the client; bootstrap.servers supplies the initial address. acks=all requests acknowledgment from the current in-sync replicas. This one-broker environment has one replica, so that setting does not demonstrate multi-broker fault tolerance. The provided murmur2_random routing setting keeps keyed routing compatible with the earlier Java console examples while the partition count stays fixed.
Read the key distinction in the program: produce queues a record locally. poll services callbacks, and flush waits for queued work while also servicing callbacks.
A callback is a function you give to the client library for it to call when a delivery result becomes available. Passing callback=delivery supplies the function itself; it does not call it immediately. poll(0) services available callbacks without waiting, while flush also waits for outstanding sends. The callback records either a broker-confirmed topic/partition/offset or a failure. After flush, the program checks remaining records, failures, and confirmed count before declaring success. A failure callback is collected and causes a nonzero process result; it is not silently printed and ignored.
The supplied argparse section maps command-line options to args attributes: for example, --input becomes args.input, and --delivery-timeout-ms becomes args.delivery_timeout_ms. Path(args.input).open(...) opens that selected local file; the loop reads one JSON event per nonempty line. You do not need to design an argument parser to use this client.
Save the complete program. The parser and callback structure are supplied, while the serialization and produce call apply the bytes you just inspected:
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
Run the producer and save both output streams. > directs standard output to the log; 2>&1 sends standard error to that same destination, so callback failures and connection diagnostics are retained too.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
Expect six DELIVERED callback lines and Delivered records: 6. Each callback's actual partition and offset can vary. A successful produce return alone would not establish these acknowledgments.
Independently inspect the broker with the already learned console consumer:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
Find py1 through py6 with keys matching their order_id values. The independent read connects the application's delivery report to real broker records; no fixed partition number or global display order is required.
Change One Field in a New Event and Reuse the Producer
In this step, you will reuse the same application for one new event. The prepared next-event fixture has a fresh event_id, py7, so it has not already been published. Inspect its channel field:
cat /home/labex/project/data/next-event.jsonl
Change only that unpublished event's channel from online to store in a separate working file. The event still refers to order-4001 and follows its earlier events with sequence 3:
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
Expect one successful new callback and seven retained events in the independent read. py7 should have channel store, while py1 through py6 retain their original content. Related order-4001 events follow their sequence within the same routed partition; cross-partition display order can vary.
The producer implementation did not change. A supplied input path let it serialize and deliver the new data instead of embedding a fixed training answer.
Observe a Bounded Failure at an Unreachable Address
In this step, you will check that queueing is not mistaken for delivery. The same client can create a local queue even when localhost:1 has no Kafka broker. The provided delivery timeout option limits message completion to 3000 milliseconds, and flush adds a small bounded margin. The real broker remains at localhost:9092.
Send the one-event file to that deliberately wrong address. The next command is expected to fail.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
Capture the status immediately before another command replaces it. $? is the preceding command's exit status: zero for success, nonzero for failure. printf saves it as a line of text:
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
Expect a nonzero status, connection diagnostics, a DELIVERY_FAILED callback, and Delivery incomplete rather than a successful final count. The program checked callback failures even when the queue finished draining after failed messages.
The failed address should not have added a second py7 to the real topic. All seven earlier successful events remain available. Correcting an endpoint and retrying a business event is a separate decision from assuming a queued record was acknowledged; this lab's deliberate failure sent nothing to the real broker. Next, you will decode received JSON, save it, and commit reading progress in a Python consumer.
Summary
You serialized dictionaries and order keys as UTF-8 bytes, published through a real Python client, and matched successful callbacks to actual broker positions. A changed new event reused the same program, while an unreachable address produced a bounded nonzero failure instead of false success. Next, a Python consumer will make decoding, file output, and write-before-commit behavior explicit.
