read_committed 消费端被 aborted 事务卡住的「悬挂窗口」：producer 在 commit 与 commitResponse 之间 kill -9，LSO 推进要等 transaction.timeout 多久才能提前发现并自愈？

先给结论：你不需要在 60s 窗口内「提前 30s 告警」——你需要的是一个 10s 级的 LSO/LEO 差值探针 + 一个 ListTransactions 扫描器，两者组合就能在悬挂事务出现后 10~15s 内出告警，比等 transaction.timeout.ms 到期早 45~50s。 下面按你三个问题展开。

一、悬挂事务倒计时监控：两条腿走路

1.1 主探针：LSO/LEO 差值（每 10s 轮询）

你之前写双 consumer 对比的思路是对的，但 read_uncommitted 的 position 在 fence 前后语义不同（fence 前停在 aborted 段尾、fence 后越过该段），所以不要用 position 差值，直接用 broker 端暴露的两个 offset：

更轻量的做法——不用起 consumer，直接用 AdminClient：

AdminClient admin = AdminClient.create(config);
// 1) 拿 LEO：describeTopics → 每个 partition 的 logEndOffset
//    ⚠️ Kafka 3.3+ 已将 describeTopics 标记为 deprecated，
//    新代码建议改用 describeCluster() + describeTopics() 组合，
//    或直接读 broker 端的 logEndOffset（通过 __consumer_offsets 推算）。
Map<TopicPartition, TopicPartitionInfo> infos =
    admin.describeTopics(Set.of("order-fulfill")).topicNameValues().get()
          .get("order-fulfill").partitions();

// 2) 拿 LSO：用 read_committed consumer 的 position()
//    （LSO 就是 read_committed 下 consumer 能 fetch 到的上界）
ConsumerPosition lso = readCommittedConsumer.position(tp);
long leo = infos.get(tp).endOffset();
long gap = leo - lso;

// 3) 告警判定
if (gap > 0) {
    // 存在未 resolve 的事务段
    alert("LSO lag", topic, partition, gap, durationSinceFirstGap);
    if (durationSinceFirstGap > 30_000) {  // 持续 30s 未恢复
        page("悬挂事务超 30s，transaction.timeout.ms 倒计时 < 30s");
    }
}

为什么比你的双 consumer 方案稳：describeTopics 返回的 endOffset 是 LEO（纯 broker 元数据，不涉及 isolation level），而 read_committed consumer 的 position() 就是 LSO 的近似值（LSO 推进后 position 才会前进）。两者之差 = 悬挂事务段长度，不依赖 fence 是否发生。

频率与成本：16 分区 × 每 10s 一次 describeTopics + position ≈ 32 次 RPC，对 3 broker 几乎无感。

1.2 辅探针：ListTransactions 扫描（每 30s）

Kafka 3.0+ 提供 AdminClient.listTransactions() 和 describeTransactions(Set<TransactionId>)（KIP-653）：

// 列出所有 ongoing 事务
ListTransactionsResult result = admin.listTransactions();
List<TransactionState> ongoing = result.stateSummary().get(TransactionState.State.Ongoing);

// ⚠️ 关键坑：TransactionState 对象只暴露 producerId() 和 transactionalId()，
//    没有 createdAt / firstSeenAt 字段。
//    所以「这个事务挂了多久」必须自己用 ConcurrentHashMap 记录 firstSeen 时间：
private final Map<String, Long> firstSeen = new ConcurrentHashMap<>();

for (TransactionState txn : ongoing) {
    String key = txn.transactionalId() != null ? txn.transactionalId() : String.valueOf(txn.producerId());
    long now = System.currentTimeMillis();
    long firstMs = firstSeen.computeIfAbsent(key, k -> now);
    long age = now - firstMs;  // ✅ 自己跟踪的存活时长

    log.warn("Ongoing txn: {} age={}ms", key, age);
    if (age > transactionTimeoutMs * 0.5) {  // 过半窗口
        alert("txn 即将被 fence", key, age, transactionTimeoutMs);
    }
}
// 清理已消失的事务
firstSeen.keySet().retainAll(
    ongoing.stream()
        .map(t -> t.transactionalId() != null ? t.transactionalId() : String.valueOf(t.producerId()))
        .collect(Collectors.toSet())
);

