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

# RSS Source Connector

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

## 概述

RSS Source Connector 定期读取一个或多个 RSS 或 Atom Feed，并将 Feed 中的条目写入同一个 Kafka Topic。每个可处理的条目对应一条 Kafka 记录，记录值包含条目标题、标识、链接、`description` 内容、作者、日期，以及 Feed 的 URL 和标题；记录键为空。

Connector 每次轮询当前 Feed 内容，并使用为各 URL 保存的条目指纹避免重复发送仍可识别的旧条目。指纹由条目的标题、链接、标识、`description` 内容和作者生成，不包含日期或 Feed 标题。它适合把新闻、博客、公告或其他 Feed 更新接入 Kafka，供下游检索、通知、分析或流处理应用消费。

## 授权许可

使用 MIT License。

## 快速开始

提前准备 Connect Cluster、Kafka 和可访问的 RSS 或 Atom Feed，并确认网络连通和访问权限。具体准备和管理操作请参阅 [管理 Connector](../manage-connectors)。

```properties theme={null}
connector.class=org.kaliy.kafka.connect.rss.RssSourceConnector
rss.urls=<rss-feed-url>
topic=<topic-name>
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
```

将 `<rss-feed-url>` 替换为 Worker 可访问的 RSS 或 Atom URL，将 `<topic-name>` 替换为接收记录的 Kafka Topic。URL 中的空格必须进行百分号编码。该示例使用带 Schema 的 JSON 输出，以保留 Connector 定义的字段结构；提交配置后，首次轮询会读取 Feed 当前包含的条目，后续轮询只输出尚未记录或指纹发生变化的条目。仅日期或 Feed 标题变化不会生成新指纹。

## 配置

### Feed 输入

#### `rss.urls`

要轮询的 RSS 或 Atom Feed URL。

* **类型**：`string`
* **默认值**：无
* **重要级别**：高
* **有效值 / 注意事项**：至少提供一个有效 URL；多个 URL 使用单个空格分隔。URL 自身包含的空格必须进行百分号编码，不要使用重复空格、制表符或换行符作为分隔符。应保持 URL 唯一；同一 Task 收到重复 URL 时会因初始化冲突而失败。URL 不应包含凭据、访问令牌或签名查询参数。
* **必填**：是

### Kafka 输出

#### `topic`

接收所有 Feed 条目的 Kafka Topic。

* **类型**：`string`
* **默认值**：无
* **重要级别**：高
* **有效值 / 注意事项**：使用有效且可写的 Kafka Topic 名称。一个 Connector 实例中的所有 Feed 都写入该 Topic，不支持按 Feed 分别配置 Topic。
* **必填**：是

### 轮询与任务

#### `sleep.seconds`

首次轮询之后，每个 Task 开始下一轮轮询前等待的秒数。

* **类型**：`int`
* **默认值**：`60`
* **重要级别**：中
* **有效值 / 注意事项**：必须大于或等于 `0`。设置为 `0` 会取消轮询间等待，可能形成高频循环；实际轮询周期还包含该 Task 对所有 Feed 的网络读取和解析时间。

#### `tasks.max`

该 Connector 允许创建的最大 Task 数量。

* **类型**：`int`
* **默认值**：`1`
* **重要级别**：高
* **有效值 / 注意事项**：必须大于或等于 `1`。实际 Task 数不会超过 `rss.urls` 中的 URL 数量；多个 URL 会按输入顺序轮转分配给各 Task，同一 Task 内的 URL 仍然串行轮询。

### Connector 与序列化

#### `connector.class`

要加载的 RSS Source Connector 实现类。

* **类型**：`string`
* **默认值**：无
* **重要级别**：高
* **有效值 / 注意事项**：使用 `org.kaliy.kafka.connect.rss.RssSourceConnector`。
* **必填**：是

#### `value.converter`

在 Connector 级别指定记录值使用的 Converter。

* **类型**：`class`
* **默认值**：`null`
* **重要级别**：低
* **有效值 / 注意事项**：省略或使用 `null` 时继承 Worker 的值 Converter。显式配置时必须是可实例化的 Kafka Connect `Converter` 实现，并按所选 Converter 配置其 Schema 或序列化选项；Connector 输出固定的 Connect Schema 和 Struct 值。

## 最佳实践

### 按 Feed 数量扩展轮询任务

**适用业务场景**：一个 Connector 需要持续读取多个彼此独立的 Feed，单个 Task 串行抓取所有 URL 已使更新延迟增加。此时可按 URL 数量逐步增加 Task，让不同 Feed 分配到不同 Task 并行轮询。

**配置示例**：

保留快速开始中的 Connector、Topic 和 Converter 配置，将 `rss.urls` 扩展为唯一 URL 列表，并增加或覆盖：

```properties theme={null}
rss.urls=<feed-url-1> <feed-url-2> <feed-url-3> <feed-url-4> <feed-url-5>
tasks.max=3
```

**关键说明**：实际 Task 数为 `tasks.max` 与 URL 数量中的较小值；示例中的 5 个 URL 会轮转分配给最多 3 个 Task。一个 URL 不会拆分给多个 Task，同一 Task 内的 Feed 仍按顺序抓取，因此响应缓慢的 Feed 会延迟该 Task 中后续 Feed。应从较小并行度开始，根据 Task 延迟、源站承载能力和 Worker 资源逐步调整。还应确保 URL 列表没有重复项，避免重复 URL 被分到同一 Task 后导致初始化失败。

