概述
Ably Sink Connector 将 Kafka Topic 中的事件发布到 Ably Channel，连接后端事件流与实时订阅客户端，适用于状态更新、业务事件通知和实时看板。每条 Kafka 记录对应一条 Ably 消息，目标 Channel 可以固定，也可以根据 Topic、分区或记录字段动态选择；消息名称可用于区分事件类型。 消息内容取决于 Kafka Connect Converter 的输出。带 Schema 的 Struct 会转换为 JSON 内容，不附带 Schema 定义；字符串和字节内容走相应的消息转换路径。无 Schema 的 Map 或 List 不会自动转换为结构化 JSON，因此需要 JSON 内容时，应提供 JSON 字符串或带 Schema 的 Struct。 恢复消费依赖 Kafka Connect 已提交的 Offset。发布成功但尚未提交 Offset 时发生故障，或请求响应丢失后重试，都可能产生重复消息。不能将此 Connector 视为恰好一次投递或无条件至少一次投递方案；对于不能遗漏或重复处理的业务，应在事件内容中保留业务事件 ID，并配合下游去重、对账和补发机制。前置条件
- Ably API Key 对所有静态或动态目标 Channel 具备
publish权限。 - 使用嵌套字段模板时，Converter 必须提供对应的 Connect Schema 和 Struct。
授权许可
使用 Apache License 2.0。快速开始
提前准备 Connect Cluster、Kafka Topic 和 Ably 账号及目标 Channel，并确认网络连通与访问权限。集群准备和 Connector 管理操作参见管理 Connector。以下示例消费 UTF-8 字符串消息，将其发布到固定 Channel。client.id 是非空的 Ably 客户端身份，不是 Kafka 消费者客户端标识。示例不依赖记录 Key，因此沿用 Worker 的 Key Converter；若要使用 Key 模板，需另外确认 Key 解码方式。凭证应通过部署环境的安全配置机制注入，不要写入版本库或共享日志。
配置
下面列出全部公开插件配置，以及直接影响订阅、并行度和消息解码的六项 Kafka Connect 框架配置。默认值“无”表示没有默认值，必填项需显式配置；null 表示未指定，[] 表示空列表，二者不等同于空字符串。配置名称区分大小写。
Connector 与订阅
connector.class
指定 Sink Connector 实现类。
- 类型：
STRING - 默认值：无
- 必填：是
- 重要级别：高
- 有效值 / 注意事项：使用
com.ably.kafka.connect.ChannelSinkConnector，所有运行 Task 的 Worker 均需能够加载该插件。
topics
指定需要消费的 Kafka Topic。
- 类型：
LIST - 默认值：
[] - 重要级别：高
- 有效值 / 注意事项：多个名称用逗号分隔；与
topics.regex二选一且必须配置其中一项。启用 DLQ 时不能订阅 DLQ Topic。
topics.regex
通过正则表达式选择需要消费的 Kafka Topic。
- 类型：
STRING - 默认值：
"" - 重要级别：高
- 有效值 / 注意事项：使用 Java 正则语法；与
topics互斥，启用 DLQ 时不能匹配 DLQ Topic。
tasks.max
设置最大 Task 数量。
- 类型：
INT - 默认值：
1 - 重要级别：高
- 有效值 / 注意事项：至少为
1；实际消费并行度受订阅分区数约束。同一分区不会同时分配给同一消费组的多个 Task，与每个 Task 的批执行线程数分别控制。
认证与日志
client.key
设置用于发布消息的 Ably API Key。
- 类型：
PASSWORD - 默认值：无
- 必填：是
- 重要级别：高
- 有效值 / 注意事项：使用有效 API Key，权限需覆盖所有目标 Channel；不要将真实值暴露在配置示例、版本库或日志中。
client.id
设置 Ably 客户端身份。
- 类型：
STRING - 默认值：无
- 必填：是
- 重要级别：高
- 有效值 / 注意事项：非空字符串；不是 Kafka 消费者的
client.id。
client.loglevel
设置 Ably SDK 日志级别。
- 类型：
INT - 默认值：
2 - 重要级别：低
- 有效值 / 注意事项：已定义级别为
2（VERBOSE）、3（DEBUG）、4（INFO）、5（WARN）、6（ERROR）、99（NONE）。生产环境按排障需要选择，避免长期输出过量调试信息。
消息解码与映射
key.converter
设置 Kafka 记录 Key 的 Converter。
- 类型：
CLASS - 默认值：
null - 重要级别：低
- 有效值 / 注意事项：未指定时继承 Worker 设置。使用
#{key}或#{key.<path>}时，需选择与实际编码一致的 Converter；嵌套路径要求 Schema/Struct。
value.converter
设置 Kafka 记录 Value 的 Converter。
- 类型：
CLASS - 默认值：
null - 重要级别：低
- 有效值 / 注意事项：未指定时继承 Worker 设置。字符串可使用
org.apache.kafka.connect.storage.StringConverter；#{value.<path>}需要 Schema/Struct，不能仅提供 JSON 文本或无 Schema Map。Converter 自身的选项需按所选 Converter 配置。
channel
设置目标 Ably Channel 的静态名称或模板。
- 类型：
STRING - 默认值：无
- 必填：是
- 重要级别：高
- 有效值 / 注意事项：非空字符串，首字符不能为冒号、逗号、空白或
[，名称不能包含换行。模板支持#{topic}、#{topic.name}、#{topic.partition}、#{key}、#{key.<path>}和#{value.<path>}。嵌套路径沿 Struct 字段读取，不支持数组索引或 Map 路径；字段缺失等映射错误受onFailedRecordMapping控制。模板输出也应满足 Ably Channel 命名及权限要求。
message.name
设置 Ably 消息名称，用于标识事件类型。
- 类型：
STRING - 默认值：
null - 重要级别：中
- 有效值 / 注意事项：可使用静态文本或与
channel相同的模板；未指定或为空时不设置名称。没有直接#{value}占位符。
messagePayloadSizeMax
消息载荷大小相关的公开参数。
- 类型：
INT - 默认值：
65536 - 重要级别：中
- 有效值 / 注意事项：此项不执行载荷大小限制，不应依赖它校验、截断或拆分消息。应在上游控制实际消息和请求大小，并遵守 Ably 服务限制。
批处理与并发
batchExecutionThreadPoolSize
设置每个 Task 的批发布执行线程数。
- 类型：
INT - 默认值：
10 - 重要级别：中
- 有效值 / 注意事项：必须大于
0。设置为1可串行执行该 Task 的批发布；多个线程可使批次并行完成，不能保证跨批完成顺序，也不能提供跨 Task 全局顺序。
batchExecutionMaxBufferSize
设置提交一个批次前缓冲的最大 Kafka 记录数。
- 类型：
INT - 默认值：
100 - 重要级别：中
- 有效值 / 注意事项：使用大于
0的值；达到数量阈值即提交批次。此项不是字节大小上限，也不是所有待发送数据的积压上限。
batchExecutionMaxBufferSizeMs
设置未满批缓冲的定时发送等待时间，单位毫秒。
- 类型：
INT - 默认值：
100 - 重要级别：中
- 有效值 / 注意事项：使用非负值；
0可立即触发定时发送。满批可提前发送，此项不是每条记录的最低延迟，也不是端到端延迟上限。
client.async.http.threadpool.size
设置 Ably SDK 异步 HTTP 线程池大小。
- 类型：
INT - 默认值：
64 - 重要级别：中
- 有效值 / 注意事项：这是 SDK 异步请求设置，不是此 Connector 的批发布并发控制；批发布并发使用
batchExecutionThreadPoolSize调整。
连接与备用端点
client.tls
设置是否使用 TLS 连接 Ably 服务。
- 类型：
BOOLEAN - 默认值：
true - 重要级别：中
- 有效值 / 注意事项：
true或false；生产环境保持启用，避免以明文传输凭证和消息。
client.rest.host
设置自定义 REST 主机。
- 类型：
STRING - 默认值：
null - 重要级别：低
- 有效值 / 注意事项：未指定时由 SDK 选择 REST 主机；仅在确有自定义端点需求时设置，并确认认证、网络和备用端点配置一致。使用非默认 REST 主机时，不要同时设置
client.environment。
client.port
设置非 TLS 连接的端口。
- 类型：
INT - 默认值：
0 - 重要级别：低
- 有效值 / 注意事项：
0表示使用 SDK 的非 TLS 默认端口80；仅在client.tls=false时使用。显式设置时应使用有效服务端口。
client.tls.port
设置 TLS 连接的端口。
- 类型：
INT - 默认值：
0 - 重要级别：低
- 有效值 / 注意事项：
0表示使用 SDK 的 TLS 默认端口443；仅在client.tls=true时使用。显式设置时应使用有效服务端口。
client.environment
设置非默认的 Ably 服务环境。
- 类型：
STRING - 默认值：
null - 重要级别：低
- 有效值 / 注意事项：仅在 Ably 服务部署要求时配置，端点选择交由 SDK 处理；不能与非默认
client.rest.host同时使用。它不是 Kafka Connect 或托管平台的环境标识。
client.http.open.timeout
设置建立 HTTP 连接的超时时间，单位毫秒。
- 类型：
INT - 默认值：
4000 - 重要级别：中
- 有效值 / 注意事项：按网络条件设置合理的非负超时，不要将此项作为整个消息处理流程的超时。
client.http.request.timeout
设置单次 HTTP 请求及响应的超时时间，单位毫秒。
- 类型：
INT - 默认值：
10000 - 重要级别：中
- 有效值 / 注意事项：按网络和请求规模设置合理的非负超时；不包含整个 Kafka Connect 错误重试窗口。
client.http.max.retry.count
设置 SDK HTTP 备用主机重试次数。
- 类型：
INT - 默认值：
3 - 重要级别：中
- 有效值 / 注意事项：建议使用非负值；重试需要可用备用主机且错误符合 SDK 主机失败条件。此项不会让所有 HTTP 错误重试，也不是 Kafka 消费重试策略。
client.fallback.hosts
设置 HTTP 请求可使用的备用主机列表。
- 类型：
LIST - 默认值：
[] - 重要级别：中
- 有效值 / 注意事项：多个主机用逗号分隔，必须与实际 Ably 环境匹配。空列表表示没有备用候选，不会自动启用 SDK 内置备用主机；仅增加重试次数不足以启用备用主机重试。
代理
client.proxy
设置是否启用 HTTP 代理。
- 类型：
BOOLEAN - 默认值：
false - 重要级别：中
- 有效值 / 注意事项：启用时必须提供代理主机和实际端口。未启用代理时，代理参数的类型和枚举值仍需合法。
client.proxy.host
设置 HTTP 代理主机。
- 类型：
STRING - 默认值：
null - 重要级别：中
- 有效值 / 注意事项：
client.proxy=true时需设置非空、可达的代理主机。
client.proxy.port
设置 HTTP 代理端口。
- 类型：
INT - 默认值：
0 - 重要级别：中
- 有效值 / 注意事项：启用代理时必须设置实际有效端口；默认
0不会自动选择80或443。
client.proxy.username
设置代理认证用户名。
- 类型：
STRING - 默认值：
null - 重要级别：中
- 有效值 / 注意事项：不是 Ably 身份；启用代理并设置用户名时，还必须设置
client.proxy.password。
client.proxy.password
设置代理认证密码。
- 类型：
PASSWORD - 默认值：
null - 重要级别：中
- 有效值 / 注意事项：与代理用户名配套配置；通过安全配置机制注入，不记录真实值。
client.proxy.pref.auth.type
设置优先使用的代理认证类型。
- 类型：
STRING - 默认值：
BASIC - 重要级别：中
- 有效值 / 注意事项：仅允许
BASIC、DIGEST、X_ABLY_TOKEN，区分大小写；选择代理实际支持的类型。
client.proxy.non.proxy.hosts
设置不经过代理的主机列表。
- 类型：
LIST - 默认值：
null - 重要级别：中
- 有效值 / 注意事项：多个主机用逗号分隔，由 SDK 的代理配置处理；不要假定支持任意通配符语法。
错误处理与 Push 选项
onFailedRecordMapping
设置 Channel 或消息名称映射失败时的处理策略。
- 类型：
STRING - 默认值：
stop - 重要级别：中
- 有效值 / 注意事项：仅允许
stop、skip、dlq，区分大小写。stop停止批处理，skip丢弃映射失败记录；dlq需要可用的 Kafka Connect 错误记录报告机制、已配置的errors.deadletterqueue.topic.name和允许持续处理的errors.tolerance=all。缺少报告机制时不能继续按 DLQ 模式处理。此项不覆盖全部 Converter、内容转换或 HTTP 发布错误。
client.push.full.wait
设置传递给 Ably SDK 的 Push REST 操作等待选项。
- 类型：
BOOLEAN - 默认值：
false - 重要级别：中
- 有效值 / 注意事项：此 Sink 不执行设备 Push 管理操作，该项不控制批发布完成，也不保证设备通知送达。需要通知业务时，还须独立完成 Ably Push 配置和设备注册。
最佳实践
降低状态事件的批次交错
适用业务场景：订单或设备状态事件已按业务实体分区，希望减少同一 Task 并行发布批次造成的先后交错，而不是追求跨实体的全局顺序。 配置示例：在快速开始配置中添加以下配置，覆盖默认的批执行线程数。监控
监控内容
关注 Kafka Connect 集群和 Worker 健康状态、Connector / Task 状态、吞吐、消费与处理延迟、Offset 提交及失败、错误和重试，并结合 Worker JVM 堆内存、GC 与线程情况判断积压。Task 处于运行状态或 Offset 已提交不能单独证明所有消息已被业务客户端接收，应配合业务事件对账；仅在启用了相应错误处理时关注 DLQ 活动。导入 Grafana 大盘
下载 Kafka Connect Grafana 大盘，确保 Prometheus 兼容数据源已采集 Kafka Connect 指标且集群、Worker 等标签与大盘查询一致，然后在 Grafana 中导入 JSON 并选择对应数据源。限制条件
- 不提供 Kafka 到 Ably 的恰好一次投递保障，不应依赖自动幂等发布消除重复。
- 不提供跨分区、跨 Task 或跨 Channel 的全局顺序。
- 带 Schema 的 Value 仅支持顶层 Struct、String 和 Bytes，不支持顶层数字、布尔、Array 或 Map。
- 嵌套模板要求 Schema/Struct，不支持解析 JSON 字符串、无 Schema Map、数组索引或 Map 路径。
- 模板叶值支持字符串、Integer、Long、Boolean 和有效 UTF-8 的字节数组，不支持 Float、Double、Byte、Short 或 ByteBuffer。
- Struct 转 JSON 不支持直接的原始字节数组字段，不应假定任意逻辑类型或时间精度都能原样保留。
- 无 Value Schema 时，不导出 Kafka Key、Headers 或 Push Extras；带 Schema 时也不会自动导出 Topic、分区、Offset 或 Kafka 时间戳。
- 空 Value 不表示删除 Ably 数据，也不会自动跳过 Kafka tombstone。
- 缓冲记录数不能限制总待发送积压，持续慢请求可能导致内存压力。
messagePayloadSizeMax不限制实际载荷，消息大小、批请求大小及速率仍受 Ably 服务约束。
常见问题
Task 运行中，为什么目标 Channel 没有消息？
检查 Topic 是否有新记录、消费组位置是否已到末尾、订阅是否匹配，以及实际模板输出的 Channel 是否与订阅端一致。确认 API Key 对该 Channel 有发布权限，再检查 REST 错误、连接超时和 Worker 内存。低流量时还应考虑未满批定时发送的等待时间；运行状态本身不代表发布成功。JSON 中有字段，为什么字段模板仍然失败？
JSON 字符串和无 Schema Map 不能作为嵌套模板的数据来源。检查 Converter 输出是否为带 Schema 的 Struct、字段路径是否存在，以及叶值是否为支持的类型；按实际 Kafka 编码调整 Converter 或上游事件结构。字段重命名和删除也会影响模板，发布 Schema 变更前应同步检查引用字段。为什么消息内容不是预期的 JSON 对象，或者没有 Key 和 Headers？
无 Schema 的 Map/List 不会自动发布为结构化 JSON，无 Value Schema 时也不会生成 Key 和 Headers Extras。需要 JSON 内容时使用 JSON 字符串或带 Schema 的 Struct；需要 Extras 时使用受支持的带 Schema Value，并检查 Key 和 Header 的解码方式。字符串 Key 保持字符串，字节数组 Key 使用 Base64，其他 Key 类型不会自动导出。已增加重试次数，为什么发布失败后仍没有重试？
检查client.fallback.hosts 是否配置了与环境一致的可用备用主机；默认空列表没有备用候选。SDK 仅对符合主机失败条件的错误进行备用主机重试，不应期待权限错误、限流或所有服务端错误都自动重试。先处理凭证、权限或速率问题，不能通过增大重试次数替代这些修复。
为什么映射失败后没有记录进入 DLQ？
确认onFailedRecordMapping=dlq，并检查 Kafka Connect 的 DLQ Topic、写权限、错误容忍设置以及错误记录报告机制是否可用；仅设置插件策略不足以启用完整 DLQ 处理。还应区分映射失败与 Converter、内容格式或 HTTP 发布错误，该策略不覆盖全部错误类型。不要使用 skip 掩盖必须保留的业务事件，应为 DLQ 配置告警、修复和重放流程。