Skip to content

cdc: add kafka max-message-bytes related explain #23352

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
3AceShowHand wants to merge 1 commit into
base: master
Choose a base branch
from
+65 −32
Open

cdc: add kafka max-message-bytes related explain #23352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 1 addition & 7 deletions ticdc/ticdc-faq.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,12 +243,6 @@ For more information, refer to [TiCDC changefeed configurations](/ticdc/ticdc-ch
* Increase the `replica.fetch.max.bytes` value in `server.properties` to `1073741824` (1 GB).
* Increase the `fetch.message.max.bytes` value in `consumer.properties` to make it larger than the `message.max.bytes` value.

## When TiCDC replicates data to Kafka, can I control the maximum size of a single message in TiDB?

When `protocol` is set to `avro` or `canal-json`, messages are sent per row change. A single Kafka message contains only one row change and is generally no larger than Kafka's limit. Therefore, there is no need to limit the size of a single message. If the size of a single Kafka message does exceed Kafka's limit, refer to [Why does the latency from TiCDC to Kafka become higher and higher?](/ticdc/ticdc-faq.md#why-does-the-latency-from-ticdc-to-kafka-become-higher-and-higher).

When `protocol` is set to `open-protocol`, messages are sent in batches. Therefore, one Kafka message might be excessively large. To avoid this situation, you can configure the `max-message-bytes` parameter to control the maximum size of data sent to the Kafka broker each time (optional, `10MB` by default). You can also configure the `max-batch-size` parameter (optional, `16` by default) to specify the maximum number of change records in each Kafka message.

## If I modify a row multiple times in a transaction, will TiCDC output multiple row change events?

No. When you modify the same row in one transaction multiple times, TiDB only sends the latest modification to TiKV. Therefore, TiCDC can only obtain the result of the latest modification.
Expand Down Expand Up @@ -534,4 +528,4 @@ If your application is highly sensitive to changefeed latency, it is recommended

It is recommended to adjust the value of `connections.max.idle.ms` according to your actual replication workload. For example, if a TiCDC changefeed always replicates data within several minutes, you can set `connections.max.idle.ms` to several minutes instead of a very large value.

2. Restart Kafka to apply the configuration change. This prevents connections from being closed prematurely and helps reduce `broken pipe` errors.
2. Restart Kafka to apply the configuration change. This prevents connections from being closed prematurely and helps reduce `broken pipe` errors.
13 changes: 13 additions & 0 deletions ticdc/ticdc-open-protocol.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,19 @@ Value:
* The length and protocol version are the big-endian `int64` type.
* The version of the current protocol is `1`.

### Control the number and size of Events in a Message

Open Protocol can encode one or more Row Changed Events into a single Message. The following Kafka sink parameters control the number of Row Changed Events in a Message and the Message size:

| Parameter | Description |
| --- | --- |
| `max-batch-size` | The maximum number of Row Changed Events in a Message. The default value is `16`. |
| `max-message-bytes` | The Message size threshold. |

The Kafka sink compares the `max-message-bytes` value configured in the changefeed with the message size limit allowed by Kafka and uses the smaller value as the actual threshold. This prevents a Message encoded in a batch from exceeding the size allowed by Kafka. For how Kafka determines its message size limit, see [Kafka message size limit](/ticdc/ticdc-sink-to-kafka.md#kafka-message-size-limit).

When the number of Row Changed Events in a Message reaches `max-batch-size`, or adding the next Row Changed Event would cause the Message to exceed the effective size threshold, subsequent Row Changed Events are written to a new Message.

## Event format

This section introduces the formats of Row Changed Event, DDL Event, and Resolved Event.
Expand Down
40 changes: 25 additions & 15 deletions ticdc/ticdc-sink-to-kafka.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,13 +77,13 @@ The following are descriptions of sink URI parameters and values that can be con
| `kafka-version` | The version of the downstream Kafka. This value needs to be consistent with the actual version of the downstream Kafka. |
| `kafka-client-id` | Specifies the Kafka client ID of the replication task (optional. `TiCDC_sarama_producer_replication ID` by default). |
| `partition-num` | The number of the downstream Kafka partitions (optional. The value must be **no greater than** the actual number of partitions; otherwise, the replication task cannot be created successfully. `3` by default). |
| `max-message-bytes` | The maximum size of data that is sent to Kafka broker each time (optional, `10MB` by default, and the maximum value is `100MB`). From v5.0.6 and v4.0.6, the default value has changed from `64MB` and `256MB` to `10MB`. |
| `max-message-bytes` | The size threshold for an Open Protocol Message (optional, `10 MB` by default, and the maximum value is `100 MB`). For how the effective threshold is determined, see [Control the number and size of Events in a Message](/ticdc/ticdc-open-protocol.md#control-the-number-and-size-of-events-in-a-message). |
| `max-batch-size` | The maximum number of Row Changed Events in an Open Protocol Message (optional, `16` by default). For more information, see [Control the number and size of Events in a Message](/ticdc/ticdc-open-protocol.md#control-the-number-and-size-of-events-in-a-message). |
| `replication-factor` | The number of Kafka message replicas that can be saved (optional, `1` by default). This value must be greater than or equal to the value of [`min.insync.replicas`](https://kafka.apache.org/33/documentation.html#brokerconfigs_min.insync.replicas) in Kafka. |
| `required-acks` | A parameter used in the `Produce` request, which notifies the broker of the number of replica acknowledgements it needs to receive before responding. Value options are `0` (`NoResponse`: no response, only `TCP ACK` is provided), `1` (`WaitForLocal`: responds only after local commits are submitted successfully), and `-1` (`WaitForAll`: responds after all replicated replicas are committed successfully. You can configure the minimum number of replicated replicas using the [`min.insync.replicas`](https://kafka.apache.org/33/documentation.html#brokerconfigs_min.insync.replicas) configuration item of the broker). (Optional, the default value is `-1`). |
| `compression` | The compression algorithm used when sending messages (value options are `none`, `lz4`, `gzip`, `snappy`, and `zstd`; `none` by default). Note that the Snappy compressed file must be in the [official Snappy format](https://github.com/google/snappy). Other variants of Snappy compression are not supported.|
| `auto-create-topic` | Determines whether TiCDC creates the topic automatically when the `topic-name` passed in does not exist in the Kafka cluster (optional, `true` by default). |
| `enable-tidb-extension` | Optional. `false` by default. When the output protocol is `canal-json`, if the value is `true`, TiCDC sends [WATERMARK events](/ticdc/ticdc-canal-json.md#watermark-event) and adds the [TiDB extension field](/ticdc/ticdc-canal-json.md#tidb-extension-field) to Kafka messages. From v6.1.0, this parameter is also applicable to the `avro` protocol. If the value is `true`, TiCDC adds [three TiDB extension fields](/ticdc/ticdc-avro-protocol.md#tidb-extension-fields) to the Kafka message. |
| `max-batch-size` | New in v4.0.9. If the message protocol supports outputting multiple data changes to one Kafka message, this parameter specifies the maximum number of data changes in one Kafka message. It currently takes effect only when Kafka's `protocol` is `open-protocol` (optional, `16` by default). |
| `enable-tls` | Whether to use TLS to connect to the downstream Kafka instance (optional, `false` by default). |
| `ca` | The path of the CA certificate file needed to connect to the downstream Kafka instance (optional). |
| `cert` | The path of the certificate file needed to connect to the downstream Kafka instance (optional). |
Expand All @@ -108,16 +108,11 @@ The following are descriptions of sink URI parameters and values that can be con

### Best practices

* It is recommended that you create your own Kafka Topic. At a minimum, you need to set the maximum amount of data of each message that the Topic can send to the Kafka broker, and the number of downstream Kafka partitions. When you create a changefeed, these two settings correspond to `max-message-bytes` and `partition-num`, respectively.
* It is recommended that you create the Kafka topic yourself and configure its `max.message.bytes` value and number of partitions according to your business needs. When creating a changefeed, you can specify the number of partitions using `partition-num`.
* If you create a changefeed with a Topic that does not yet exist, TiCDC will try to create the Topic using the `partition-num` and `replication-factor` parameters. It is recommended that you specify these parameters explicitly.
* In most cases, it is recommended to use the `canal-json` protocol.
* If the upstream data changes in TiCDC are infrequent, such as there might be no data changes for more than 10 minutes, it is recommended to increase the Kafka connection idle timeout in the Kafka broker configuration file. For more information, see [Why do TiCDC replication tasks to Kafka often fail with `broken pipe` errors](/ticdc/ticdc-faq.md#why-do-ticdc-replication-tasks-to-kafka-often-fail-with-broken-pipe-errors).

> **Note:**
>
> When `protocol` is `open-protocol`, TiCDC encodes multiple events into one Kafka message and avoids generating messages that exceed the length specified by `max-message-bytes`.
> If the encoded result of a single row change event exceeds the value of `max-message-bytes`, the changefeed reports an error and prints logs.

### TiCDC uses the authentication and authorization of Kafka

The following are examples when using Kafka SASL authentication:
Expand Down Expand Up @@ -163,12 +158,12 @@ The following are examples when using Kafka SASL authentication:

The usage scenarios for each permission are as follows:

| Resource type | Type of operation | Scenario |
| :-------------| :------------- | :--------------------------------|
| Cluster | `DescribeConfig`| Gets the cluster metadata while the changefeed is running |
| Topic | `Describe` | Tries to create a topic when the changefeed starts |
| Topic | `Create` | Tries to create a topic when the changefeed starts |
| Topic | `Write` | Sends data to the topic |
| Resource type | Type of operation | Scenario |
| :------------ | :---------------- | :------- |
| Cluster | `DescribeConfig` | Gets the cluster metadata while the changefeed is running |
| Topic | `Describe` | Tries to create a topic when the changefeed starts |
| Topic | `Create` | Tries to create a topic when the changefeed starts |
| Topic | `Write` | Sends data to the topic |

When creating or starting a changefeed, you can disable the `Describe` and `Create` permissions if the specified Kafka topic already exists.

Expand Down Expand Up @@ -386,9 +381,24 @@ You can query the number of Regions a table contains by the following SQL statem
SELECT COUNT(*) FROM INFORMATION_SCHEMA.TIKV_REGION_STATUS WHERE DB_NAME="database1" AND TABLE_NAME="table1" AND IS_INDEX=0;
```

## Kafka message size limit

Kafka limits the message size that each topic can receive. The following configurations determine the effective limit for a target topic:

| Configuration | Description |
| --- | --- |
| Kafka topic [`max.message.bytes`](https://kafka.apache.org/43/configuration/topic-configs/#topicconfigs_max.message.bytes) | Sets the message size limit for a specific topic and overrides the broker default. |
| Kafka broker [`message.max.bytes`](https://kafka.apache.org/43/configuration/broker-configs/#brokerconfigs_message.max.bytes) | Specifies the default limit when the topic does not set `max.message.bytes`. |

When the Kafka sink starts, it reads the effective message size limit for the target topic and uses the limit to determine whether a message is too large before sending it. If a single encoded message exceeds this limit and large-message handling is not configured, the Kafka sink returns `ErrMessageTooLarge`. For how to handle this error, see [How do I handle the `ErrMessageTooLarge` error returned by Kafka sink?](/ticdc/troubleshoot-ticdc.md#how-do-i-handle-the-errmessagetoolarge-error-returned-by-kafka-sink).

> **Note:**
>
> If the Kafka sink cannot read the message size configuration of the topic or broker because of Kafka ACL or other reasons, it uses the changefeed's `max-message-bytes` value as the local message size limit. If this value differs from the effective Kafka limit, the Kafka sink might not accurately determine whether a message can be sent and might not recover automatically after you increase the Kafka limit. Make sure that the Kafka account used by the changefeed has permission to read the topic and broker configurations. If you cannot grant this permission, set the changefeed's `max-message-bytes` value to the Kafka message size limit.

## Handle messages that exceed the Kafka topic limit

Kafka topic sets a limit on the size of messages it can receive. This limit is controlled by the [`max.message.bytes`](https://kafka.apache.org/documentation/#topicconfigs_max.message.bytes) parameter. If TiCDC Kafka sink sends data that exceeds this limit, the changefeed reports an error and cannot proceed to replicate data. To solve this problem, TiCDC adds a new configuration `large-message-handle-option` and provides the following solution.
When a message exceeds the Kafka size limit, you can configure `large-message-handle-option` to prevent the replication task from failing because the message is too large.

Currently, this feature supports two encoding protocols: Canal-JSON and Open Protocol. When using the Canal-JSON protocol, you must specify `enable-tidb-extension=true` in `sink-uri`.

Expand Down
36 changes: 26 additions & 10 deletions ticdc/troubleshoot-ticdc.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,18 +106,34 @@ If the result of `pd-ctl service-gc-safepoint --pd <pd-addrs>` does not have `gc
- If your PD version is v4.0.8 or earlier, refer to [PD issue #3128](https://github.com/tikv/pd/issues/3128) for details.
- If your PD is upgraded from v4.0.8 or an earlier version to a later version, refer to [PD issue #3366](https://github.com/tikv/pd/issues/3366) for details.

## When I use TiCDC to replicate messages to Kafka, Kafka returns the `Message was too large` error. Why?
## How do I handle the `ErrMessageTooLarge` error returned by Kafka sink?

To control the size of messages that TiCDC sends to Kafka, you can configure the `max-message-bytes` parameter in the Sink URI. However, you must also ensure that your Kafka server is configured to accept messages of that size. If a message from TiCDC exceeds the Kafka server's limit, Kafka returns a `Message was too large` error. To increase the message size limit on the Kafka server, add the following configuration to its configuration file.
How you handle this error depends on the TiCDC version:

**v8.5.8 and later versions**: increase the Kafka topic's `max.message.bytes` value based on the message size reported in the error. When the Kafka sink is rebuilt, it reads the configuration again, and the changefeed automatically resumes replication.

If the changefeed does not resume automatically, make sure that the Kafka account used by the changefeed has permission to read the topic and broker configurations. For more information, see [Kafka message size limit](/ticdc/ticdc-sink-to-kafka.md#kafka-message-size-limit).

**Versions earlier than v8.5.8**:

1. Determine the size of the message to be sent from the error message.
2. Set the Kafka topic's `max.message.bytes` value to at least the message size.
3. Pause the changefeed, set `max-message-bytes` to the same value as `max.message.bytes`, and then resume the changefeed.

When adjusting the Kafka message size limit, also check the following related configurations:

```ini
# The maximum message size accepted by the topic
max.message.bytes=<a value no smaller than the message to be sent>
# The maximum message size accepted by the broker; configure this when using the broker default limit
message.max.bytes=<a value no smaller than the message to be sent>
# The maximum message size that the broker can replicate
replica.fetch.max.bytes=<a value no smaller than message.max.bytes>
# The maximum message size that the consumer can read
fetch.message.max.bytes=<a value no smaller than the effective Kafka message size limit>
Comment on lines +125 to +133

@coderabbitai coderabbitai Bot Jul 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== Candidate files =="
git ls-files | rg '(^|/)ticdc/troubleshoot-ticdc\.md$|resources/doc-templates|scripts/(markdownlint|verify-links)' || true

echo
echo "== Target snippet =="
if [ -f ticdc/troubleshoot-ticdc.md ]; then
  nl -ba ticdc/troubleshoot-ticdc.md | sed -n '100,150p'
fi

echo
echo "== Kafka config name occurrences =="
rg -n "fetch\.message\.max\.bytes|fetch\.max\.bytes|max\.partition\.fetch\.bytes|replica\.fetch\.max\.bytes|message\.max\.bytes|max\.message\.bytes" -S . || true

echo
echo "== Surrounding context in target =="
nl -ba ticdc/troubleshoot-ticdc.md | sed -n '110,135p'

Repository: pingcap/docs

Length of output: 597

🌐 Web query:

Apache Kafka 4.3 configuration consumer fetch.max.bytes max.partition.fetch.bytes fetch.message.max.bytes replica.fetch.max.bytes

💡 Result:

In Apache Kafka 4.3, these configurations manage the amount of data transferred during fetch operations. Note that fetch.message.max.bytes is not a standard configuration key in modern Kafka versions; it is often confused with broker-level or topic-level message size limits [1][2]. 1. fetch.max.bytes This is a consumer-side configuration [1]. It defines the maximum amount of data the server should return for a fetch request [1]. While it serves as a soft limit, if the first record batch in the first non-empty partition is larger than this value, the batch is still returned to ensure consumer progress [1]. The default value is 52,428,800 bytes (50 MiB) [1]. 2. max.partition.fetch.bytes This is a consumer-side configuration [1]. It sets the maximum amount of data the server will return for each partition in a fetch request [1]. Like fetch.max.bytes, if the first record batch is larger than this limit, it will still be returned to ensure progress [1]. The default value is 1,048,576 bytes (1 MiB) [1]. 3. replica.fetch.max.bytes This is a broker-side configuration that governs the maximum number of bytes the broker will attempt to fetch from another broker during replication for a single partition [3]. Note that historical documentation shows this setting exists to manage inter-broker traffic, separate from consumer-facing fetch settings [3]. 4. Regarding fetch.message.max.bytes There is no standard configuration named fetch.message.max.bytes in Apache Kafka 4.3 [1][2][4]. You are likely referring to the broker-level message size limit, message.max.bytes, or the topic-level equivalent, max.message.bytes [1][2][4]. These settings define the maximum allowed size of a record batch and are crucial because if they are set larger than the consumer's fetch settings, the consumer may be unable to fetch batches, potentially causing processing issues [1][4]. Summary Table Configuration | Type | Purpose | Default (4.3) fetch.max.bytes | Consumer | Max data for a full fetch request [1] | 52,428,800 [1] max.partition.fetch.bytes | Consumer | Max data per partition in a fetch [1] | 1,048,576 [1] replica.fetch.max.bytes | Broker | Max data per partition during replication [3] | 1,048,576 (varies) message.max.bytes | Broker | Max batch size allowed by broker [1][4] | 1,048,588 [4]

Citations:

Use the current Kafka consumer fetch settings.

fetch.message.max.bytes is not a Kafka config key for Kafka 4.3; replace it with the consumer configs fetch.max.bytes and max.partition.fetch.bytes. Keep replica.fetch.max.bytes sized for the effective Kafka message size limit, including topic settings that override broker defaults.

Committable replacement
Suggested change
```ini
# The maximum message size accepted by the topic
max.message.bytes=<a value no smaller than the message to be sent>
# The maximum message size accepted by the broker; configure this when using the broker default limit
message.max.bytes=<a value no smaller than the message to be sent>
# The maximum message size that the broker can replicate
replica.fetch.max.bytes=<a value no smaller than message.max.bytes>
# The maximum message size that the consumer can read
fetch.message.max.bytes=<a value no smaller than the effective Kafka message size limit>
# The maximum message size accepted by the topic
max.message.bytes=<a value no smaller than the message to be sent>
# The maximum message size accepted by the broker; configure this when using the broker default limit
message.max.bytes=<a value no smaller than the message to be sent>
# The maximum message size that a broker replica fetches per partition
replica.fetch.max.bytes=<a value no smaller than the effective Kafka message size limit>
# The maximum amount of data returned in a consumer.fetch request
fetch.max.bytes=<a value no smaller than the effective Kafka message size limit>
# The maximum amount of data returned per partition to the consumer
max.partition.fetch.bytes=<a value no smaller than the effective Kafka message size limit>

Source: Path instructions

```
# The maximum byte number of a message that the broker receives
message.max.bytes=2147483648
# The maximum byte number of a message that the broker copies
replica.fetch.max.bytes=2147483648
# The maximum message byte number that the consumer side reads
fetch.message.max.bytes=2147483648
```

If you do not want to increase the Kafka message size limit, configure `large-message-handle-option` to handle large messages using Claim-Check or by sending only the Handle Key. For more information, see [Handle messages that exceed the Kafka topic limit](/ticdc/ticdc-sink-to-kafka.md#handle-messages-that-exceed-the-kafka-topic-limit).

## How can I find out whether a DDL statement fails to execute in downstream during TiCDC replication? How to resume the replication?

Expand Down Expand Up @@ -151,5 +167,5 @@ cdc cli changefeed create --server=http://127.0.0.1:8300 --sink-uri="mysql://roo
This error is typically caused by the connection failure between TiCDC and the Kafka cluster. To troubleshoot, you can check the Kafka logs and network status. One possible reason is that you did not specify the correct `kafka-version` parameter when creating the replication task, causing the Kafka client inside TiCDC to use the wrong Kafka API version when accessing the Kafka server. You can fix this issue by specifying the correct `kafka-version` parameter in the [`--sink-uri`](/ticdc/ticdc-sink-to-kafka.md#configure-sink-uri-for-kafka) configuration. For example:

```shell
cdc cli changefeed create --server=http://127.0.0.1:8300 --sink-uri "kafka://127.0.0.1:9092/test?topic=test&protocol=open-protocol&kafka-version=2.4.0"
cdc cli changefeed create --server=http://127.0.0.1:8300 --sink-uri "kafka://127.0.0.1:9092/test?topic=test&protocol=open-protocol&kafka-version=2.4.0"
```
Loading