介绍
一家工作室保留了投票历史，而原展示应用已暂停。你需要在不回退展示应用进度、不改变日志的情况下得到新计数。新消费者尚未收到冻结批次时，一次轮询可能短暂返回没有消息。
你将构建原生回放循环，为每个分区分别获取结束位置，并且只在读完整个有限批次后才发布结果。前一个实验中的逻辑身份函数已在这个独立 VM 中提供。你需要具备那里使用的 Python 和 Kafka 基础。请使用 Terminal。作者估计需要 50–60 分钟，包含阅读和普通错误处理，并非初学者计时试验。
这是冻结批次的重建。实验期间不会继续向这些主题发布消息，也不需要实现事务；不承诺可靠的崩溃恢复或端到端恰好一次处理。
获取每个分区的边界并读到边界
本步骤将在已准备的外围程序内部，实现完整的独立原生读取。请从 Terminal 开始。编辑 replay.py，必要时可编辑 reducer.py 或 helpers/ 下的 Python 辅助程序；派生文件写在 output/ 下。保留所提供的工具/数据、原主题和已暂停的展示应用消费组：
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
只读描述显示真实的 studio-votes 分区 ID、保留起点 low、当前冻结末端和副本。分区是一段有序日志，并非独立 VM。末端是保留批次之后的下一个位置，不是需要计数的一条记录。预测哪些分区需要读取记录：空分区满足 low == high，已经完成。
偏移量跨度不等于记录条数。Kafka 可能跳过内部/控制位置或已移除的位置。此来源包含一个原生跳过位置；不需要凭空补造一张缺失选票。检查真实封装记录，将其实际位置与描述比较：
bigdata-python -B vote_tool.py inspect main
重建前，用先修课程中的消费组工具读取已暂停展示应用保存的位置：
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
记下每个 CURRENT-OFFSET，以便稍后比较。没有活跃成员是正常的：展示应用已暂停，但仍有保存的进度。这里的 LAG 是日志位置差，不是逻辑选票数。你的独立消费者不得推进该组进度。
阅读已准备的外围程序
cat replay.py
导入、arguments()、客户端配置、错误报告和 finally/关闭逻辑都已提供。核心部分有意引发 NotImplementedError；它不会读取分区，也不会发布样例答案。前一个实验中正确的身份函数已单独准备在 reducer.py 中。
客户端使用所选 bootstrap.servers 地址及独立的 group.id。allow.auto.create.topics 为 false。自动提交和自动偏移存储已关闭。assign 将赋予明确的起始位置，不使用动态消费组分配；本示例没有显式提交调用。等效的独立消费组实现可以保存自己的进度，但绝不能加入原展示应用组、向其提交、重置或删除该组。
已准备的 vote_support.py 提供 arguments()，用于主题/目录/输出/broker/消费组参数，以及用于写出完整结果的 finish(...)。它不会发现分区或获取选票。finish 在写入/替换结果之前调用身份规则。finally 在成功或失败时均关闭消费者；保留这个外围结构。
分成简短且相互连接的部分实现原生核心
使用前一个实验中的编辑器流程：
nano replay.py
保留导入、参数解析、Consumer 配置、try:、except 和 finally 部分。在 try: 内，只删除两条解释性注释和表示核心未完成的 raise NotImplementedError(...)。用方向键定位、Backspace/Delete 编辑；Ctrl+K 会剪切当前整行，因此只在确实打算替换的行上使用。
在这个空缺处插入以下相互连接的部分。它们构成一个核心：不要在 shell 中运行单个代码块，也不要保存/运行只完成一半的循环。完成所有部分，保留提供的外围错误/关闭框架，再保存并运行。四个空格让指令位于 try: 内，再加四个则进入循环或条件内部。展示的相对缩进属于 Python 的语义。
首先发现完整的真实分区集合。不指定主题调用 list_topics，可以避免为未知名称请求元数据。使用所选条目前先检查其是否可用。将发现的 ID 排序，使本示例的分配列表稳定；这不会产生全局事件顺序：
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
为每个已发现分区调用一次 get_watermark_offsets(..., timeout=10)。TopicPartition 标识这次请求的主题和分区。保留起点 low 和冻结末端 high，并将用于诊断的下一位置初始化为 low。不要在轮询循环内刷新 high：追赶不断增长的末端就不再是有限重新计数。本练习禁止向冻结来源追加或修改内容：
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
只将还有保留位置需要读取的分区放入 pending。assign 接收明确起始偏移量的列表，包括空分区的 low。每次运行都初始化新的物理记录列表。time.monotonic() 测量经过时间，不受系统时钟调整影响；截止时间限制未完成读取的等待：
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
现在开始循环。poll(0.2) 可能返回消息、携带错误/事件的消息，或在超时时返回 None。预测为什么 None 不能表示完成：客户端可能只是尚未收到某条冻结记录。应在整体截止时间内继续等待。配置中已开启分区 EOF 事件；_PARTITION_EOF 是预期的末端事件，其他错误则让本次运行失败。这使原生进度能够越过最后一个内部位置：
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
用以下 elif 继续同一个 if message is not None 块，与 if message.error(): 对齐，缩进十二个空格。只有待处理分区中、位于固定 high 之前的普通消息才成为封装记录。也要保留物理重试；稍后由 summarize 处理逻辑身份。先解码 UTF-8 键/值字节，再用 json.loads 将值转换为字典：
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
处理本次轮询后，为每个剩余分区查询 consumer.position(...)。这是该消费者的下一位置，与任何消费组已提交的进度不同。将各位置与先前获取的 high 比较，不要猜测记录条数。只从 pending 移除已完成分区。for position 行缩进八个空格，位于 while 内、消息条件外，因此即使空闲轮询后也会检查位置。最后，在循环完成之后，以四个空格缩进调用 finish：
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
现在每个分区都有固定边界。空分区不需要选票，跳过的位置不需要编造记录，None 也没有终止读取。只有全部待处理分区的下一位置都到达各自 high，才发布完整快照。
按 Ctrl+O，用 Enter 确认 replay.py，再按 Ctrl+X 保存退出。检查保存的程序：
cat replay.py
检查六个部分都在提供的 try 内，封装处理仍在普通消息分支内，位置检查仍在循环内，而 finish 位于循环外。立即对所选原生批次运行：
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
预期有八条物理记录、五张逻辑选票，且 cedar=3、maple=2、willow=0。公共入口先在一次性样例上检查身份函数，再记录实际原生程序/辅助进程树从 broker 接收的数据。另一项完整业务比较检查真实结果。快照报告的边界/位置用于诊断，不是原生处理的独立证明。
每次成功调用都必须实际写出结果，可以直接写入，也可以写临时文件后重命名为所选输出。重复时字节相同合法，辅助子进程也可写入。只保留旧文件、更新旧文件时间戳或写完成标记，不构成发布。提供的 finish 使用所教的临时文件/替换方法。使用这个已准备的写入器时，发布前的循环或验证错误会保留先前快照；入口不会回滚任意原位写入。
遇到语法、名称或缩进错误时，依据报告的行号重新打开 nano replay.py，修正该部分，保留已正常工作的部分。保存退出，用 cat replay.py 查看，再运行公共入口。遇到回放未完成超时时，检查所选主题参数以及 pending、EOF 和位置处理；这不构成回退原消费组的理由。用 pwd 和 ls replay.py reducer.py 定位拼错的路径。本地程序错误无需重跑 setup 或改变源记录。
从零重建并使用另一批数据
本步骤将证明再次运行能重建相同的有限结果，然后改用新的目录和分区分布。继续在 Terminal 的 /home/labex/project 下操作。
已准备的写入器先按分区/偏移量排列物理封装记录，再按 ID 排列逻辑选票。因此完成的同一快照可以保持稳定，但并不声称事件具有全局顺序。sha256sum 计算内容指纹。> 将指纹保存在输出目录下；稍后用 -c 将实际指定文件与保存的指纹比较：
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
预期看到 output/main-snapshot.json: OK。每次调用程序都会从空的 records、seen 和计数开始，替换完整快照，而不是把历史选票再次加到旧总数上。消费并没有删除这些记录，也没有回退已暂停的展示应用。
重复运行后，再次读取原展示应用位置：
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
每个 CURRENT-OFFSET 都必须与之前记录的展示应用进度表一致。虽然独立重建读取了保留记录，但没有向该组提交或回退它。没有活跃成员的提示仍然是正常的。
列出 check 批次并查看其自己的展位目录：
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
这一批使用不同展位名称，分区数也从四个变为三个。保持同一份程序文件。解析器选择实际主题和目录，发现操作则找出完整的真实分区集合：
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
预期有六条物理记录、四张逻辑选票，且 amber=2、violet=2、moss=0。硬编码第一个目录，或按第一个主题的记录条数停止，都不能正确描述这一批数据。
回到第一份真实文件：
bigdata-python -B read_snapshot.py output/main-snapshot.json
它仍然包含工作室的结果。读取器直接打开指定文件，不会从 Kafka 补充任何内容。如果提示文件缺失，请将文件名与 vote_tool.py list 对照并修正路径。只有对应结果确实需要重建时，才重新回放该批次。
拒绝冲突并保留有效快照
本步骤将观察一次数据错误，同时不替换有用结果。继续在 Terminal 的 /home/labex/project 下操作。列出的 conflict 主题有意包含同一个 ID 对应不同有效展位内容的记录，其输出目标是现有主快照。没有获准采用的胜出值，必须拒绝成功发布。
上一步保存的指纹描述了有效主文件。用同一个程序运行冲突批次。下一条命令应失败：
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
预期看到非零退出状态以及指出实际 ID 的 Conflicting event_id。规则在准备好的写入器创建/替换结果之前执行。这里的 .pending/替换写入器是一种完整结果发布技术，并非崩溃恢复或持久性保证。
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
有效主文件必须仍然匹配指纹，保持工作室的原始计数。失败的冲突运行与成功的 main/check 运行分别记录。不要编辑冲突来源，也不要随意选择先到的事件。逆序或逐分区遍历也必须拒绝同一个冲突。
如果只是批次名拼错，入口会在开始回放前打印可用名称。用 vote_tool.py list 恢复，再运行正确名称。对于本地程序错误，修复文件并重新运行受影响的有效批次。任何源程序编辑后，都要重新运行两个有效批次，让结果对应当前程序文件，再重复冲突自检。
保持原展示应用消费组暂停：不得加入、提交、重置或删除。即使重置后恢复到相同最终偏移量，也不属于独立回放。功能检查组合了原生程序树接收事实与完整真实输出，但不证明计算因果，也不保证能抵御管理员伪造环境。
总结
你获取了真实的逐分区边界，跨越短暂空闲轮询、空分区和跳过的位置，读完整个有限日志。同一个原生程序重建了两份目录对应的结果，没有累积旧总数；内容冲突的 ID 导致运行失败，而有用的主快照和原消费者进度均得到保留。
