> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 管理 Connector

> 了解如何创建、配置和管理 AutoMQ 托管 Kafka Connect 服务中的 Connector 任务。

在 Kafka Connect 框架中，Connector 定义并管理了一组用于执行特定数据集成任务（ETL）的 Task。在 AutoMQ 托管 Kafka Connect 服务中，每一个 Connector 底层运行的多个 Task 都被调度在独立的 Worker 集群中，确保了任务间的资源隔离和稳定运行。

## 前置条件

在创建和部署 Connector 之前，您需要确保已经准备好以下依赖资源：

* 一个正常运行的 AutoMQ 实例及用于数据访问的 ACL 用户凭证。
* 用于数据集成源头或目标的 Kafka Topic 和 Consumer Group。
* 一个配置正确的 Amazon EKS（Elastic Kubernetes Service）集群，并在 EKS 集群内为 Connector 运行准备的 Service Account 和 Namespace。
* 一个具备特定权限的 AWS IAM Role，用于授权 Connector 访问云资源，该 IAM Role 和 Service Account 必须绑定。

### 步骤 1：创建 AutoMQ 实例

数据集成任务需要一个 AutoMQ 实例作为元数据服务以及数据集成的 Sink、Source。如果您尚未创建实例，请参考以下文档完成创建。

* **相关文档**：[《管理实例》](../managing-automq/manage-instances)

### 步骤 2：创建 ACL 用户凭证，完成 ACL 授权

Connector 需要使用 Kafka 客户端协议访问 AutoMQ 集群，因此必须为其提供一套有效的 ACL 用户凭证。

* **相关文档**：[《管理 Kafka ACL》](../manage-security/manage-kafka-acls)

<Tip>
  **重要约束：**
  在创建 ACL 用户时，您必须确保为其授予了 Connector 预期要访问的所有 Topic 的读/写权限。如果您的 Connector 是一个 Sink Connector（数据导出），还必须为其授予相关 Consumer Group 的读取权限。否则，Connector 任务将因权限不足而无法正常启动。
</Tip>

### 步骤 3：准备 EKS 环境及 IAM 角色

AutoMQ 托管 Kafka Connect 的 Worker 运行在您自己的 EKS 集群中，以保证数据的安全性和网络隔离性。同时，它需要通过一个绑定的 IAM Role 来获取访问其他云资源（如 S3）的权限。

* **相关文档**：[《部署 AutoMQ on EKS》](../deploy-automq-on-kubernetes/deploy-to-aws-eks)

在准备 EKS 环境时，请特别注意以下几点：

* **服务账号与 IAM 角色的绑定**：您需要创建一个 Kubernetes Service Account，并将其与一个 AWS IAM Role 进行绑定。我们推荐使用 IRSA (IAM Roles for Service Accounts) 方式实现，**详情请参考 AWS 官方文档：[适用于服务账户的 IAM 角色](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html)**。
* **IAM Role 权限策略**：用户可以为 IAM Role 添加一个访问 AutoMQ 运维存储桶的读写权限。这样的话，系统可以完成日志的上传和监控。

以下是一个 IAM Policy 的示例（请将 `YOUR_OPS_BUCKET_NAME` 替换为实际的存储桶名称）：

```json theme={null}
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
                "s3:DeleteObject",
                "s3:ListBucket",
                "s3:GetObjectVersion",
                "s3:DeleteObjectVersion",
                "s3:ListBucketVersions"
            ],
            "Resource": [
                "arn:aws:s3:::YOUR_OPS_BUCKET_NAME/*",
                "arn:aws:s3:::YOUR_OPS_BUCKET_NAME"
            ]
        }
    ]
}
```

<Tip>
  **重要提醒：**
  如果当前 Connector 需要访问除 AutoMQ 存储桶以外的其他云资源（如 RDS、DynamoDB 等），请确保在 IAM Role 的权限策略中添加相应的访问权限。
</Tip>

* **K8s 授权**: 您需要对当前的 Service Account 授予一定的权限。

```bash theme={null}
kubectl create clusterrole node-reader --verb=get--resource=nodes
kubectl create clusterrolebinding default-node-reader --clusterrole=node-reader --serviceaccount=<YOUR_NAMESPACE>:<YOUR_SERVICE_ACCOUNT_NAME> # 例如: default:default
```

完成以上所有云资源和相关配置准备后，即可创建 Kafka Connector。

## 创建 Connector

创建 Connector 的流程主要分为部署信息、插件参数、Metrics 集成等步骤。

### 步骤 1：填写部署配置

