> ## 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.

# Adobe Experience Platform Sink Connector

> 介绍如何在 AutoMQ Connect 中配置和运行 Adobe Experience Platform Sink Connector，包括前置条件、配置、监控和故障排查。

## 概述

Adobe Experience Platform Sink Connector 消费 Kafka Topic 中的 JSON 消息，经 HTTP 流式采集入口批量发送到 Adobe Experience Platform（AEP）。它位于事件生产端与 AEP 数据流之间，适用于将客户活动、应用事件等数据接入已配置的 XDM Schema 和 Dataset，供后续分析或客户画像处理使用。

每条 Kafka 消息值对应一个 AEP 消息，Connector 将这些消息组成 `messages` 数组发送。目标资源和字段映射取决于 AEP 数据流配置及消息内容。Connector 不自动创建 Dataset、生成映射或补充 AEP 消息头，也不会把 Kafka 消息键、分区、Offset 和 Headers 自动写入业务数据。

## 前置条件

* 在有权限的 AEP 项目、组织和 Sandbox 中准备目标 XDM Schema、Dataset，以及 HTTP API Source 的基础连接、源连接、目标连接和 Dataflow；取得基础连接返回的 `inletUrl` 及对应的 `flowId`。RAW 数据需要转换时，应在 AEP 配置 Data Prep Mapping 并关联到 Dataflow；已符合目标 Schema 的 XDM 数据可使用不带转换的路径。
* 确认 AEP 基础连接的 `authenticationRequired` 策略。需要认证的入口必须具备有效且获授权的 Adobe 身份集成；涉及个人身份信息的数据应使用认证入口。资源管理 API 的 Bearer Token、API Key、IMS 组织和 Sandbox 请求头用于 AEP 资源管理，不应直接作为一套 Connector 配置照搬。
* 上游消息应具有与所选入口和数据流匹配的 AEP 消息结构及字段类型。创建 Dataflow 后等待约五分钟再发送数据，并考虑创建或更新数据流时的短暂采集暂停。

## 授权许可

使用 Apache License 2.0。

## 快速开始

提前准备 Connect Cluster、Kafka 和上述 AEP 资源，确认网络连通和访问权限；创建和管理操作参见[管理 Connector](../manage-connectors)。以下最小配置适用于已预创建、`authenticationRequired=false` 的入口，以及消息键为 UTF-8 字符串或空值、消息值为普通 JSON 对象的 Topic；不适用于必须认证的入口或包含个人身份信息的数据。

```properties theme={null}
name=adobe-experience-platform-sink
connector.class=com.adobe.platform.streaming.sink.impl.AEPSinkConnector
topics=<input-topic>
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
aep.endpoint=<aep-inlet-url>
```

将 Topic 和入口地址替换为实际资源。`aep.endpoint` 使用 AEP 返回的 HTTPS `inletUrl`，路径应为 `/collection/<connection-id>`，不要填已经包含 `/collection/batch/` 的地址；Connector 会自行转换为批量入口。无需显式填写默认的单 Task、批量阈值或关闭认证设置。`value.converter.schemas.enable=false` 使输入无需 Connect 的 `schema` / `payload` 外壳，但不能省略 AEP 所需的消息结构。

下面是上游生产者写入 Kafka 的单条消息值，展示 HTTP API Source 的 XDM 消息结构，不是 Connector 配置，也不是最终 HTTP 请求的外层 `messages` 数组。替换 Schema、Dataflow、Dataset 和业务字段后使用，`xdmEntity` 必须符合实际目标 Schema。

```json theme={null}
{
  "header": {
    "schemaRef": {
      "id": "<xdm-schema-id>",
      "contentType": "application/vnd.adobe.xed-full-notext+json; version=1.0"
    },
    "flowId": "<aep-flow-id>",
    "datasetId": "<aep-dataset-id>"
  },
  "body": {
    "xdmMeta": {
      "schemaRef": {
        "id": "<xdm-schema-id>",
        "contentType": "application/vnd.adobe.xed-full-notext+json; version=1.0"
      }
    },
    "xdmEntity": {
      "_id": "customer-event-001",
      "person": {
        "name": {
          "firstName": "Example",
          "lastName": "Customer"
        }
      }
    }
  }
}
```

这里的 `header` 是 JSON 消息头，不是 HTTP Headers；`flowId`、`datasetId`、`schemaRef` 和 `body` 均不属于 Connector 参数。Connector 不会补齐这些字段，不要提前在每条 Kafka 消息外再套一层 `messages`。其他 AEP 采集路径可能要求不同的消息元数据，例如批量消息中的 `imsOrgId`；应以实际入口要求为准，不要将不同路径的样例混用。

