cdc: add kafka max-message-bytes related explain

ticdc/ticdc-faq.md

@@ -238,12 +238,6 @@ cdc cli changefeed create --server=http://127.0.0.1:8300 --sink-uri="kafka://127
     * `replica.fetch.max.bytes`，将 Kafka 的 `server.properties` 中该参数调大到 `1073741824` (1 GB)。
     * `fetch.message.max.bytes`，适当调大 `consumer.properties` 中该参数，确保大于 `message.max.bytes`。
 
-## TiCDC 把数据同步到 Kafka 时，能在 TiDB 中控制单条消息大小的上限吗？
-
-对于 Avro 和 Canal-JSON 格式，消息是以行变更为单位发送的，一条 Kafka Message 仅包含一条行变更。一般情况下，消息的大小不会超过 Kafka 单条消息上限，因此，一般不需要限制单条消息大小。如果单条 Kafka 消息大小确实超过 Kafka 上限，请参考[为什么 TiCDC 到 Kafka 的同步任务延时越来越大](/ticdc/ticdc-faq.md#为什么-ticdc-到-kafka-的同步任务延时越来越大)。
-
-对于 Open Protocol 格式，一条 Kafka Message 可能包含多条行变更。因此，有可能存在某条 Kafka Message 消息过大。可以通过 `max-message-bytes` 控制每次向 Kafka broker 发送消息的最大数据量（可选，默认值 10 MB），通过 `max-batch-size` 参数指定每条 kafka 消息中变更记录的最大数量（可选，默认值 `16`）。
-
 ## 在一个事务中对一行进行多次修改，TiCDC 会输出多条行变更事件吗？
 
 不会，在进行事务操作时，对于在一个事务内多次修改同一行的情况，TiDB 仅会将最新一次的修改结果发送给 TiKV。因此 TiCDC 仅能获取到最新一次修改的结果。

ticdc/ticdc-open-protocol.md

@@ -44,6 +44,25 @@ Value:
 * 长度及协议版本号均为大端序 int64 类型
 * 当前协议版本号为 `1`
 
+### 控制 Message 中的 Event 数量和大小
+
+使用 Open Protocol 将数据同步到 Kafka 时，TiCDC 可以将多个 Row Changed Event 编码到同一条 Kafka message 中。相关参数如下：
+
+| 参数 | 作用 |
+| --- | --- |
+| `max-batch-size` | 每条 Kafka message 最多包含的 Row Changed Event 数量，默认值为 `16`。 |
+| `max-message-bytes` | TiCDC 将多个 Row Changed Event 编码到同一条 Kafka message 时使用的大小阈值。实际生效值参见[配置 Kafka 消息大小](/ticdc/ticdc-sink-to-kafka.md#配置-kafka-消息大小)。 |
+
+当继续加入一个 Row Changed Event 会使消息大小超过 `max-message-bytes`，或者消息中的 Event 数量已经达到 `max-batch-size` 时，TiCDC 会创建一条新的 Kafka message。例如：
+
+```shell
+--sink-uri="kafka://127.0.0.1:9092/topic-name?protocol=open-protocol&max-message-bytes=1048576&max-batch-size=64"
+```
+
+`max-message-bytes` 不会拆分单个 Row Changed Event。从 v8.5.8 起，如果单个 Row Changed Event 编码后的消息超过 `max-message-bytes`，但没有超过 Kafka 的消息大小限制，TiCDC 会将其作为一条独立的 Kafka message 发送。
+
+关于 Kafka 消息大小限制以及 `ErrMessageTooLarge` 的处理方法，参见[配置 Kafka 消息大小](/ticdc/ticdc-sink-to-kafka.md#配置-kafka-消息大小)和[使用 TiCDC 同步消息到 Kafka 时遇到消息过大错误的处理方法](/ticdc/troubleshoot-ticdc.md#使用-ticdc-同步消息到-kafka-时-kafka-报错-message-was-too-large该如何处理)。
+
 ## Event 格式定义
 
 本部分介绍 Row Changed Event、DDL Event 和 Resolved Event 的格式定义。

ticdc/ticdc-sink-to-kafka.md

@@ -77,13 +77,13 @@ URI 中可配置的的参数如下：
 | `kafka-version`      | 下游 Kafka 版本号。该值需要与下游 Kafka 的实际版本保持一致。 |
 | `kafka-client-id`    | 指定同步任务的 Kafka 客户端的 ID（可选，默认值为 `TiCDC_sarama_producer_同步任务的 ID`）。 |
 | `partition-num`      | 下游 Kafka partition 数量（可选，不能大于实际 partition 数量，否则创建同步任务会失败，默认值 `3`）。|
-| `max-message-bytes`  | 每次向 Kafka broker 发送消息的最大数据量（可选，默认值 `10MB`，最大值为 `100MB`）。从 v5.0.6 和 v4.0.6 开始，默认值分别从 `64MB` 和 `256MB` 调整至 `10MB`。|
+| `max-message-bytes`  | TiCDC 将多条行变更编码到同一条 Kafka message 时使用的大小阈值（可选，默认值 `10 MB`，最大值为 `100 MB`）。该参数不限制单条行变更编码后的消息大小。Kafka 能接收的消息大小由 topic 的 `max.message.bytes` 或 broker 的 `message.max.bytes` 决定。从 v5.0.6 和 v4.0.6 开始，该参数的默认值分别从 `64 MB` 和 `256 MB` 调整至 `10 MB`。使用 Open Protocol 时的详细说明，参见[控制 Message 中的 Event 数量和大小](/ticdc/ticdc-open-protocol.md#控制-message-中的-event-数量和大小)。|
 | `replication-factor` | Kafka 消息保存副本数（可选，默认值 `1`），需要大于等于 Kafka 中 [`min.insync.replicas`](https://kafka.apache.org/33/documentation.html#brokerconfigs_min.insync.replicas) 的值。 |
 | `required-acks`      | 在 `Produce` 请求中使用的配置项，用于告知 broker 需要收到多少副本确认后才进行响应。可选值有：`0`（`NoResponse`：不发送任何响应，只有 TCP ACK），`1`（`WaitForLocal`：仅等待本地提交成功后再响应）和 `-1`（`WaitForAll`：等待所有同步副本提交后再响应。最小同步副本数量可通过 broker 的 [`min.insync.replicas`](https://kafka.apache.org/33/documentation.html#brokerconfigs_min.insync.replicas) 配置项进行配置）。（可选，默认值为 `-1`）。                      |
 | `compression`        | 设置发送消息时使用的压缩算法（可选值为 `none`、`lz4`、`gzip`、`snappy` 和 `zstd`，默认值为 `none`）。注意 Snappy 压缩文件必须遵循[官方 Snappy 格式](https://github.com/google/snappy)。不支持其他非官方压缩格式。|
 | `auto-create-topic` | 当传入的 `topic-name` 在 Kafka 集群不存在时，TiCDC 是否要自动创建该 topic（可选，默认值 `true`）。 |
 | `enable-tidb-extension` | 可选，默认值是 `false`。当输出协议为 `canal-json` 时，如果该值为 `true`，TiCDC 会发送 [WATERMARK 事件](/ticdc/ticdc-canal-json.md#watermark-event)，并在 Kafka 消息中添加 TiDB 扩展字段。从 6.1.0 开始，该参数也可以和输出协议 `avro` 一起使用。如果该值为 `true`，TiCDC 会在 Kafka 消息中添加[三个 TiDB 扩展字段](/ticdc/ticdc-avro-protocol.md#tidb-扩展字段)。|
-| `max-batch-size` |  从 v4.0.9 开始引入。当消息协议支持把多条变更记录输出至一条 Kafka 消息时，该参数用于指定这一条 Kafka 消息中变更记录的最多数量。目前，仅当 Kafka 消息的 `protocol` 为 `open-protocol` 时有效（可选，默认值 `16`）。|
+| `max-batch-size` |  从 v4.0.9 开始引入。当消息协议支持把多条变更记录输出至一条 Kafka 消息时，该参数用于指定这一条 Kafka 消息中变更记录的最多数量。目前，仅当 Kafka 消息的 `protocol` 为 `open-protocol` 时有效（可选，默认值 `16`）。详细说明参见[控制 Message 中的 Event 数量和大小](/ticdc/ticdc-open-protocol.md#控制-message-中的-event-数量和大小)。|
 | `enable-tls` | 连接下游 Kafka 实例是否使用 TLS（可选，默认值 `false`）。 |
 | `ca`       | 连接下游 Kafka 实例所需的 CA 证书文件路径（可选）。 |
 | `cert`     | 连接下游 Kafka 实例所需的证书文件路径（可选）。 |
@@ -108,16 +108,11 @@ URI 中可配置的的参数如下：
 
 ### 最佳实践
 
-* TiCDC 推荐用户自行创建 Kafka Topic，你至少需要设置该 Topic 每次向 Kafka broker 发送消息的最大数据量和下游 Kafka partition 的数量。在创建 changefeed 的时候，这两项设置分别对应 `max-message-bytes` 和 `partition-num` 参数。
+* TiCDC 推荐用户自行创建 Kafka Topic，并根据业务需要设置该 Topic 的 `max.message.bytes` 和 partition 数量。创建 changefeed 时，可以通过 `partition-num` 指定 partition 数量。`max-message-bytes` 只控制 TiCDC 合 batch 的大小，不会修改 Kafka Topic 的 `max.message.bytes`。
 * 如果你在创建 changefeed 时，使用了尚未存在的 Topic，那么 TiCDC 会尝试使用 `partition-num` 和 `replication-factor` 参数自行创建 Topic，建议明确指定这两个参数。
 * 在大多数情况下，建议使用 `canal-json` 协议。
 * 如果 TiCDC 上游的数据变更很少，比如可能会出现超过 10 分钟没有数据变更的情况，建议在 Kafka broker 的配置文件中调大 Kafka 的连接空闲超时时间，详情参考[为什么 TiCDC 同步到 Kafka 的任务经常因 `broken pipe` 报错而失败](/ticdc/ticdc-faq.md#为什么-ticdc-同步到-kafka-的任务经常因-broken-pipe-报错而失败)。
 
-> **注意：**
->
-> 当 `protocol` 为 `open-protocol` 时，TiCDC 会将多个事件编码到同一个 Kafka 消息中，并尽量避免在此过程中生成长度超过 `max-message-bytes` 的消息。
-> 如果单条数据变更编码得到的消息大小超过了 `max-message-bytes` 个字节，changefeed 会报错，并打印错误日志。
-
 ### TiCDC 使用 Kafka 的认证与授权
 
 使用 Kafka 的 SASL 认证时配置样例如下所示：
@@ -387,9 +382,36 @@ write-key-threshold = 30000
 SELECT COUNT(*) FROM INFORMATION_SCHEMA.TIKV_REGION_STATUS WHERE DB_NAME="database1" AND TABLE_NAME="table1" AND IS_INDEX=0;
 ```
 
+## 配置 Kafka 消息大小
+
+Kafka 通过以下参数控制可以接收的消息大小：
+
+| 参数 | 作用 |
+| --- | --- |
+| Kafka Topic `max.message.bytes` | 控制目标 Kafka Topic 可以接收的消息大小。 |
+| Kafka broker `message.max.bytes` | 当目标 Topic 没有配置独立限制时，控制 broker 可以接收的消息大小。 |
+
+Kafka sink 启动时，TiCDC 会读取目标 Topic 的 `max.message.bytes`。如果无法从 Topic 配置中获取该参数，TiCDC 会读取 broker 的 `message.max.bytes`。
+
+TiCDC 实际使用的 batch 大小阈值为：
+
+```text
+min(TiCDC max-message-bytes, Kafka 消息大小限制)
+```
+
+从 v8.5.8 起，TiCDC 将 batch 大小阈值与 Kafka 的消息大小限制分开处理。TiCDC 的 `max-message-bytes` 用于控制 batch 大小，不会阻止 Kafka 已经能够接收的单条消息。使用 Open Protocol 时，参见[控制 Message 中的 Event 数量和大小](/ticdc/ticdc-open-protocol.md#控制-message-中的-event-数量和大小)。
+
+如果 TiCDC 返回 `ErrMessageTooLarge`，或者 Kafka 返回 `Message was too large`，参见[使用 TiCDC 同步消息到 Kafka 时遇到消息过大错误的处理方法](/ticdc/troubleshoot-ticdc.md#使用-ticdc-同步消息到-kafka-时-kafka-报错-message-was-too-large该如何处理)。
+
+> **注意：**
+>
+> 如果 TiCDC 因 Kafka ACL 等原因无法读取 Topic 或 broker 配置，会使用 TiCDC `max-message-bytes` 的值作为 producer 消息大小上限。若需要 TiCDC 自动读取 Kafka 的消息大小限制，请确保 TiCDC 使用的 Kafka 账号具有相应的配置读取权限。
+
 ## 处理超过 Kafka Topic 限制的消息
 
-Kafka Topic 对可以接收的消息大小有限制，该限制由 [`max.message.bytes`](https://kafka.apache.org/documentation/#topicconfigs_max.message.bytes) 参数控制。当 TiCDC Kafka sink 在发送数据时，如果发现数据大小超过了该限制，会导致 changefeed 报错，无法继续同步数据。为了解决这个问题，TiCDC 新增一个参数 `large-message-handle-option` 并提供如下解决方案。
+Kafka Topic 对可以接收的消息大小有限制，该限制由 [`max.message.bytes`](https://kafka.apache.org/documentation/#topicconfigs_max.message.bytes) 参数控制。当原始消息超过 Kafka 当前的消息大小限制时，TiCDC 可以通过 `large-message-handle-option` 处理该消息。
+
+`large-message-handle-option` 不会仅因为消息超过 TiCDC `max-message-bytes` 而触发。如果消息超过 TiCDC `max-message-bytes`，但没有超过 Kafka 的消息大小限制，TiCDC 会直接发送原始消息。
 
 目前，如下功能支持 Canal-JSON 和 Open Protocol 两种编码协议。使用 Canal-JSON 协议时，你需要在 `sink-uri` 中设置 `enable-tidb-extension=true`。
 

ticdc/troubleshoot-ticdc.md

@@ -104,17 +104,23 @@ Warning: Unable to load '/usr/share/zoneinfo/zone1970.tab' as time zone. Skippin
 
 ## 使用 TiCDC 同步消息到 Kafka 时 Kafka 报错 `Message was too large`，该如何处理？
 
-仅在 Sink URI 中为 Kafka 配置 `max-message-bytes` 参数不能有效控制输出到 Kafka 的消息大小，需要在 Kafka server 配置中加入如下配置以增加 Kafka 接收消息的字节数限制。
+当 TiCDC 返回 `ErrMessageTooLarge` 或 Kafka 返回 `Message was too large` 时，表示待发送的消息超过了 Kafka 当前允许的消息大小。请根据错误信息中的消息大小调大 Kafka Topic 的 `max.message.bytes`。如果目标 Topic 使用 broker 的默认配置，请调大 Kafka server 的 `message.max.bytes`。
 
 ```
-# broker 能接收消息的最大字节数
-message.max.bytes=2147483648
+# Topic 能接收消息的最大字节数
+max.message.bytes=<不小于待发送消息的大小>
+# broker 能接收消息的最大字节数，使用 broker 默认限制时配置
+message.max.bytes=<不小于待发送消息的大小>
 # broker 可复制的消息的最大字节数
-replica.fetch.max.bytes=2147483648
+replica.fetch.max.bytes=<不小于 message.max.bytes>
 # 消费者端的可读取的最大消息字节数
-fetch.message.max.bytes=2147483648
+fetch.message.max.bytes=<不小于 Kafka 中实际允许的消息大小>
 ```
 
+从 v8.5.8 起，Kafka sink 重试或重建后会重新读取 Kafka 配置。调大 Kafka 的消息大小限制后，TiCDC 可以在重试过程中自动恢复同步，不需要同时修改 changefeed 的 `max-message-bytes`，也不需要暂停或恢复 changefeed。
+
+如果不希望调大 Kafka 的消息大小限制，可以配置 `large-message-handle-option`，使用 Claim-Check 或只输出 Handle Key 的方式处理大消息。更多信息参见[配置 Kafka 消息大小](/ticdc/ticdc-sink-to-kafka.md#配置-kafka-消息大小)和[处理超过 Kafka Topic 限制的消息](/ticdc/ticdc-sink-to-kafka.md#处理超过-kafka-topic-限制的消息)。
+
 ## TiCDC 同步时，在下游执行 DDL 语句失败会有什么表现，如何恢复？
 
 如果某条 DDL 语句执行失败，同步任务 (changefeed) 会自动停止，checkpoint-ts 断点时间戳为该条出错 DDL 语句的结束时间戳 (finish-ts)。如果希望让 TiCDC 在下游重试执行这条 DDL 语句，可以使用 `cdc cli changefeed resume` 恢复同步任务。例如：