mcpField
K
kafka-consumer-lag

kafka-consumer-lag

github.com/lensesio数据与数据库6660 次安装10 周前更新
代码仓库

关于 kafka-consumer-lag

kafka-consumer-lag 是用于分析 Kafka 消费者组延迟的技能，面向处理消费积压和集群事件的工程师。它按组检查状态与分区延迟，并结合分区和吞吐数据诊断原因、提出处理建议。

主要功能

  1. 01获取消费者组状态与延迟
  2. 02识别高延迟、空闲或再平衡组
  3. 03结合分区和吞吐数据诊断原因
  4. 04按组提供处理建议
  5. 05抽样验证主题数据流

适用场景

  1. 01调查消费者积压不断增长
  2. 02检查指定主题的消费者延迟
  3. 03查找带有已提交偏移量的失效组

MCP 相关

通过 Lenses MCP 查询消费者组、主题分区和消息吞吐，并可用 SQL 抽样检查近期消息，以辅助诊断延迟。

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/lensesio/agentic-engineering-for-apache-kafka --skill kafka-consumer-lag

安装后在 Claude Code 中输入/kafka-consumer-lag确认已加载。

附带文件references/test-cases.md

详细介绍

Kafka 消费者延迟分析

分析所有组的消费者组延迟并诊断潜在原因。消费者延迟是最常监控的 Kafka 指标，也是工程师在事故期间首先检查的指标。

目标环境：$ARGUMENTS

工作流

复制此清单并跟踪进度：

Lag Analysis Progress:
- [ ] Step 1: Fetch all consumer groups
- [ ] Step 2: Identify problematic groups
- [ ] Step 3: Diagnose root causes
- [ ] Step 4: Generate report
  1. 获取所有消费者组及其状态和延迟信息
  2. 识别有问题的组（高延迟、空闲、正在重新平衡）
  3. 使用分区和吞吐量数据诊断根本原因
  4. 报告发现并提供各组的补救步骤

步骤 1：获取消费者组

使用 Lenses MCP list_consumer_groups 工具获取所有消费者组，包括：

  • 组状态（Stable、Rebalancing、Empty、Dead）
  • 各主题分区的延迟
  • 活跃成员数量
  • 协调器信息

若要分析特定主题，请使用 list_consumer_groups_by_topic 缩小范围。

预期输出：包含状态、延迟和成员数量的所有消费者组列表。

验证：如果未返回消费者组，请报告此发现并停止——集群可能没有活跃的消费者。

步骤 2：识别有问题的组

将消费者组标记为以下类别：

严重
  • 具有已提交偏移量但处于 Dead 或 Empty 状态的组（消费者已停止）
  • 延迟随时间增长的组（使用 get_topic_partitions 将当前延迟与分区末端偏移量进行比较）
警告
  • 处于 Rebalancing 状态的组（消费已暂停）
  • 分区分配不均的组（部分消费者分配到的分区明显多于其他消费者）
  • 没有已提交偏移量的组（消费者可能从未成功启动）
建议
  • 成员数量相对于分区数量过多的组（存在空闲消费者）
  • 从近期没有消息的主题消费的组（使用 get_dataset_message_metrics 检查生产者吞吐量）

步骤 3：诊断根本原因

针对每个有问题的组，判断可能的原因：

延迟增长 + 生产者速率稳定
  • 消费者吞吐量瓶颈
  • 补救措施：扩容消费者、提高 max.poll.records、缩短每条消息的处理时间，并检查是否存在同步阻塞调用
延迟激增 + 正在重新平衡
  • 重新平衡导致消费暂停
  • 补救措施：切换到 cooperative sticky assignor、提高 session.timeout.ms，降低触发 max.poll.interval.ms 风险的可能性
单个分区有延迟（其他分区正常）
  • 热分区／键倾斜
  • 补救措施：检查分区策略、排查热点键，并考虑使用自定义分区器
所有分区的延迟都相同
  • 消费者整体处理速度较慢
  • 补救措施：分析消费者处理过程、检查外部依赖（数据库、API 调用），并增加消费者实例
零延迟但组为空
  • 消费者崩溃或已停用
  • 补救措施：检查消费者进程运行状况，必要时重启，并清理过期组

步骤 4：验证数据流

使用 Lenses MCP execute_sql 工具，从延迟较高的主题中抽取近期消息样本：

SELECT * FROM {topic} LIMIT 5

这可以确认主题有活跃的生产者且消息正在流动。

成功标准

定量
  • 覆盖 90% 的延迟相关查询（使用 10–20 种不同表述进行测试）
  • 在 12 次 MCP 工具调用以内完成分析
  • 每次运行的 MCP 调用失败次数为 0
定性
  • 每个被标记的组都有具体诊断和补救措施
  • 报告明确区分症状和根本原因
  • 在同一环境中重新运行时结果一致

示例

示例 1：事故调查

用户说：“消费者处理落后了，检查一下延迟。”

操作：

  1. 获取该环境中的所有消费者组
  2. 按总延迟从高到低排序，找出问题最严重的组
  3. 诊断每个严重问题组的根本原因 结果：提供按优先级排列的报告，包含各组的诊断和补救措施
示例 2：检查特定主题的延迟

用户说：“检查 orders.payment.completed 的消费者延迟”

操作：

  1. 使用 list_consumer_groups_by_topic 获取从该主题消费的组
  2. 检查分区级别的延迟分布
  3. 使用 execute_sql 验证数据流 结果：针对单个主题的消费者组生成的重点报告
示例 3：清理过期组

用户说：“查找所有失效或空的消费者组”

操作：

  1. 获取所有消费者组
  2. 筛选状态为 Dead 或 Empty 且具有已提交偏移量的组
  3. 报告可安全清理的过期组 结果：可删除的候选组列表

故障排除

消费者组列表为空

原因：环境中不存在消费者组，或权限受到限制。 解决方案：确认消费者正在运行。检查 Lenses Agent 权限是否允许列出消费者组。

延迟数字似乎不正确

原因：如果消费者最近停止，消费者组偏移量可能已过期。 解决方案：将延迟与 get_topic_partitions 返回的分区末端偏移量进行比较。注意偏移量上次提交的时间。

execute_sql 未返回数据

原因：主题为空，或查询时间窗口之外存在非常旧的消息。 解决方案：使用 get_dataset_message_metrics 检查主题近期是否有吞吐量。主题为空但消费者存在延迟，表明生产者可能存在问题。

输出格式

## Consumer Lag Report

### Environment: {name}

### Critical (immediate action)
| Consumer Group | Topic | Total Lag | State | Diagnosis | Remediation |
|---------------|-------|-----------|-------|-----------|-------------|
| group-name | topic | 50000 | Stable | Throughput bottleneck | Scale consumers |

### Warning (investigate)
| Consumer Group | Topic | Total Lag | State | Diagnosis | Remediation |
|---------------|-------|-----------|-------|-----------|-------------|

### Suggestion (optimise)
| Consumer Group | Topic | Total Lag | State | Diagnosis | Remediation |
|---------------|-------|-----------|-------|-----------|-------------|

### Summary
- X consumer groups analysed
- Y groups with critical lag
- Z groups with warnings
- Stale/empty groups: N

评论

支持 Markdown