### 按更新频率安排轮询并保留足够内容

**适用业务场景**：Feed 会持续滚动更新，需要在源站请求频率、消息新鲜度和停机恢复范围之间取得平衡。此时应根据业务可接受延迟设置轮询间隔，并让上游 Feed 保留足够长的条目窗口。

**配置示例**：

在快速开始配置中增加或覆盖以下属性：

```properties theme={null}
sleep.seconds=300
```

**关键说明**：`300` 秒只是示例。应根据业务可接受的发现延迟设置该值，并计入每轮抓取和解析所需的时间。较短间隔可更快发现新条目，但会增加源站和 Worker 负载；较长间隔会扩大条目在两次轮询之间出现后又滚出 Feed 的风险。Connector 的 Offset 是当前 Feed 中条目指纹的窗口，不是源站历史游标，因此还应在上游保留足够多的条目，并在维护或长时间停机后检查是否存在已滚出 Feed 的内容。

## 监控

### 监控内容

关注 Kafka Connect 健康状态、Connector 和 Task 状态、吞吐、延迟、Offset 提交、错误、重试和 Worker JVM 信号；仅在启用了相应错误处理时关注 DLQ 活动。对于 RSS 抓取，还应结合 Task 日志区分“Feed 没有新条目”和“URL 读取或 XML 解析失败”。

### 导入 Grafana 大盘

确认 Connect 指标已接入 Grafana 数据源，且采集标签满足大盘筛选条件；下载 [Kafka Connect Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json)，在 Grafana 中导入 JSON 并选择对应数据源。

## 限制条件

* Connector 只读取每次轮询时 Feed 中仍可见的当前条目，没有可回放的上游日志；轮询间隙或停机期间已经滚出 Feed 的条目无法补取。
* Kafka 记录发送成功与 Source Offset 持久化之间存在重复窗口，而滚动 Feed 又存在不可恢复窗口，因此不能保证恰好一次、无重复、无遗漏或无条件的至少一次交付。
* 条目必须包含可用的 `title`、`id` 和 `link`。缺少这些必填字段的条目无法构造固定 Schema，会被记录异常后过滤，且不会进入 Connector 自有 DLQ。
* 版本 0.1.1 不提供 Connector 级 HTTP 认证、自定义请求头、条件请求、代理、连接超时或读取超时配置。不要把凭据或令牌放入 `rss.urls`，因为 Task 启动日志可能包含完整配置。
* 单次轮询会返回分配给该 Task 的所有 Feed 中全部新条目，没有 Connector 级记录数、字节数或内存上限。大型 Feed 或积压内容可能形成较大的单次批次。
* Kafka 记录键和目标 Partition 均未由 Connector 指定。Feed URL 只用于 Source Offset 分区，不提供固定的 Kafka Partition 路由，也不保证跨 Feed、跨 Task 或跨 Kafka Partition 的全局顺序。
* `content` 只映射 Feed 条目的 description；category、enclosure、comments、原始 XML 及其他扩展字段不会写入记录值。

## 常见问题

### 为什么 Task 在启动时失败？

检查 `rss.urls` 是否至少包含一个有效 URL、多个 URL 之间是否只有一个空格、URL 中的空格是否已百分号编码，以及列表中是否存在重复 URL；同一 Task 收到重复 URL 时会发生初始化冲突。随后确认 `topic` 是有效且可写的 Kafka Topic，并检查 Task 日志中的配置校验或初始化异常。修正配置后重启失败的 Task。

### 为什么 Connector 显示运行中，但 Topic 中没有新记录？

Feed 可能没有新条目，也可能在本轮发生 URL 读取或 XML 解析失败。检查 Worker 是否能访问 Feed、响应内容是否仍是有效的 RSS 或 Atom XML，并查看 Task 日志中的抓取警告。若 Feed 有内容，再确认条目包含 `title`、`id` 和 `link`；缺少这些字段的条目会被过滤。修复源站内容或网络问题后，Connector 会在后续轮询中再次读取该 URL。

### 为什么重启后出现重复记录或缺少停机期间的条目？

Kafka 已确认记录但对应 Source Offset 尚未持久化时发生故障，重启后仍在 Feed 中的条目可能再次输出。另一方面，Connector 只能查看当前 Feed 快照；停机期间已经滚出 Feed 的条目无法恢复。下游应使用稳定业务字段进行幂等处理，并根据最长维护窗口调整上游 Feed 的条目保留范围。

### 为什么增加 `tasks.max` 后 Task 数量没有继续增加？

一个 Feed URL 最多分配给一个 Task，因此实际 Task 数不会超过 URL 数量。确认 `rss.urls` 中包含足够多且唯一的 URL，并检查 Connector 状态中的实际 Task 数。若多个 URL 仍位于同一 Task，可继续增加 `tasks.max`，但超过 URL 数量后不会获得额外并行度。

### 为什么输出字段与 Feed 原始 XML 不完全一致？

Connector 使用固定字段模型而不是转发原始 XML。`content` 只取条目 description，`date` 优先取更新时间、其次取发布时间，并作为 ISO-8601 字符串写入记录值，而不是 Kafka 记录时间戳。需要 category、enclosure、原始 XML 或其他扩展字段时，应在上游转换 Feed，或在下游使用能够保留这些字段的其他采集方式。