应用配置后，先发送少量符合 Schema 的数据，分别检查 Connect 运行状态、采集响应，以及 AEP 验证和 Dataset 入库结果。HTTP 成功响应仅表示采集层的接受情况，不能证明最终 Dataset 入库成功。

**安全警示**：Connector 和 Task 启动时会在 INFO 日志中输出原始配置，包括可能存在的 Client Secret、授权码、代理密码或自定义 Authorization 请求头。仅关闭 DEBUG 或保持 `errors.log.include.messages=false` 不能解决此风险，也不能假定 Config Provider 替换后自动脱敏。在传入生产凭证前，应先落实启动日志脱敏或抑制方案及日志访问控制；敏感数据不要启用 DEBUG，凭证和消息内容不要附入工单或共享日志。

## 配置

### 实例与输入订阅

#### `name`

Connector 实例名称。

* **类型**：`string`
* **默认值**：无
* **重要级别**：高
* **有效值 / 注意事项**：必填；非空且不含 ISO 控制字符，名称应唯一。

#### `connector.class`

选择 AEP Sink Connector。

* **类型**：`string`
* **默认值**：无
* **重要级别**：高
* **有效值 / 注意事项**：必填；使用 `com.adobe.platform.streaming.sink.impl.AEPSinkConnector`。

#### `topics`

指定消费的 Kafka Topic。

* **类型**：`list`
* **默认值**：空列表
* **重要级别**：高
* **有效值 / 注意事项**：逗号分隔的 Topic 名称；与 `topics.regex` 必须且只能选择一个非空配置，不得包含 DLQ Topic。

#### `topics.regex`

通过正则表达式订阅 Kafka Topic。

* **类型**：`string`
* **默认值**：空字符串
* **重要级别**：高
* **有效值 / 注意事项**：有效的 Java 正则表达式；与非空 `topics` 互斥，不得匹配 DLQ Topic。

#### `tasks.max`

允许创建的最大 Task 数量。

* **类型**：`int`
* **默认值**：`1`
* **重要级别**：高
* **有效值 / 注意事项**：至少为 `1`。实际并行度受输入分区分配限制；各 Task 使用同一入口，不提供跨 Task 全局顺序。

### 消息转换

#### `key.converter`

反序列化 Kafka 消息键。

* **类型**：`class`
* **默认值**：`null`
* **重要级别**：低
* **有效值 / 注意事项**：省略时使用 Worker Converter。须与上游键编码一致；快速开始针对 UTF-8 字符串键使用 `org.apache.kafka.connect.storage.StringConverter`，消息键不发送给 AEP。

#### `value.converter`

反序列化 Kafka 消息值。

* **类型**：`class`
* **默认值**：`null`
* **重要级别**：低
* **有效值 / 注意事项**：省略时使用 Worker Converter。快速开始显式使用 `org.apache.kafka.connect.json.JsonConverter`，这不是默认类；输入必须符合 AEP 消息要求，而不只是语法合法的 JSON。

#### `value.converter.schemas.enable`

控制 JsonConverter 是否使用 Connect Schema 外壳。

* **类型**：`boolean`
* **默认值**：`true`
* **重要级别**：高
* **有效值 / 注意事项**：仅针对 JsonConverter。普通 JSON 对象使用 `false`；`true` 要求 Connect 的 `schema` / `payload` 格式。此项不控制 AEP 的 XDM Schema 校验。

### 目标入口

#### `aep.endpoint`

指定 AEP 流式采集入口。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：必填，无运行时回退；不能为空。填写完整 HTTPS 入口 URL，使用原始 `/collection/` 路径；该路径会被替换为 `/collection/batch/`，已经转换过的地址会被再次改写。

#### `aep.connection.endpoint.headers`

为采集请求附加 HTTP Headers。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：值为请求头名称到字符串值的 JSON 对象文本。缺失或空字符串时运行时使用空映射；JSON 解析失败时不会添加这些请求头。此项不同于消息值中的 `header`，包含授权信息时有原始配置日志泄密风险。

### 批量发送

#### `aep.flush.interval.seconds`

