+
+
+To use the new TiCDC architecture, set the TiCDC configuration item [`newarch`](/ticdc/ticdc-server-config.md#newarch-new-in-v854-release1) to `true`.
+
+For the new TiCDC architecture, the Debezium protocol supports the following types of events:
+
+- DDL event: represents a DDL change record. After the upstream DDL statement is successfully executed, the DDL event is sent to every Message Queue (MQ) partition.
+
+- DML event: represents a row data change record. The DML event is sent when a row change occurs. It contains the information about the row after the change occurs.
+
+- WATERMARK event: represents a special point in time. It indicates that the events received before this point are complete. The WATERMARK event applies only to the TiDB extension field and takes effect when you set [`enable-tidb-extension`](/ticdc/ticdc-sink-to-kafka.md#configure-sink-uri-for-kafka) to `true` in `sink-uri`.
+
+
+
+
+For the classic TiCDC architecture, the Debezium protocol only supports Row Changed events and directly ignores DDL events and WATERMARK events. A Row changed event represents a data change in a row. When a row changes, the Row Changed event is sent, including relevant information about the row both before and after the change. A WATERMARK event marks the replication progress of a table, indicating that all events earlier than the watermark have been sent to the downstream.
+
+
+
The configuration example for using the Debezium message format is as follows:
@@ -25,30 +44,425 @@ In addition, the original Debezium format does not include important fields such
## Message format definition
-This section describes the format definition of the DML event output in the Debezium format.
+This section describes the message formats of DDL events, DML events, and WATERMARK events.
-### DML event
+### DDL event (new TiCDC architecture)
-TiCDC encodes a DML event into a Kafka message, with both the key and value encoded in the Debezium format.
+> **Note:**
+>
+> DDL events are supported only in the [new TiCDC architecture](/ticdc/ticdc-architecture.md). In the [classic TiCDC architecture](/ticdc/ticdc-classic-architecture.md), DDL events are ignored.
+
+TiCDC encodes a DDL event into a Kafka message, with both the key and value encoded in the Debezium format.
#### Key format
```json
{
"payload": {
- "a": 4
+ "databaseName": "test"
+ },
+ "schema": {
+ "type": "struct",
+ "name": "io.debezium.connector.mysql.SchemaChangeKey",
+ "optional": false,
+ "version": 1,
+ "fields": [
+ {
+ "field": "databaseName",
+ "optional": false,
+ "type": "string"
+ }
+ ]
+ }
+}
+```
+
+The fields in the key only include the database name. The fields are explained as follows:
+
+| Field | Type | Description |
+|:------------------|:--------|:------------------------------------------------|
+| `payload` | JSON | The information about database name. |
+| `schema.fields` | JSON | The type information of each field in the payload. |
+| `schema.type` | String | The data type of the field. |
+| `schema.optional` | Boolean | Indicates whether the field is optional. When it is `true`, the field is optional. |
+| `schema.version` | String | The schema version. |
+
+#### Value format
+
+```json
+{
+ "payload": {
+ "source": {
+ "version": "2.4.0.Final",
+ "connector": "TiCDC",
+ "name": "test_cluster",
+ "ts_ms": 0,
+ "snapshot": "false",
+ "db": "test",
+ "table": "table1",
+ "server_id": 0,
+ "gtid": null,
+ "file": "",
+ "pos": 0,
+ "row": 0,
+ "thread": 0,
+ "query": null,
+ "commit_ts": 1,
+ "cluster_id": "test_cluster"
+ },
+ "ts_ms": 1701326309000,
+ "databaseName": "test",
+ "schemaName": null,
+ "ddl": "RENAME TABLE test.table1 to test.table2",
+ "tableChanges": [
+ {
+ "type": "ALTER",
+ "id": "\"test\".\"table2\",\"test\".\"table1\"",
+ "table": {
+ "defaultCharsetName": "",
+ "primaryKeyColumnNames": [
+ "id"
+ ],
+ "columns": [
+ {
+ "name": "id",
+ "jdbcType": 4,
+ "nativeType": null,
+ "comment": null,
+ "defaultValueExpression": null,
+ "enumValues": null,
+ "typeName": "INT",
+ "typeExpression": "INT",
+ "charsetName": null,
+ "length": 0,
+ "scale": null,
+ "position": 1,
+ "optional": false,
+ "autoIncremented": false,
+ "generated": false
+ }
+ ],
+ "comment": null
+ }
+ }
+ ]
},
"schema": {
+ "optional": false,
+ "type": "struct",
+ "version": 1,
+ "name": "io.debezium.connector.mysql.SchemaChangeValue",
"fields": [
{
- "field": "a",
+ "field": "source",
+ "name": "io.debezium.connector.mysql.Source",
+ "optional": false,
+ "type": "struct",
+ "fields": [
+ {
+ "field": "version",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "connector",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "name",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "ts_ms",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "snapshot",
+ "optional": true,
+ "type": "string",
+ "parameters": {
+ "allowed": "true,last,false,incremental"
+ },
+ "default": "false",
+ "name": "io.debezium.data.Enum",
+ "version": 1
+ },
+ {
+ "field": "db",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "sequence",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "table",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "server_id",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "gtid",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "file",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "pos",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "row",
+ "optional": false,
+ "type": "int32"
+ },
+ {
+ "field": "thread",
+ "optional": true,
+ "type": "int64"
+ },
+ {
+ "field": "query",
+ "optional": true,
+ "type": "string"
+ }
+ ]
+ },
+ {
+ "field": "ts_ms",
+ "optional": false,
+ "type": "int64"
+ },
+ {
+ "field": "databaseName",
"optional": true,
- "type": "int32"
+ "type": "string"
+ },
+ {
+ "field": "schemaName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "ddl",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "tableChanges",
+ "optional": false,
+ "type": "array",
+ "items": {
+ "name": "io.debezium.connector.schema.Change",
+ "optional": false,
+ "type": "struct",
+ "version": 1,
+ "fields": [
+ {
+ "field": "type",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "id",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "table",
+ "optional": true,
+ "type": "struct",
+ "name": "io.debezium.connector.schema.Table",
+ "version": 1,
+ "fields": [
+ {
+ "field": "defaultCharsetName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "primaryKeyColumnNames",
+ "optional": true,
+ "type": "array",
+ "items": {
+ "type": "string",
+ "optional": false
+ }
+ },
+ {
+ "field": "columns",
+ "optional": false,
+ "type": "array",
+ "items": {
+ "name": "io.debezium.connector.schema.Column",
+ "optional": false,
+ "type": "struct",
+ "version": 1,
+ "fields": [
+ {
+ "field": "name",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "jdbcType",
+ "optional": false,
+ "type": "int32"
+ },
+ {
+ "field": "nativeType",
+ "optional": true,
+ "type": "int32"
+ },
+ {
+ "field": "typeName",
+ "optional": false,
+ "type": "string"
+ },
+ {
+ "field": "typeExpression",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "charsetName",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "length",
+ "optional": true,
+ "type": "int32"
+ },
+ {
+ "field": "scale",
+ "optional": true,
+ "type": "int32"
+ },
+ {
+ "field": "position",
+ "optional": false,
+ "type": "int32"
+ },
+ {
+ "field": "optional",
+ "optional": true,
+ "type": "boolean"
+ },
+ {
+ "field": "autoIncremented",
+ "optional": true,
+ "type": "boolean"
+ },
+ {
+ "field": "generated",
+ "optional": true,
+ "type": "boolean"
+ },
+ {
+ "field": "comment",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "defaultValueExpression",
+ "optional": true,
+ "type": "string"
+ },
+ {
+ "field": "enumValues",
+ "optional": true,
+ "type": "array",
+ "items": {
+ "type": "string",
+ "optional": false
+ }
+ }
+ ]
+ }
+ },
+ {
+ "field": "comment",
+ "optional": true,
+ "type": "string"
+ }
+ ]
+ }
+ ]
+ }
}
+ ]
+ }
+}
+```
+
+The key fields of the preceding JSON data are explained as follows:
+
+| Field | Type | Description |
+|:----------|:-------|:-------------------------------------------------------|
+| `payload.ts_ms` | Number | The timestamp (in milliseconds) when TiCDC generates this message. |
+| `payload.ddl` | String | The SQL statement of the DDL event. |
+| `payload.databaseName` | String | The name of the database where the event occurs. |
+| `payload.source.commit_ts` | Number | The `CommitTs` value of the event. |
+| `payload.source.db` | String | The name of the database where the event occurs. |
+| `payload.source.table` | String | The name of the table where the event occurs. |
+| `payload.tableChanges` | Array | A structured representation of the entire table schema after the schema change. The `tableChanges` field contains an array that includes entries for each column of the table. Because the structured representation presents data in JSON or Avro format, consumers can easily read messages without first processing them through a DDL parser. |
+| `payload.tableChanges.type` | String | Describes the kind of change. The value is one of the following: `CREATE`, indicating that the table is created; `ALTER`, indicating that the table is modified; `DROP`, indicating that the table is deleted. |
+| `payload.tableChanges.id` | String | Full identifier of the table that was created, altered, or dropped. In the case of a table rename, this identifier is a concatenation of `