Some transaction(s) could generate empty values in open protocol

[amyangfei]

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do? If possible, provide a recipe for reproducing the error.

   1. Setup a tidb and ticdc cluster, create changefeed to replicate all tables.
   2. Execute the following SQLs in upstream TiDB

mysql root@127.0.0.1:test> create table t2 (id varchar(32) not null primary key, a int);
Query OK, 0 rows affected

mysql root@127.0.0.1:test> begin;
Query OK, 0 rows affected

mysql root@127.0.0.1:test> insert into t2 values (2,2);
Query OK, 1 row affected

mysql root@127.0.0.1:test> delete from t2 where id = 2;
Query OK, 1 row affected

mysql root@127.0.0.1:test> commit;
Query OK, 0 rows affected

2. What did you expect to see?

   The mq sink should not output any row changes.

3. What did you see instead?

   The following output using kafka-console-consumer.sh
?{"ts":427235938124431362,"scm":"test","tbl":"t2","rid":2,"t":1} {}

   The log of MySQL sink, note both columns and pre-columns are empty

   [2021/08/24 11:36:52.871 +08:00] [DEBUG] [mysql.go:1111] ["prepare DMLs"] [rows="[{\"start-ts\":427236053953019905,\"commit-ts\":427236055407656962,\"row-id\":2,\"table\":{\"db-name\":\"test\",\"tbl-name\":\"t2\",\"tbl-id\":59,\"is-partition\":false},\"table-info-version\":427235957064073226,\"replica-id\":0,\"columns\":null,\"pre-columns\":null}]"] [sqls="[]"] [values="[]"]
   [2021/08/24 11:36:52.871 +08:00] [DEBUG] [mysql.go:994] ["Exec Rows succeeded"] [changefeed=test-cf] ["num of Rows"=0] [bucket=0]
   

4. Versions of the cluster

   • Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

     All versions since v4.0.x

   • TiCDC version (execute cdc version):

     All versions since v4.0.x

5. Subtasks

   • Open protocol. sink/mq: fix empty value in open protocol #2615
   • Avro protocol
   • Canal/Canal-json protocol
   • Maxwell protocol

[Rustin170506]

/assign

I am working on the Canal JSON.

[Rustin170506]

/assign

I am working on the Canal JSON.

/opt/kafka_2.13-2.7.1/bin # kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ticdc-test --from-beginnin
g
{"id":0,"database":"test","table":"t2","pkNames":null,"isDdl":true,"type":"CREATE","es":1635249653589,"ts":0,"sql":"create table t2 (id varchar(32) not null primary key, a int)","sqlType":null,"mysqlType":null,"data":null,"old":null}
{"id":0,"database":"test","table":"t2","pkNames":[],"isDdl":false,"type":"INSERT","es":1635249684739,"ts":0,"sql":"","sqlType":{},"mysqlType":{},"data":[null],"old":[null]}

[Rustin170506]

canal:

*UTF-80�����/8BtestJt2XHPZ<create table t2 (id varchar(32) not null primary key, a int)rtest
*<:
*UTF-80�����/8BtestJt2Xb
	rowsCount1Pb

[Rustin170506]

maxwell:

/opt/kafka_2.13-2.7.1/bin # ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ticdc-test
 --from-beginning --property print.key=true
{"ts":428693224948760587,"scm":"test","tbl":"t2","t":2}	{"type":"table-create","database":"test","table":"t2","old":{"database":"","table":"","columns":null,"primary-key":null},"def":{"database":"test","table":"t2","columns":[{"type":"string","name":"id"},{"type":"int","name":"a"}],"primary-key":null},"ts":428693224948760587,"sql":"create table t2 (id varchar(32) not null primary key, a int)"}
	{"database":"test","table":"t2","type":"insert","ts":1635334916}

[Rustin170506]

arvo:

      "error": {
        "addr": "127.0.0.1:8300",
        "code": "CDC:ErrProcessorUnknown",
        "message": "AppendRowChangedEvent could not encode to Avro: AvroEventBatchEncoder: get-or-register failed: GetCachedOrRegister: Could not make goavro codec: [CDC:ErrAvroSchemaAPIError]Record \"t2\" fields ought to be non-nil array: \u003cnil\u003e: Record \"t2\" fields ought to be non-nil array: \u003cnil\u003e"
      }