控制一次消费批次内按经过时间触发发送的阈值，单位秒。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `1`，按 Java `int` 运算乘以 `1000`；乘法结果小于 `1` 时回退为 `1000` ms。非法整数文本或超出 `int` 范围的文本会抛出 `NumberFormatException`，不会回退。没有 ConfigDef 范围校验；建议使用正整数并避免乘法溢出。它不是后台定时器，消费批次结束时剩余消息仍会发送，不会跨消费批次等待凑满。

#### `aep.flush.bytes.kb`

控制一次消费批次内按累计消息大小触发发送的阈值，单位 KB。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `4`，按 Java `int` 运算乘以 `1024`；乘法结果小于 `1` 时回退为 `4096`。非法整数文本或超出 `int` 范围的文本会抛出 `NumberFormatException`，不会回退。没有 ConfigDef 范围校验；建议使用正整数并避免乘法溢出。实际累计的是 Java 字符串长度，不是 UTF-8 字节数，也不包含最终请求包装开销。加入记录后才检查阈值，不能以此确保请求满足 Adobe 的 1 MB 大小限制。

### HTTP 连接与重试

#### `aep.connection.timeout`

采集 HTTP 请求的连接超时，单位毫秒。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `5000`；使用正整数。只控制采集连接，不是 Token 请求超时设置。

#### `aep.connection.readTimeout`

采集 HTTP 请求的读取超时，单位毫秒。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `60000`；使用正整数，保留名称大小写。不用于 Token 请求。

#### `aep.connection.maxRetries`

限制单次采集 HTTP 调用的总尝试次数。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `3`，表示最多三次总尝试，不是初次请求再加三次重试。没有参数范围校验；要执行发送应使用正数，不要设置为零或负数。对 5xx 和 I/O 异常按固定间隔重新发送同一请求体；429 等其他非 2xx 不走该重试路径，响应内容解析失败也不在此范围。重发可能造成重复。

#### `aep.connection.retryBackoff`

采集 HTTP 尝试失败后的固定等待时长，单位毫秒。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `300`；使用非负整数。最后一次 5xx 或 I/O 失败后也可能等待；等待会阻塞 Task，不是认证重试或自适应限速设置。

### 身份认证

#### `aep.connection.auth.enabled`

启用 Connector 的 Bearer Token 获取与附加功能。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失时运行时使用字符串 `false`。只有精确的小写 `true` 才启用；`TRUE` 或带空白的值不会启用。此项不能改变 AEP 入口的认证策略。

#### `aep.connection.auth.token.type`

选择 Adobe Token 获取方式。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：启用认证且未配置时运行时使用 `jwt_token`；支持精确值 `access_token`、`jwt_token`、`oauth2_access_token`，空值或未知值会失败。JWT 认证路径已弃用；迁移时应按 Adobe 身份集成要求显式选择 OAuth2 并确认授权和 scope 兼容性，不能因选项存在就假定旧 JWT 仍获服务端支持。

#### `aep.connection.auth.client.id`

指定 Adobe 身份集成的 Client ID。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；三种认证方式启用时均必填。它不是 Access Token。

#### `aep.connection.auth.client.secret`

指定 Adobe 身份集成的 Client Secret。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；三种认证方式启用时均必填。此项没有 PASSWORD 类型的自动脱敏保护，传入前必须解决原始配置日志泄密风险。

#### `aep.connection.auth.endpoint`

指定 Adobe IMS 基础 URL。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；启用认证时必填，不能依靠环境变量省略。三种方式分别附加 `/ims/token/v1`、`/ims/exchange/jwt/`、`/ims/token/v3`，不要提前附加这些路径。

#### `aep.connection.auth.client.code`

指定 IMS 授权码交换所需的授权码。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；仅 `access_token` 方式必填，其他方式不使用。它不是预签发的 Bearer Token；属于敏感凭证，可能出现在原始配置日志中。

### 旧版 JWT 认证

#### `aep.connection.auth.imsOrg`

指定 JWT 认证的 IMS 组织 ID。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；仅旧版 `jwt_token` 方式必填，其他认证方式不使用。该认证路径已弃用，替代路径为符合 Adobe 身份集成要求的 `oauth2_access_token`。

#### `aep.connection.auth.accountKey`

指定 JWT 认证的技术账号标识。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；仅旧版 `jwt_token` 方式必填，不是 RSA 私钥。该认证路径已弃用；其他方式不使用，迁移时改用对应的 OAuth2 Client 配置。

#### `aep.connection.auth.filePath`