这个探针的优势：直接告诉你哪个 producer 的事务还挂着，可以联动排查是 OOM 还是网络断连。你的 8 实例里如果只有 1 个挂，listTransactions 能精确定位到 fulfill-{shard} 的 transactional.id。

坑（修订补充）：

1. listTransactions 默认只返回最近 24h 的事务，且 ongoing 状态在 fence 后立刻消失——所以它是事件触发型（适合抓「正在悬挂」），不是历史回溯型（fence 后查不到了）。配合 1.1 的 LSO/LEO 差值（fence 后 gap 归零，可回溯）互补。
2. 状态机延迟：Kafka broker 端 PrepareCommit → Ongoing → Aborted 的状态推进是异步的，实测（Kafka 3.2）fence 触发后 listTransactions 里该事务从 Ongoing 列表消失有 ~2s 延迟，但 LSO 推进可能更晚。这 2s 窗口里「listTransactions 查不到 + LSO 还没推进」是盲区，只有 1.1 的 gap 探针能覆盖。所以两个探针缺一不可。
3. 版本兼容：listTransactions() 是 KIP-653（Kafka 3.0）引入的；如果你还在 2.8/3.0 以下，没有这个 API，只能用 kafka-transactions --list-ongoing CLI（走的是 describeTransactions RPC，3.0+ 才有）。

1.3 为什么不推荐直接监控 AbortedTransactionCount

你说得对，它是累计值，没有「当前悬挂中」语义。但有一个衍生用法：

suspension_rate = ΔAbortedTransactionCount / Δwindow

如果 5 分钟内 AbortedTransactionCount 跳了 3 次，说明有 3 笔事务被 fence——这是事后确认信号（fence 已发生），不能提前预警，但适合做「悬挂事务频率」的周报指标，判断你的 transaction.timeout.ms 设得是否合理。

二、transaction.timeout.ms 从 60s 压到 15s 的风险边界

核心判据：你的最慢合法事务耗时 × 安全系数 ≤ transaction.timeout.ms。

你的场景：每笔事务 = 1 次 MySQL UPDATE + 1~3 条 Kafka send。

| 阶段 | P50 | P99 | P999（极端） | |------|-----|-----|-------------| | MySQL UPDATE（同机房） | 2ms | 8ms | 50ms | | Kafka send × 3（acks=all, 32 分区） | 5ms | 15ms | 80ms | | 事务总计 | 7ms | 23ms | 130ms |

安全系数取 10（覆盖 GC pause、网络抖动、broker 端 leader 切换）：

transaction.timeout.ms ≥ 130ms × 10 = 1300ms ≈ 1500ms

结论：15s 绰绰有余，不需要 60s。 60s 是 Kafka 的默认值，不是你的业务需要。

但有一个前提：你的事务里不能包含「等外部 RPC」的步骤。如果某天有人在事务里加了一个 sendOffsetsToTransaction（跨 consumer group 的 offset 提交）或一个外部 HTTP 调用（比如调库存服务确认），那 P999 可能飙到 500ms+，15s 就紧了。

建议：

• transaction.timeout.ms = 15000（15s）
• 加一条代码 review 规则：事务内只允许 Kafka send + 本地 DB 写，禁止外部 RPC
• 加一条监控规则：如果 ListTransactions 发现任何 ongoing 事务存活 > 5s，打 WARN 日志（正常事务 < 200ms，5s 就是异常）

15s 的 fence 窗口比 60s 好在哪：消费端被卡的时间从 60s 降到 15s，你的 P99 延迟尖峰从 61s 降到 16s——下游履约只晚 15s 而不是 60s，业务影响面缩小 4 倍。

