kafka-connector-review

Kafka Connect 配置审查

检查 Kafka Connect 连接器配置中常见的错误配置。连接器以 JSON/YAML 定义，完全与语言无关。

目标环境：$ARGUMENTS

工作流

复制此检查清单并跟踪进度：

Connector Review Progress:
- [ ] Step 1: List all connectors
- [ ] Step 2: Inspect each connector's configuration
- [ ] Step 3: Validate configurations against plugin schemas
- [ ] Step 4: Audit for common misconfigurations
- [ ] Step 5: Generate report

1. 列出所有连接器及其状态和任务状态
2. 详细检查每个连接器的配置
3. 根据插件架构验证配置
4. 审查常见的错误配置
5. 报告检查结果，并列出当前配置和建议配置

步骤 1：列出所有连接器

使用 Lenses MCP list_kafka_connectors 工具获取所有连接器，并列出：

• 连接器名称、类和类型（source/sink）
• 状态（RUNNING、PAUSED、FAILED、UNASSIGNED）
• 任务状态和数量
• 集群信息

立即标记：

• 严重：处于 FAILED 状态的连接器
• 严重：处于 FAILED 状态的任务
• 警告：处于 PAUSED 状态且没有明显原因的连接器

预期输出：列出所有连接器的名称、类、状态和任务状态。

验证：如果没有返回连接器，则报告未配置 Connect 集群并停止。

步骤 2：检查配置

使用 Lenses MCP get_kafka_connector_target_definition 工具获取每个连接器的完整配置 YAML。

步骤 3：验证配置

使用 Lenses MCP validate_connector_configuration 工具，根据每个连接器插件的架构验证其配置。此操作可检测：

• 缺少必需的配置字段
• 无效的配置值
• 类型不匹配

步骤 4：审查常见的错误配置

错误处理

• 严重：设置了 errors.tolerance=all，但未设置 errors.deadletterqueue.topic.name（会静默丢弃消息）
• 警告：未设置 errors.tolerance（默认为 none，连接器遇到任何错误都会停止）
• 警告：未设置 errors.log.enable=true（不会记录错误）
• 建议：添加 errors.deadletterqueue.context.headers.enable=true，以获取更丰富的 DLQ 元数据

转换器

• 警告：key.converter 或 value.converter 与主题的序列化格式不匹配
• 警告：结构化数据使用了 org.apache.kafka.connect.storage.StringConverter（应使用 Avro/JSON/Protobuf 转换器）
• 警告：预期使用架构时未设置 schemas.enable

转换（SMT）

• 警告：转换链过于复杂（超过 3 个 SMT），可能难以调试
• 建议：对于复杂转换，可考虑使用 Kafka Streams 或 ksqlDB
• 验证转换类名称是否有效，顺序是否正确

任务数量

• 警告：高吞吐量连接器设置了 tasks.max=1（会限制并行度）
• 建议：Source 连接器的 tasks.max 应与源分区情况相匹配
• 建议：Sink 连接器的 tasks.max 应与主题分区数量相匹配

连接和重试

• 警告：与外部系统交互的连接器未配置重试
• 警告：未设置连接超时
• 建议：设置明确的 consumer.override.* 或 producer.override.*，以进行性能调优

命名约定

• 建议：连接器名称应遵循一致的模式（例如，{source/sink}-{system}-{entity}）

成功标准

定量标准

• 覆盖 90% 的连接器相关查询（使用 10–20 种不同表述进行测试）
• 在少于 10 次 MCP 工具调用内完成审查
• 捕获插件架构报告的 100% 验证错误

定性标准

• 在状态概览中立即标记失败的连接器
• 始终识别错误处理方面的缺陷（缺少 DLQ、静默丢弃）
• 报告对连接器运维人员和开发人员都有帮助

示例

示例 1：例行连接器审查

用户说：“审查 staging 中的所有连接器”

操作：

1. 列出所有连接器及其状态
2. 检查并验证每个连接器的配置
3. 审计常见的配置错误 结果：提供涵盖所有连接器的完整报告，并按优先级列出发现的问题

示例 2：调查连接器故障

用户说：“我的 sink 连接器出了故障，能帮忙查查原因吗？”

操作：

1. 列出连接器，并找出状态为 FAILED 的连接器
2. 获取故障连接器的完整配置
3. 根据插件架构验证配置
4. 检查常见问题（DLQ、转换器、错误处理） 结果：诊断具体故障并提供修复步骤

示例 3：深入检查单个连接器

用户说：“检查 elasticsearch-sink 连接器的配置”

操作：

1. 获取指定连接器的目标定义
2. 根据其插件架构进行验证
3. 审计错误处理、转换器和任务数 结果：提供针对单个连接器的专项报告

故障排除

未返回连接器

原因：环境中未配置 Kafka Connect 集群，或未部署任何连接器。 解决方案：通过 Lenses UI 确认 Connect 集群是否存在。检查 get_deployment_targets，查看可用的 Connect 集群。

验证因未知插件而失败

原因：Connect 集群上未安装该连接器插件类。 解决方案：报告缺失的插件。这属于部署问题，而非配置问题。

连接器显示 RUNNING，但任务处于 FAILED 状态

原因：连接器框架正在运行，但各个任务遇到了错误。 解决方案：检查每个任务的状态。常见原因包括身份验证失败、网络问题，或与外部系统之间的架构不匹配。

输出格式

## Connector Review Report

### Environment: {name}

### Connector Status Overview
| Connector | Class | Status | Tasks | Failed Tasks |
|-----------|-------|--------|-------|-------------|
| name | class | RUNNING | 3/3 | 0 |

### Critical (must fix)
- [connector-name] Description of the issue
  Current: {current config} | Recommended: {recommended config}

### Warning (should fix)
- [connector-name] Description of the issue
  Current: {current config} | Recommended: {recommended config}

### Validation Errors
- [connector-name] {field}: {validation error message}

### Suggestion (consider improving)
- [connector-name] Description of the suggestion
  Recommendation: {how to improve}

### Summary
- X connectors reviewed
- Y critical issues found
- Z warnings found
- Failed connectors: N
- Failed tasks: M