Kafka Connect / MirrorMaker 2 REST API 查询指南

1. 说明

核心 REST 资源定义在以下类中：

• RootResource
• ConnectorsResource
• ConnectorPluginsResource
• LoggingResource
• InternalClusterResource

完整 OpenAPI 定义见：

• docs/generated/connect_rest.yaml

通用约定：

• 默认 REST 端口为 8083，实际以 listeners 配置为准。
• 接口通常使用 Content-Type: application/json。
• {connector} 表示 connector 名称。
• {task} 表示整数类型的 task ID。
• ?forward= 是 Connect worker 内部转发参数，普通用户通常不需要指定。
• 配置了 rest.extension.classes 时，第三方扩展可能注册额外接口，本清单只包含 Kafka Connect 内建接口。

2. Worker 基础接口

3. Connector Plugin 接口

注意：GET /connector-plugins 查询的是处理该请求的 worker，不是整个集群的统一视图。滚动升级或各 worker 插件不一致时，结果可能不同。

4. Connector 管理接口

4.1 创建 connector

POST /connectors
Content-Type: application/json

{
  "name": "flink-to-mm2-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "tasks.max": "4"
  },
  "initial_state": "STOPPED"
}

initial_state 可选值：

• RUNNING：默认值
• PAUSED
• STOPPED

4.2 查看 connector 列表

GET /connectors

返回 connector 名称数组。

GET /connectors?expand=status&expand=info

返回带状态和详情的 Map，适合一次查询所有 connector 的运行状态。

5. Connector 生命周期接口

5.1 Restart 参数

行为：

• 默认 includeTasks=false&onlyFailed=false：只重启 connector，返回 204。
• 指定 includeTasks=true 或 onlyFailed=true：返回 202 和 connector 状态信息。
• 单个 task 重启使用 /connectors/{connector}/tasks/{task}/restart。

5.2 STOPPED 与 PAUSED 的区别

修改 connector offset 前必须是 STOPPED，PAUSED 不够。

PUT /connectors/{connector}/stop 返回后，状态变化和 task 清理仍可能有短暂延迟。建议通过以下接口确认 connector 和所有 task 均已停止：

GET /connectors/{connector}/status

6. Task 管理接口

7. Offset 管理接口

7.1 修改 offset

PATCH /connectors/flink-to-mm2-source/offsets
Content-Type: application/json

{
  "offsets": [
    {
      "partition": {
        "cluster": "source",
        "topic": "orders",
        "partition": 0
      },
      "offset": {
        "offset": 999
      }
    },
    {
      "partition": {
        "cluster": "source",
        "topic": "orders",
        "partition": 1
      },
      "offset": null
    }
  ]
}

其中 offset: null 表示单独重置该分区。

7.2 状态约束

错误响应示例：

{
  "error_code": 400,
  "message": "Connectors must be in the STOPPED state before their offsets can be modified. This can be done for the specified connector by issuing a 'PUT' request to the '/connectors/flink-to-mm2-source/stop' endpoint"
}

8. Topic Tracking 接口

9. 日志级别管理接口

这些接口属于 admin REST API：

• 默认情况下使用普通 listeners。
• 配置了 admin.listeners 后，只暴露在 admin listener 上。
• 当前版本使用 Log4j 1.x 风格级别。

9.1 设置日志级别

PUT /admin/loggers/org.apache.kafka.connect?scope=worker
Content-Type: application/json

{
  "level": "DEBUG"
}

scope 可选值：

10. Distributed 模式内部接口

以下接口是 Connect worker 之间通信使用的，不是给用户调用的，也没有列入正常 OpenAPI 文档：

特点：

• 内部请求需要签名。
• 不保证稳定的用户调用契约。
• 可能因为 non-leader、签名错误、rebalance 返回 400/401/409 等错误。
• 运维和迁移脚本不应调用这两个接口。

11. Distributed 模式请求路由行为

11.1 Leader 转发

创建、修改配置、删除、停止、重启、修改 offset 等操作通常需要 leader 执行。请求发到 follower 时，follower 会内部转发到 leader。

11.2 Rebalance 冲突

如果集群正在 rebalance、配置快照落后或暂时找不到 leader，可能返回：

409 Conflict

11.3 GET 查询

一般可以访问任意 worker，但部分数据具有 worker 局部性或最终一致性：

11.4 forward 参数

API 中的 ?forward= 是内部转发参数，正常用户不需要传。接收到请求的 worker 会根据 leader 信息自动转发。

11.5 REST Extension

如果在 distributed worker 中配置了 rest.extension.classes，第三方扩展可以额外注册资源、filter 或 endpoint，因此实际对外接口可能比内建列表更多。

12. 运维场景速查

13. 参考资料

• connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/ConnectRestServer.java
• connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/RootResource.java
• connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
• connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
• connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/LoggingResource.java
• connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/InternalClusterResource.java
• docs/generated/connect_rest.yaml
• docs/kafka-connect/user-guide.md