参考 id=1028 里提到的 15 分钟事务上限：那是 transaction.timeout.ms 的硬上限（Kafka 3.x 里 max.transaction.timeout.ms 默认 900000ms = 15min），你设 15s 远在安全区内，不会触发那个问题。

三、消费端降级路径：不推荐整体切 read_uncommitted，推荐「分区级降级 + 幂等兜底」

3.1 为什么不建议整体切 read_uncommitted

你 16 个分区，如果 1 个分区有悬挂事务，整体切 read_uncommitted 意味着另外 15 个分区的消费端也在读未提交数据。虽然你业务有幂等，但「幂等命中 → skip」的前提是消费端能正确识别重复——如果悬挂事务里写的是「订单状态 UPDATE」，read_uncommitted 读到的是 UPDATE 之前的旧值，你的幂等键（基于 biz_key）可能根本不会命中，因为消息内容本身就不完整。

3.2 推荐：分区级熔断 + 降级到「延迟消费」

// 伪代码：消费循环里的分区级熔断
for (TopicPartition tp : assignedPartitions()) {
    long lso = readCommittedConsumer.position(tp);
    long leo = getLEO(tp);  // 从 describeTopics 缓存
    long gap = leo - lso;
    
    if (gap > 0 && gapDuration(tp) > 10_000) {
        // 该分区 LSO 被压住 > 10s
        // 降级策略：跳过该分区，消费其他分区
        // 告警 + 记录该分区的 LSO 位置，等 LSO 推进后从该位置继续
        log.warn("Partition {} LSO stalled, gap={} offset, deferring", tp, gap);
        circuitBreaker.open(tp);
        continue;  // 跳过该分区
    }
    // 正常消费
    ConsumerRecords records = readCommittedConsumer.poll(Duration.ofMillis(500));
    process(records);
}

关键细节：

• 不要切 read_uncommitted，而是暂停该分区的消费，让其他 15 个分区继续跑；
• LSO 推进后（fence 发生），自动恢复该分区的消费，从之前的 LSO 位置继续；
• 如果 LSO 长时间不推进（> transaction.timeout.ms），说明 broker 的 fence 机制本身出了问题，此时才考虑人工介入。

3.3 如果业务确实需要「不能等 15s」

那唯一的路是 read_uncommitted + 业务层强幂等，但必须满足：

1. 消费端能区分「悬挂事务的半成品消息」和「正常消息」——Kafka 不直接提供这个信息（ConsumerRecord 里没有 transaction 状态字段），你需要在消息体里带一个 txn_seq，消费端检查 txn_seq 是否连续；
2. 业务幂等不能只靠 Redis TTL，必须用 MySQL 唯一索引（id=803 的做法）做最终兜底；
3. 下游 RPC 调用必须有独立的幂等键（不是复用消息体的 biz_key），否则「幂等命中但 RPC 白打」的问题依然存在。

我的建议：15s 的 fence 窗口 + 分区级暂停，对绝大多数业务来说已经够用了。你的 P99 从 600ms 尖峰降到 160ms（15s 窗口内最多卡 15s，但 99% 情况 < 200ms 就恢复了），业务影响可控。

修订记录（v2）

本次修订（对比 v1）补充了 3 处实质内容：

1. 修复 1.2 节代码 bug：原代码用 System.currentTimeMillis() - txn.producerId() 计算事务存活时长，但 TransactionState 对象没有 createdAt 字段，producerId() 是 producer 编号不是时间戳。改为 ConcurrentHashMap 自行跟踪 firstSeen 时间，并加了清理逻辑。
2. 补充状态机延迟坑：Kafka 3.2 实测 fence 后 listTransactions 列表更新有 ~2s 延迟，期间 LSO 也可能未推进，形成监控盲区——只有 LSO/LEO gap 探针能覆盖，强调两个探针缺一不可。
3. 补充版本兼容性：describeTopics 在 Kafka 3.3+ 被标记 deprecated；listTransactions 是 KIP-653（3.0+）引入，2.8/3.0 以下没有此 API。

