使用 Python 消费并保存 Kafka JSON 事件

简介

消费者显示了事件，并不代表已经保存了可供交付的结果。本应用会解码每个值，将其 Key 和 Kafka 位置保存在本地 JSONL 文件中，并且只在对应写入成功且完成刷新后提交进度。

你将把已有的消费者组与 offset 模型对应到 Python：subscribe 选择 Topic，poll 请求记录，显式同步提交为处理过的消息保存下一个 offset。自动提交和自动 offset 存储均被禁用，避免它们独立于你的代码推进进度。文件写入与 Kafka 提交仍是独立操作，并非跨系统的原子事务。

这个独立实验在新 Topic 中准备了生产者实验里熟悉的六条初始订单事件，不依赖之前的 VM 状态。你需要了解 Python 函数、循环、字典、JSON 和脚本运行。解析器和截止时间控制结构已经提供。请使用终端和已安装的 bigdata-python 环境。

订阅、轮询并解码一条真实记录

这一步，你将检查一条接收到的记录，但不提交预览进度。先读取准备好的事件值：

cd /home/labex/project

cat /home/labex/project/data/events.jsonl

Consumer 配置了消费者组标识和初始 earliest 策略。subscribe 注册需要读取的 Topic；poll 返回消息、带错误的消息，或在没有可用记录时返回 None，分配过程中也可能如此。循环会在总体截止时间内持续轮询。

提供的控制流结构确保读取有时间限制。time.monotonic() 测量经过的时间，不依赖系统日期。没有消息就绪时，continue 开始下一次轮询；显示记录后，break 跳出循环。只有循环结束时没有执行该 break，while 的 else 才会运行，在这里表示截止时间前没有记录到达。

无论读取成功还是抛出错误，finally 都会关闭消费者。可以将这段框架理解为 Kafka 调用外围的等待与清理逻辑；你无需设计重试框架。

这个预览使用独立的组 file-preview，两项自动设置都已禁用。解码后的值是 Python 字典；外层封装还保留 Topic、分区、offset 和解码后的 Key。程序不会调用 commit：

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

预期会看到一个 JSON 封装对象，其中 value 包含 py1 到 py6 中的一条事件，key 与该事件的 order_id 匹配。在三个分区之间，最先观察到的事件并不固定。分区和 offset 标识它在 Broker 中的真实记录。

关闭这个预览不会提交进度，因为自动提交被禁用，脚本也没有显式提交。下一个应用将使用自己的组完成文件交付。

提交消息前写入并刷新每个封装对象

这一步，你将构建有运行时限的文件消费者。仔细阅读处理顺序：解码并验证 JSON 对象和 Key，写入一个 JSONL 封装对象，刷新文件，同步提交该消息，然后将它计为已完成。如果写入、解码或提交失败，应用会退出，不会跳过错误后再声称批次成功。

with 语句会在代码块退出时关闭本地文件。Path 标识本地文件名。输出以追加模式打开，避免重启时悄悄截断之前的交付证据。干净的首次运行开始时还没有该文件。max-messages 是本次调用的目标数量；总体截止时间使未完成的运行以非零状态退出。JSONL 每行保存一个完整 JSON 对象，外面没有数组包裹。flush() 会在提交请求前，将 Python 缓冲的文件数据交给操作系统。它不保证数据能承受断电或磁盘故障，也不会让文件写入与 Kafka 提交成为一个原子事务。本练习观察的是运行中 VM 内的进程中断和保存进度。

commit(message=..., asynchronous=False) 为已处理消息所在的分区提交下一个 offset。程序会检查返回的各分区错误。enable.auto.offset.store=false 防止无关的自动进度存储，而显式 message 参数精确标明当前提交的是哪个已处理位置。

保存完整的预置解析器与循环结构，其中采用了上述处理顺序：

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

预期会看到 Saved and committed records: 6 和六个 JSONL 封装对象。每个封装对象应保留完整的值、匹配的 Key 和真实记录位置。回调或控制台日志不能替代保存的文件；下一步将比较文件证据与真实已提交进度。

将文件与各分区已提交的下一个 offset 关联

这一步，你将使用已经熟悉的命令行工具检查协调器进度：

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

对于每个包含已交付记录的分区，CURRENT-OFFSET 应到达该分区的 LOG-END-OFFSET，且 lag 为零。本课不限定具体的分区分配和记录分布。没有记录的分区可能还没有已提交位置，因为其中没有已处理的消息可供提交。

比较已保存封装对象的 offset 与其分区已提交的下一个 offset。如果某分区最后处理的记录位于 offset n，那么提交的下一个位置是 n+1。这个下一个位置不是跨所有分区的全局计数。

消费者已经关闭，因此出现没有活跃成员的提示是正常的。存储的文件和保存的组进度仍然存在。你分别获得了处理输出和协调器状态的证据，但不声称它们在同一个事务中写入。

重启同一组并观察没有新交付

这一步，你将测试已保存位置，而不假定 earliest 总会回退。六条记录仍被保留，但 file-writer 已有有效的已提交末尾位置。预测使用同一组再次启动的结果：即使 auto.offset.reset 为 earliest，也应找不到新记录。

使用单独的输出路径，以保留原有六记录文件。在较短的总体截止时间内请求一条记录。由于无法达到请求数量，应用会有意报告非零的批次未完成状态。下一条命令预期会失败。

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

立即保存其状态：

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

预期状态为非零，并显示 Only received 0 of 1 requested records before the deadline.。新文件应为空；原始 received-events.jsonl 仍包含六个封装对象。同一组有效的已提交 offset 没有被 earliest 默认起始策略替换。

若要有意再次读取保留历史，请使用已经学过的独立组或非活跃组重置流程。如果进程在写入后、提交前停止，情况则不同：同一组可能再次收到该记录。下一个实验将展示这个故障窗口。

总结

你使用 Python 订阅并轮询，解码 JSON/UTF-8 记录，将 Key 和 Kafka 位置保存在 JSONL 中。成功刷新发生在显式同步提交之前，真实协调器状态与完成的文件相匹配。同组重启没有找到新记录，而保留的 Topic 历史并未删除。接下来，解码失败和写入后提交前的中断将展示记录为什么可能再次交付。