指定 JWT 认证使用的 RSA 私钥文件路径。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：无运行时回退；仅旧版 `jwt_token` 方式必填。每个承担 Task 的 Worker 都必须能读取该本地文件；使用 PEM 编码的 PKCS#8 私钥，文件不得超过 1 MiB。不要将私钥正文写入配置。该认证路径已弃用，OAuth2 方式不使用此项。

### HTTP 代理

#### `aep.connection.proxy.host`

指定代理主机，供采集请求及认证提供方使用。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `null`，不指定代理主机；代理行为由 HTTP 连接处理。

#### `aep.connection.proxy.port`

指定代理端口。

* **类型**：`int`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `443`。配置代理主机时填写有效的 TCP 端口；不存在 Connector 级范围校验。

#### `aep.connection.proxy.user`

指定需要认证的代理用户名。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `null`；按代理要求与密码配合。代理认证使用 JVM 全局默认 Authenticator，需考虑同 Worker 内其他连接的影响。

#### `aep.connection.proxy.password`

指定代理认证密码。

* **类型**：`string`（运行时读取类型）
* **默认值**：无 ConfigDef 默认值
* **重要级别**：未声明
* **有效值 / 注意事项**：缺失或空白时运行时使用 `null`。没有自动 PASSWORD 脱敏保护；启动原始配置日志和代理 DEBUG 日志均存在凭证暴露风险。

### 错误处理

#### `errors.tolerance`

控制框架可处理记录错误的容忍策略。

* **类型**：`string`
* **默认值**：`none`
* **重要级别**：中
* **有效值 / 注意事项**：`none` 或 `all`。`all` 不代表所有远端 HTTP 错误均可恢复，也不能防止 Connector 内部未报告的记录丢弃；401/403 仍可能导致 Task 失败。

#### `errors.deadletterqueue.topic.name`

指定 Sink 错误记录的 DLQ Topic。

* **类型**：`string`
* **默认值**：空字符串
* **重要级别**：中
* **有效值 / 注意事项**：空值禁用 DLQ；非空时必须从输入订阅排除。只有进入框架错误报告路径的记录才可能写入，不能将其视为全部 AEP 失败记录的备份。

#### `errors.deadletterqueue.topic.replication.factor`

设置自动创建 DLQ Topic 时的副本因子。

* **类型**：`short`
* **默认值**：`3`
* **重要级别**：中
* **有效值 / 注意事项**：针对尚不存在的 DLQ Topic，应与 Kafka 部署可支持的副本数匹配；不改变已有 Topic 的副本数。

#### `errors.deadletterqueue.context.headers.enable`

为 DLQ 记录附加错误上下文请求头。

* **类型**：`boolean`
* **默认值**：`false`
* **重要级别**：中
* **有效值 / 注意事项**：启用时添加 `__connect.errors.` 前缀的 Headers；需评估上下文信息暴露风险。这些 Headers 不属于发往 AEP 的消息头。

#### `errors.log.enable`

控制框架错误日志。

* **类型**：`boolean`
* **默认值**：`false`
* **重要级别**：中
* **有效值 / 注意事项**：独立于 Connector 日志级别；关闭时也不能阻止 Connector 启动日志输出原始配置。

#### `errors.log.include.messages`

控制框架错误日志是否包含消息详情。

* **类型**：`boolean`
* **默认值**：`false`
* **重要级别**：中
* **有效值 / 注意事项**：敏感输入应保持关闭；此项不是 Adobe 配置或代理凭证的脱敏机制。

## 最佳实践

### 积压增长时增加消费并行度

**适用业务场景**：已完成接入并确认数据能正常写入 Dataset，但持续运行时单 Task 的消费积压不断增长。希望利用输入 Topic 的多个分区并行发送，同时保持消息格式和目标入口不变。

**配置示例**：在快速开始的配置上添加以下设置。这里的 `2` 是扩展起点示例，不是通用最优值；输入 Topic 应具有足够的可分配分区。

```properties theme={null}
tasks.max=2
```

**关键说明**：先检查是否因 AEP 限流、请求失败或入库验证错误导致积压；这些情况不应靠盲目增加 Task 解决。各 Task 独立向同一入口发送，增加 Task 会增加并发 HTTP 请求，但不会提供跨分区全局顺序。调整后比较消费积压、请求延迟和 AEP 采集及入库情况；若目标端压力或错误增加，应降低并行度。输入分区不足时提高上限不会获得相应并行度。

## 监控

### 监控内容