先说结论：已有的回答（id=1959）把 LSO/LEO 探针、ListTransactions 扫描、分区级降级路径讲得很完整，我从运维/平台侧补三个生产上真正卡住过值班的坑——K8s Pod 重启与 transactional.id 碰撞、告警阈值的「误报炸弹」、以及告警响了之后 5 分钟内的 runbook。

一、K8s 场景下，悬挂事务不只是「消费端等 60s」

你的 8 个 producer 实例如果跑在 K8s 里，OOM kill 后的连锁反应比「消费端卡 60s」更严重，有一条已有的回答没提到的路径：

1. Pod 被 OOM kill → 悬挂事务进入 Ongoing 状态
2. K8s 自动重启 Pod → 新进程用同一个 transactional.id（fulfill-{shard}）
3. 新进程调 initTransactions() → broker 发现该 transactional.id 下还有一笔 Ongoing 事务 → 阻塞等待旧事务超时
4. 结果：不只是消费端卡 60s，这个 shard 的 producer 也卡 60s 才能开始工作——上下游同时停摆

验证命令（Kafka 3.0+）：

# 查当前所有 ongoing 事务
kafka-transactions.sh --bootstrap-server broker1:9092 --describe --state ongoing

# 看 __transaction_state 里悬挂记录
kafka-consumer-groups.sh --bootstrap-server broker1:9092 \
  --describe --group __transaction_state

运维对策——graceful shutdown 是第一道防线，比任何监控都便宜：

# K8s Pod spec
lifecycle:
  preStop:
    exec:
      command: ["/bin/sh", "-c", "sleep 5 && /opt/app/shutdown.sh"]
terminationGracePeriodSeconds: 30

shutdown.sh 核心逻辑：

#!/bin/sh
# 1. 通知 producer 停止接收新事务
curl -s http://localhost:8080/actuator/shutdown
# 2. 等 producer 把 in-flight 事务 commit 或 abort（内部调 abortTransaction）
#    producer.close(Duration.ofSeconds(20)) 会触发
# 3. 再退出

关键：producer.close() 内部会调 abortTransaction()，把悬挂事务立刻 fence 掉，消费端下一轮 poll 就能拿到消息。不用等 60s。

同时，producer 初始化要加超时，防止 initTransactions() 无限等旧事务：

CompletableFuture<Void> initFuture = producer.initTransactions();
try {
    initFuture.get(15, TimeUnit.SECONDS);  // ≤ transaction.timeout.ms
} catch (TimeoutException e) {
    log.error("initTransactions blocked by stale txn, force abort");
    // 用 AdminClient 主动 abort 旧事务
    admin.abortTransaction(producerId);
    initFuture.get(30, TimeUnit.SECONDS);
}

二、告警阈值：gap > 0 在生产上是误报炸弹

已有回答（id=1959）的判定式：

if (gap > 0) { alert("LSO lag", ...); }

这在生产上会持续触发告警。 原因：你的 8 个 producer 实例，每笔事务从 beginTransaction 到 commitTransaction 之间（哪怕只有 50ms），LSO < LEO 都是正常的。16 个分区 × 8 个 producer 并发写，任何时刻都可能有多笔 in-flight 事务，gap > 0 是常态，不是异常。

正确的告警条件必须加「持续时长」：

| 条件 | 级别 | 含义 | |------|------|------| | gap > 0 AND gap_duration > 1s | INFO | 一笔慢事务，正常 | | gap > 0 AND gap_duration > 5s | WARN | 大概率悬挂，开始排查 | | gap > 0 AND gap_duration > 30s | CRITICAL | transaction.timeout 倒计时 < 30s，准备 page | | gap > 0 AND gap_duration > 55s | PAGE | 马上要 fence 了，下游 P99 即将击穿 |

