Python后端中间件专题30重放一天的审计事件——消费者组、乱序与去重审计员要重看昨天的工单变更。值班同事说“把 offset 往回拨不就好了”这句话只答了一半回拨让旧消息再次进入消费者却不会自动撤销或去重它已经写下的审计行。若业务效果没有唯一键重放一天就会把同一事件记两份。这一课把“消费进度”和“已经发生过的持久效果”分开。我们会在数据库故障时观察 offset 不前进在同一event_id重新出现时观察唯一约束让效果只存一份最后给出安全重放一天的操作决策而不是鼓励随便在现有消费组上 reset。上一课练习答案答案 EX-29-01 [code]保存为audit_fail_probe.py从project/运行fromticketflow.messaging.kafka_auditimportAuditConsumer,AuditEventclassFailingStore:defapply_once(self,event):raiseOSError(database unavailable)eventAuditEvent(event_idevt-30,tenant_idtenant-a,ticket_idticket-7,actionticket.updated,occurred_at2026-09-22T10:00:0000:00,payload{version:3},)commits[]consumerAuditConsumer(FailingStore(),commitlambdatopic,partition,offset:commits.append(offset),)try:consumer.consume(topicticketflow.audit,partition0,offset10,eventevent)exceptOSErroraserror:assertstr(error)database unavailableelse:raiseAssertionError(storage failure was swallowed)assertcommits[]print(failure leaves offset uncommitted)运行命令python audit_fail_probe.py预期输出failure leaves offset uncommitted真实消费进程还要设置enable_auto_commitFalse否则这段正确的控制流会被框架后台自动提交抵消。答案 EX-29-02 [prose]同一工单的 v2/v3 事件使用同一个tenant_id:ticket_idkeyKafka 把相同 key 映射到同一分区在同一分区里 offset 给出顺序。不同工单可能落在不同分区不存在跨分区全局顺序即使同一分区有序重试/异步投影仍要按业务 version 防止旧状态覆盖。同一event_id第二次出现时审计数据库唯一约束使apply_once返回 false既然此效果以前已经持久化消费者可以提交第二次投递对应的offset1。如果不是唯一冲突而是数据库断开、字段校验失败或其他异常就不能把它假扮成重复事件后推进 offset。Offset 是“从哪继续”不是“做过哪些效果”对每个 topic partitionKafka consumer group 保存下一条要读的 offset。消费者看到 offset 41处理成功后提交 42。崩溃有两个窗口SQL 审计行提交之前崩溃 → offset 未提交 → 再读 41 → 本次首次写入 SQL 审计行已提交、offset 42 尚未提交 → 再读 41 → unique event_id 命中 → 提交 42这就是数据库效果至少一次尝试、持久化一次的设计。没有一个跨 SQL 与 Kafka 的共同事务不能因为正常情况下只看到一行就对外承诺整条链路 exactly-once。AuditConsumer的先后顺序是 “store.apply_once返回后才commit(offset1)”SqlAlchemyAuditStore在独立 SQL 事务里插入AuditEventModel只把 event_id 唯一冲突识别为重复其他IntegrityError继续抛。历史 revision 本地合同 selector未运行python tools/verify_executable_revisions.py--lesson 30--python python第一条验证故障时不 commit第二条用 SQLite 验证 SQL 记录和唯一边界。它们不证明真实 Kafka group rebalance、offset reset 或 24 小时 retention。对重放的 broker 验收有更窄的前提只能在获授权的 Ubuntu 远端由远端流程准备好 Kafka profile并设置TICKETFLOW_REMOTE_TESTS1后执行python scripts/run_remote_selector.py--checkpoint 30--selector tests/integration/test_kafka_remote.py::test_bounded_replay_preserves_live_offsets_and_business_effect_counts它观察独立 replay group、每个分区的[start,stop)、live group offset 不变、审计行按 event_id 去重以及 notification/SLA 的业务 effect count 不变。这里没有 Windows 本地 Docker 命令未运行的远端结果仍是SKIP/PENDING不是 PASS。如何“重放一天”而不伤正在运行的消费组先回答业务问题是重新生成审计报表还是修复漏处理的事件两者都要界定 topic、时间范围、租户、工单 ID、版本和目标存储。检查 topic retention 是否覆盖昨天记录每个 partition 的起止 offset不同 partition 的 timestamp→offset 不是一个全局数字。更安全的办法是为调查使用单独的 replay consumer group或暂停原组、精确记录原位置并受控 reset。不能一边让原组继续推进一边随手修改它的 offset。重放程序仍调用同一个持久化幂等边界只想做报告时可以写到专用只读输出/独立结果表不应把历史 event 重新触发通知、SLA 或其他业务副作用。auto_offset_resetearliest只在该 group 没有已提交位置时生效它不会自动把现有 group 从今天跳回昨天。重放实验要明确 group ID 与 offset 操作、每分区范围和停止条件。保留原来的 offset 清单故障后才有可靠恢复点。执行边界把“安全”写进运行入口最终项目的受限入口是python -m ticketflow.messaging.kafka_probe replay它创建ReplayPlan后交给run_bounded_replay()。live 与 replay group 必须不同范围永远写成[start, stop)所以 stop 不是一条要处理的消息。程序在seek之前捕获该分区的end_offsets新写入不会让这次调查的上界悄悄变长stop_offset captured_end直接失败根本不开始 seek。每收到一条记录都要求 offset 恰好等于下一期望值跳号是范围不连续立即失败。消费者本身有有限 idle timeout外层还有独立 wall-clock deadline到 exclusive stop 前空闲、或者时钟到了 deadline都失败而不是把已经做了一半的工作说成成功。SQLapply_once先执行成功包括唯一 receipt 表示的重复才提交 replay group 的下一个 offsetSQL exception 不提交当前 offset。planplan_replay(live_groupticketflow-audit-v1,replay_groupticketflow-replay-20260922,topicticketflow.audit.v1,partition2,start_offset120,stop_offset160,)resultrun_bounded_replay(plan,consumer,applyapply_audit_receipt,partition_factoryTopicPartition,captured_end_offsetlambdaitem:consumer.end_offsets([item])[item],deadline_seconds20.0,)kafka_replay_runner.execute_audit_replay()只注入SqlAlchemyAuditStore作为这个 audit-only effect它不导入通知或 SLA dispatcher。于是重放能补审计投影却不会把昨天的通知再发一遍、也不会改变 SLA 业务计数。用数据库唯一键观察两次应用保存下面完整的sql_audit_probe.py它在内存 SQLite 上创建课程模型表消费同一事件两次并检查持久行数fromdatetimeimportdatetime,timezonefromsqlalchemyimportcreate_engine,func,selectfromsqlalchemy.ormimportsessionmakerfromsqlalchemy.poolimportStaticPoolfromticketflow.messaging.kafka_auditimportAuditConsumer,AuditEvent,SqlAlchemyAuditStorefromticketflow.storage.modelsimportAuditEventModel,Base enginecreate_engine(sqlitepysqlite:///:memory:,connect_args{check_same_thread:False},poolclassStaticPool,)Base.metadata.create_all(engine)sessionssessionmaker(bindengine,expire_on_commitFalse)eventAuditEvent(event_idevt-30-sql,tenant_idtenant-a,ticket_idticket-7,actionticket.updated,occurred_atdatetime(2026,9,22,tzinfotimezone.utc).isoformat(),payload{version:3},)commits[]consumerAuditConsumer(SqlAlchemyAuditStore(sessions),commitlambdatopic,partition,offset:commits.append(offset),)try:firstconsumer.consume(topicticketflow.audit,partition1,offset41,eventevent)againconsumer.consume(topicticketflow.audit,partition1,offset42,eventevent)withsessions()assession:rowssession.scalar(select(func.count()).select_from(AuditEventModel))assertfirst.appliedisTrueandagain.appliedisFalseassertrows1andcommits[42,43]print(audit rows1; offsets42,43)finally:engine.dispose()运行命令python sql_audit_probe.py预期输出audit rows1; offsets42,43这段有真实 SQL 唯一约束却仍不是 PostgreSQL/Kafka 分布式验收。它让“数据库效果一次、offset 可两次推进”的不同职责可见。错误处理与恢复判据如果审计行重复检查是否所有消费者共用相同 event ID、唯一约束是否真的迁移到了远端数据库、是否把同一动作重新包装成新 UUID。若 offset 已前进而行数没增加检查 auto commit 和异常捕获若消费者反复卡在某个坏 JSON不应无边界重试或手工跳过应先保存原消息与原因制定 quarantine/人工修复再有证据地推进位置。恢复不能只看 consumer 进程 Running。要比较每分区 lag、计划重放范围、audit_events 去重计数、event_id与工单事实的对应以及可选 profile 关闭后主 API 是否仍能创建工单。本课练习练习 EX-30-01 [code]写audit_sql_duplicate_probe.py创建内存 SQLite 课程模型表用SqlAlchemyAuditStore.apply_once()连续存同一AuditEvent两次断言返回True, False且audit_events只有一行打印True False 1。第 31 课给完整答案。练习 EX-30-02 [prose]如果原 consumer group 已提交到每分区 offset 500但需要重读昨天的 100–200为什么设置auto_offset_resetearliest不起作用给出一套不影响原消费组的重放安排与防止重放触发通知的边界。下一篇把局部故障组合起来第 31 课暂时放下可选 Kafka按 Redis、RabbitMQ、ES、一台 API 的顺序逐个停止并恢复。每步都要记录“哪些请求应该成功/失败”和“恢复后还有没有确切的一份通知效果”不能用容器 Up 当作验收。本课完整核心模块messaging/kafka_replay.py这是 lesson-30 实际使用的 bounded replay primitive。把它和上一课的kafka_audit.py放在一起读上一课负责“审计效果后才推进 live offset”本课负责“另开 group、冻结 end、只走[start,stop)并把失败保留成失败”。Bounded, audit-only Kafka replay primitives. The caller supplies the SQL effect. This module deliberately knows nothing about notification or SLA workers, so replay cannot dispatch business work. from__future__importannotationsfromdataclassesimportdataclassfrommathimportisfinitefromtimeimportmonotonicfromtypingimportCallable,Protocoldataclass(frozenTrue,slotsTrue)classReplayPlan:live_group:strreplay_group:strtopic:strpartition:intstart_offset:intstop_offset:intdataclass(frozenTrue,slotsTrue)classReplayResult:records_seen:inteffects_inserted:intfinal_replay_offset:intclassReplayRecord(Protocol):topic:strpartition:intoffset:intclassReplayConsumer(Protocol):config:dict[str,object]defassign(self,partitions:object)-None:...defseek(self,partition:object,offset:int)-None:...defcommit_offset(self,topic:str,partition:int,offset:int)-None:...def__iter__(self):...defplan_replay(*,live_group:str,replay_group:str,topic:str,partition:int,start_offset:int,stop_offset:int,)-ReplayPlan:Validate and describe an explicit half-open replay range.ifnotlive_groupornotreplay_group:raiseValueError(live and replay groups are required)iflive_groupreplay_group:raiseValueError(replay group must be different from the live group)ifnottopic:raiseValueError(replay topic is required)ifpartition0orstart_offset0orstop_offsetstart_offset:raiseValueError(replay requires a non-negative, non-empty half-open range)returnReplayPlan(live_grouplive_group,replay_groupreplay_group,topictopic,partitionpartition,start_offsetstart_offset,stop_offsetstop_offset,)defrun_bounded_replay(plan:ReplayPlan,consumer:ReplayConsumer,*,apply:Callable[[ReplayRecord],bool],partition_factory:Callable[[str,int],object],captured_end_offset:Callable[[object],int],deadline_seconds:float20.0,clock:Callable[[],float]monotonic,)-ReplayResult:Apply exactly [start_offset, stop_offset) and commit only afterward. Consumers used here must also have a finite idle timeout. The wall-clock deadline is a second, independent bound and an incomplete range is always an error rather than a partial-success result. ifnotisfinite(deadline_seconds)ordeadline_seconds0:raiseValueError(replay deadline must be a positive finite duration)ifconsumer.config.get(group_id)!plan.replay_group:raiseValueError(consumer must use the independent replay group)deadlineclock()deadline_seconds partitionpartition_factory(plan.topic,plan.partition)log_endcaptured_end_offset(partition)ifclock()deadline:raiseTimeoutError(replay wall-clock deadline reached before seeking)ifplan.stop_offsetlog_end:raiseValueError(fstop offset{plan.stop_offset}exceeds captured log end{log_end})consumer.assign((partition,))consumer.seek(partition,plan.start_offset)records_seen0effects_inserted0next_offsetplan.start_offsetforrecordinconsumer:ifclock()deadline:raiseTimeoutError(replay wall-clock deadline reached before stop offset)ifrecord.topic!plan.topicorrecord.partition!plan.partition:raiseRuntimeError(consumer returned a record outside the replay partition)ifrecord.offset!next_offset:raiseRuntimeError(freplay range is not contiguous at offset{next_offset})ifrecord.offsetplan.stop_offset:raiseRuntimeError(consumer passed the exclusive replay stop offset)appliedapply(record)next_offsetrecord.offset1consumer.commit_offset(record.topic,record.partition,next_offset)records_seen1effects_insertedint(applied)ifnext_offsetplan.stop_offset:returnReplayResult(records_seen,effects_inserted,next_offset)raiseTimeoutError(freplay consumer became idle before stop offset{plan.stop_offset}; ffinal replay offset was{next_offset})