关注 Kafka Connect 集群健康、Connector / Task 状态、消费吞吐与积压、端到端延迟、Offset 提交、错误和重试，以及 Worker JVM 的内存、GC 和线程信号；只有配置了相关错误处理时才关注 DLQ 活动。RUNNING、HTTP 2xx 或已提交 Offset 均不代表 Dataset 入库成功，还需独立观察 AEP Dataflow、Dataset 批次状态和流式验证结果，将采集层接受量与最终入库结果分别核对。

### 导入 Grafana 大盘

下载共享的 [Kafka Connect Grafana Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json)，确认已采集 Connect / Worker 指标并配置兼容的数据源及大盘所需的集群、Connector、Task 等标签，然后在 Grafana 导入 JSON 并选择对应数据源。

## 限制条件

* Connector 不自动创建 AEP 资源、执行 Data Prep 映射或补充 XDM 消息结构；也不提供按 Topic 配置不同目标入口的路由。
* 不提供 CDC 删除、更新合并或 Schema 演进管理；Kafka 空值和 tombstone 不代表 AEP 删除操作，可能触发数据转换异常。
* 原始 JSON 字符串解析失败时可能被直接排除且不进入错误报告路径；部分目标端错误也可能在记录日志后继续推进 Offset，不能将 Offset 推进解释为完整交付。
* 不提供事务写入、幂等键或自动去重；HTTP 重试、Offset 提交前的重启及重平衡可能产生重复，不保证无条件至少一次送达 AEP，也不保证恰好一次。
* 批量阈值不等于最终 HTTP 请求字节大小，不能自动确保 Adobe 的 1 MB 请求上限；单条大消息也可能超过阈值。
* 不提供全局事件顺序或 AEP 最终入库顺序保障，各 Task 和分区可能并行处理。

## 常见问题

### Task 显示 RUNNING，为什么 Dataset 中仍没有数据？

启动状态不是入口连接检查，HTTP 接受也不等于 Dataset 入库。确认入口地址、Dataflow 状态、消息内的 `flowId` 和 `datasetId` 是否对应，再检查目标 Schema、XDM 字段类型和 AEP 流式验证及 Dataset 批次错误。新建 Dataflow 后等待约五分钟再发送。不要只凭 Offset 推进判断成功，应使用可识别的业务事件核对最终入库结果。

### 为什么收到 401 或 403 后没有自动恢复？

检查 AEP 入口是否要求认证，以及 Token 是否有效、身份集成是否拥有所需权限。启用 Connector 认证必须使用精确的小写 `true`，填写选定方式必需的参数；不要依赖隐式选择旧版 JWT。OAuth2 按 Token 过期时间缓存和刷新，但不会因 401 自动清除缓存并重新获取 Token 后重发。处理授权问题后再恢复 Task；排查时先确保日志不会暴露凭证。

### HTTP 返回成功，为什么仍可能有部分消息失败？

批量请求的 207 响应可以包含逐条结果，不能只检查 HTTP 状态码。Connector 在 `status` 字段存在且值非 `null` 时判断为失败，不要求字符串非空，并从 `xactionId` 最后一个连字符后的内容获取记录索引；Adobe 公开样例则使用 `statusCode`，成功标识可能以冒号分隔，失败项还可能没有 `xactionId`。这是样例与插件解析规则的静态契约差异，并不代表已经确认真实云端发生故障。上线前应核对实际入口响应，尤其不能依赖插件识别只含 `statusCode` 的失败项；它也不会自动重试失败的单条消息。结合 AEP 验证和 Dataset 状态核对结果，必要时联系维护方确认响应兼容性，获取响应时应避免泄露业务数据。

### 配置了 DLQ，为什么没有找到所有失败消息？

DLQ 只覆盖被交给框架错误报告路径的记录，不覆盖所有内部解析丢弃或所有远端响应问题；启用错误容忍也不会解决 AEP 入库后的验证失败。检查错误容忍策略及输入订阅是否排除了 DLQ，再确认 DLQ Topic 已存在，或具备自动创建所需的权限和副本数条件。结合 Task 日志和 AEP 结果核对，不要把没有 DLQ 记录当作全部消息成功，也不要把记录报告等同于同步持久化完成。

### 为什么恢复或重启后出现重复数据？

请求可能已被 AEP 接受，但连接中断或 Kafka Offset 尚未提交，随后 HTTP 重试或恢复消费会重新发送。Connector 不自动去重，应按业务事件标识核对重复，并在目标数据处理链路单独设计、确认去重规则；不能假定 XDM 的 `_id` 字段本身就让该 Connector 获得恰好一次语义。