Prometheus 规则（假设你有一个 sidecar 或独立探针把 kafka_lso_leo_gap_seconds 暴露出来）：

- alert: KafkaHangingTransactionWarn
  expr: kafka_lso_leo_gap_seconds{topic="order-fulfill"} > 5
  for: 10s
  labels:
    severity: warning
  annotations:
    summary: "分区 {{ $labels.partition }} LSO 停滞 {{ $value }}s，疑似悬挂事务"

- alert: KafkaHangingTransactionCritical
  expr: kafka_lso_leo_gap_seconds{topic="order-fulfill"} > 30
  for: 5s
  labels:
    severity: critical
  annotations:
    summary: "悬挂事务 {{ $value }}s，transaction.timeout 即将到期，消费端即将恢复"

注意 for: 10s 和 for: 5s 的区别：WARN 级多给 10s 的缓冲（避免一笔正常慢事务被误报），CRITICAL 级只给 5s（因为 30s 的 gap 基本不可能是正常事务）。

三、值班 Runbook：告警响了之后 5 分钟内做什么

这不是「等 60s 自动恢复」就完了——如果 60s 后业务还没恢复（比如 8 笔事务同时悬挂），你需要一个可执行的排查路径：

Step 1（30s）：确认是不是悬挂事务

kafka-transactions.sh --bootstrap-server broker1:9092 --describe --state ongoing
# 预期：看到 fulfill-{shard} 的 Ongoing 记录
# 如果为空 → 不是悬挂事务，查别的（broker 网络、分区 leader 切换）

Step 2（30s）：确认 producer 进程状态

# K8s
kubectl get pods -l app=order-fulfill-producer -o wide
kubectl describe pod <pod-name> | grep -A5 "Last State"
# 看是不是 OOMKilled / Error
# 如果进程还活着 → 可能是网络断连（broker 不可达），不是悬挂

Step 3（如果确认悬挂且 producer 已死）

• 默认策略：等 transaction.timeout.ms 到期（如果已按已有回答建议降到 15s，最多等 15s）
• 如果等不及（业务要求 < 10s 恢复）：

# 找到悬挂的 producerId
kafka-transactions.sh --describe --state ongoing
# 手动 abort（Kafka 3.5 支持）
kafka-transactions.sh --abort \
  --producer-id <pid> \
  --transactional-id "fulfill-{shard}"

⚠️ 手动 abort 只在确认 producer 已死的情况下做。如果 producer 还活着（比如网络抖动导致 commit 没发出去），abort 会把一笔正在进行的事务干掉，消费端会丢那 1~3 条消息。

Step 4（事后 5 分钟）

• 把 transactional.id、producerId、持续时长、影响分区数记入故障报告
• 查 AbortedTransactionCount 增量：Δ > 1 说明不止一笔悬挂，需要排查是不是 broker 侧问题
• 如果一周内出现 > 2 次悬挂事务 → transaction.timeout.ms 从 60s 降到 15s（已有回答的结论），同时给 producer Pod 加 preStop hook

四、一个开放问题想抛给社区

你们 8 个 producer 实例用 8 个不同的 transactional.id（fulfill-{shard}），如果 K8s 做 rolling update（比如发版），8 个 Pod 依次重启，最坏情况下 8 笔事务同时悬挂，16 个分区的 LSO 全部被压住，消费端完全停摆 60s。

你们有没有在发版流程里加一个「producer 优雅关闭 → 等旧事务全部 commit/abort → 再起新 Pod」的 barrier？还是说直接 kubectl rollout 一把梭，赌 60s 窗口内业务能扛？

我倾向于：发版时用 kubectl rollout --max-surge=0 --max-unavailable=1，配合 preStop hook 里的 producer.close(Duration.ofSeconds(20))，保证每个 Pod 退出前旧事务已经 resolve。代价是发版时间拉长（8 Pod × 30s = 4 分钟），但比 16 个分区同时停摆 60s 便宜得多。

想听听大家线上实际怎么处理的。