cdc: add kafka max-message-bytes related explain

ticdc/ticdc-faq.md

@@ -240,9 +240,11 @@ cdc cli changefeed create --server=http://127.0.0.1:8300 --sink-uri="kafka://127
 
 ## TiCDC 把数据同步到 Kafka 时，能在 TiDB 中控制单条消息大小的上限吗？
 
-对于 Avro 和 Canal-JSON 格式，消息是以行变更为单位发送的，一条 Kafka Message 仅包含一条行变更。一般情况下，消息的大小不会超过 Kafka 单条消息上限，因此，一般不需要限制单条消息大小。如果单条 Kafka 消息大小确实超过 Kafka 上限，请参考[为什么 TiCDC 到 Kafka 的同步任务延时越来越大](/ticdc/ticdc-faq.md#为什么-ticdc-到-kafka-的同步任务延时越来越大)。
+`max-message-bytes` 控制 TiCDC 将多条行变更合并为一条 Kafka message 时的大小，默认值为 `10 MB`。该参数不限制单条行变更编码后的消息大小。Kafka 能接收的消息大小由 Topic 的 `max.message.bytes` 或 broker 的 `message.max.bytes` 决定。
 
-对于 Open Protocol 格式，一条 Kafka Message 可能包含多条行变更。因此，有可能存在某条 Kafka Message 消息过大。可以通过 `max-message-bytes` 控制每次向 Kafka broker 发送消息的最大数据量（可选，默认值 10 MB），通过 `max-batch-size` 参数指定每条 kafka 消息中变更记录的最大数量（可选，默认值 `16`）。
+对于 Open Protocol，一条 Kafka message 可能包含多条行变更。可以通过 `max-message-bytes` 控制合 batch 的大小，并通过 `max-batch-size` 控制每条 Kafka message 中的最大行变更数量。如果单条行变更编码后的消息超过 `max-message-bytes`，但没有超过 Kafka 的消息大小限制，TiCDC 仍会将其作为一条独立的 Kafka message 发送。
+
+如果需要限制 Kafka 实际接收的消息大小，请配置 Kafka Topic 的 `max.message.bytes` 或 broker 的 `message.max.bytes`。详情参见[配置 Kafka 消息大小](/ticdc/ticdc-sink-to-kafka.md#配置-kafka-消息大小)。
 
 ## 在一个事务中对一行进行多次修改，TiCDC 会输出多条行变更事件吗？
 

ticdc/ticdc-sink-to-kafka.md

@@ -77,7 +77,7 @@ 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`。|
 | `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)。不支持其他非官方压缩格式。|
@@ -108,15 +108,14 @@ 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 会报错，并打印错误日志。
+> 当 `protocol` 为 `open-protocol` 时，TiCDC 会将多个事件编码到同一个 Kafka message 中，并使用 `max-message-bytes` 控制合 batch 的大小。如果单条数据变更编码后的消息超过 `max-message-bytes`，但没有超过 Kafka Topic 的 `max.message.bytes`，TiCDC 仍会将其作为一条独立的 Kafka message 发送。
 
 ### TiCDC 使用 Kafka 的认证与授权
 
@@ -387,9 +386,45 @@ 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 消息大小
+
+TiCDC Kafka sink 和 Kafka 分别通过以下参数控制消息大小：
+
+| 参数 | 作用 |
+| --- | --- |
+| TiCDC `max-message-bytes` | 控制 TiCDC 将多条行变更合并为一条 Kafka message 时的大小。默认值为 `10 MB`。 |
+| 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 消息大小限制)
+```
+
+`max-message-bytes` 只影响 TiCDC 是否继续向当前 batch 合入更多行变更，不会阻止 Kafka 已经能够接收的单条消息。
+
+例如，使用以下配置时：
+
+```text
+TiCDC max-message-bytes = 10 MiB
+Kafka Topic max.message.bytes = 20 MiB
+单条行变更编码后的消息大小 = 12 MiB
+```
+
+TiCDC 不会再向这条 12 MiB 的消息合入其他行变更，但可以将它作为一条独立的 Kafka message 发送。
+
+如果单条消息超过 Kafka 当前的消息大小限制，TiCDC 会返回 `ErrMessageTooLarge`。此时，请调大 Kafka Topic 的 `max.message.bytes`。如果目标 Topic 使用 broker 的默认配置，请调大 broker 的 `message.max.bytes`。Kafka sink 重试或重建后会重新读取 Kafka 配置，不需要同时修改 TiCDC changefeed 的 `max-message-bytes`。
+
+> **注意：**
+>
+> 如果 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,21 @@ 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 `max-message-bytes` 参数只控制 TiCDC 合 batch 的大小，不控制 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 中实际允许的消息大小>
 ```
 
+Kafka sink 重试或重建后会重新读取 Kafka 配置。通常不需要同时修改 changefeed 的 `max-message-bytes`。更多信息参见[配置 Kafka 消息大小](/ticdc/ticdc-sink-to-kafka.md#配置-kafka-消息大小)。
+
 ## TiCDC 同步时，在下游执行 DDL 语句失败会有什么表现，如何恢复？
 
 如果某条 DDL 语句执行失败，同步任务 (changefeed) 会自动停止，checkpoint-ts 断点时间戳为该条出错 DDL 语句的结束时间戳 (finish-ts)。如果希望让 TiCDC 在下游重试执行这条 DDL 语句，可以使用 `cdc cli changefeed resume` 恢复同步任务。例如：