1. 登录 AutoMQ 控制台，在左侧导航栏中选择 **Kafka Connect**。
2. 点击 **Connector** 页签，然后点击 **创建 Connector** 按钮。
3. 在创建页面，您需要填写以下基础参数：
   * **选择插件**：从下拉列表中选择一个您希望使用的插件及其版本。此为必选项。
   * **选择 Kafka 集群**：在此处配置 Connector 访问 AutoMQ 集群所需的信息。
     * **AutoMQ 实例**：选择您已准备好的 AutoMQ 实例。
     * **ACL 用户名/密码**：填写用于数据访问的 ACL 用户凭证。
     * **Topic / Consumer Group**：根据插件是 Source 还是 Sink 类型，填写需要读写的 Topic 或消费组。
   * **选择 EKS 集群**：配置 Connector 运行的底层计算资源。
     * **EKS 集群**：选择您已准备好的 EKS 集群。
     * **Namespace**：填写为 Connector 准备的 Kubernetes 命名空间。
     * **Service Account**：填写已绑定了 IAM Role 的服务账号。
     * **ScheduleSpec**：定义将 Connector 任务调度到特定 Kubernetes 节点的策略。此字段允许您利用 Kubernetes 原生的节点调度功能来优化工作负载的放置，以满足性能、成本或高可用性等要求。AutoMQ 支持以下标准的 Kubernetes 调度策略：
       * **`nodeSelector`**：通过标签选择器（label selector）将 Pod 直接调度到具有特定标签的节点上。
       * **`affinity` 和 `anti-affinity`**：配置节点亲和性（`nodeAffinity`）与反亲和性（`nodeAntiAffinity`），以及工作负载间亲和性（`podAffinity`）与反亲和性（`podAntiAffinity`），实现更灵活的调度决策。
       * **`tolerations`**：允许（但不强制）Pod 调度到具有匹配污点（`taints`）的节点上，通常用于将工作负载放置在专用或特殊配置的节点上。
   * **填写容量配置**：根据您的数据负载，为 Connector 分配资源。
     * **最大任务数**：指定 Connector 可并行运行的最大 Task 数量。
     * **Worker 数量**：指定运行这些 Task 的 Worker 进程数量。
     * **Worker 规格**：为每个 Worker 选择合适的 CPU 和内存规格。您可以根据插件的资源消耗情况选择不同大小的规格。

<Tip>
  你需要为 Connector 运行提供必要的 Kubernetes 资源配额（CPU、内存等）。请确保 EKS 集群中的节点池有足够的资源来满足 Connector 的运行需求。
</Tip>

### 步骤 2：填写插件配置

完成基础配置后，点击 **下一步**，进入插件参数配置页面。AutoMQ 支持两种模式来填写插件所需的自定义配置：

* **表单模式**：AutoMQ 会自动解析插件的配置定义（Schema），并将其渲染成一个结构化的表单。您可以根据每个参数的描述和提示直接填写，操作直观便捷。
* **自定义模式**：如果您更熟悉插件的参数，或者需要填写表单模式未覆盖的高级参数，可以选择此模式。在该模式下，您可以完全自由地以键值对（Key-Value）的形式指定所有配置。

### 步骤 3：配置 Metrics 导出（可选）

为了实现自定义监控和报警，AutoMQ 支持将 Connector 运行时的 Metrics 数据导出到您自有的 Prometheus 系统中。

在这一步，您可以选择配置外部 Prometheus 的 `remote_write` 地址和访问凭证（如 Basic Auth）。配置完成后，所有相关的 Metrics 数据将被实时转发。

### 步骤 4：预览并创建

完成所有配置后，您可以在预览页面检查所有参数。确认无误后，点击 **创建** 按钮，AutoMQ 将开始部署并启动您的 Connector 任务。

## 管理 Connector

Connector 创建后，您可以在其详情页进行全方位的管理和运维操作。

### 修改配置

您可以随时进入 Connector 详情页，对其配置进行修改。

* **修改容量配置**：根据业务负载的变化，您可以动态调整 **最大任务数**、**Worker 数量** 以及 **Worker 规格**，以实现资源的弹性伸缩，满足不同场景下的性能需求。
* **修改运行参数**：您可以修改插件的自定义配置，以调整其运行行为。

### 管理任务生命周期

在 Connector 详情页，您可以对底层的 Task 执行生命周期管理操作，方便进行调试或维护。

* **暂停（Pause）**：临时停止所有 Task 的数据处理活动。
* **恢复（Resume）**：从暂停状态恢复所有 Task 的运行。
* **重启（Restart）**：批量重启所有 Task，通常用于加载新的配置或从错误状态中恢复。

## 删除 Connector

当您不再需要运行某个 Connector 时，可以随时将其删除。

**操作步骤：**

1. 在 **Connector** 列表页面，选择目标 Connector。
2. 点击 **删除** 按钮。
3. 在二次确认对话框中确认操作后，Connector 将被永久删除。

<Danger>
  **注意**：删除操作不可撤销。删除 Connector 将会一并移除其下所有正在运行的 Task，以及这些 Task 相关的消费位点和进度信息。
</Danger>

## 生产环境监控报警

对于生产环境运行的 Connector，完善的监控和报警是保障其稳定性的关键。AutoMQ 提供了 Metrics 和日志两大维度的可观测性能力。

* **Metrics 指标集成**：您可以在创建或编辑 Connector 时，通过配置 Metrics 导出，将 Connector 运行的核心指标（如吞吐、延迟、错误率等）对接到您自己的 Prometheus 监控系统中，并基于这些数据配置告警规则。
* **日志查询**：当需要排查问题时，您可以随时前往 Connector 的详情页面，查询其运行日志，快速定位问题根源。