mysql-kafka job: missing default value in message when default value is empty string

[asiroliu]

Description

mysql-kafka job: missing default value in message when default value is empty string

Steps to reproduce the issue

1. src MySQL insert data

shell> sysbench /usr/share/sysbench/oltp_common.lua --mysql-host=172.100.9.3 --mysql-port=3306 --mysql-user=test --mysql-password=test --create_secondary=off --report-interval=10 --time=0 --mysql-db=action_db --tables=2 --table_size=10 prepare

2. check create table sql

mysql> show create table sbtest1\G
*************************** 1. row ***************************
       Table: sbtest1
Create Table: CREATE TABLE `sbtest1` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `k` int(11) NOT NULL DEFAULT '0',
  `c` char(120) NOT NULL DEFAULT '',
  `pad` char(60) NOT NULL DEFAULT '',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=latin1
1 row in set (0.00 sec)

3. create mysql-kafka job

{
  "job_id": "kafka_table_level",
  "is_password_encrypted": true,
  "task_step_name": "all",
  "failover": true,
  "retry": 2,
  "src_task": {
    "task_name": "src",
    "node_id": "f872114a-b55c-6b29-9df4-de604fb362f5",
    "mysql_src_task_config": {
      "gtid": "",
      "binlog_relay": false
    },
    "drop_table_if_exists": true,
    "skip_create_db_table": false,
    "repl_chan_buffer_size": 120,
    "chunk_size": 2000,
    "group_max_size": 1,
    "group_timeout": 100,
    "connection_config": {
      "database_type": "MySQL",
      "host": "172.100.9.3",
      "port": 3306,
      "user": "test_src",
      "password": "test_src"
    },
    "replicate_do_db": [
      {
        "table_schema": "action_db",
        "tables": [
          {
            "table_name": "sbtest1"
          }
        ]
      }
    ]
  },
  "dest_task": {
    "task_name": "dest",
    "node_id": "47cf4c1a-49f5-7693-ac17-ea65c7b6a2df",
    "parallel_workers": 1,
    "kafka_topic": "dtle",
    "kafka_broker_addrs": [
      "172.100.9.21:9092"
    ]
  }

4. compare dtle and dbz full kafka messages

5. src MySQL insert data

shell> sysbench /usr/share/sysbench/oltp_write_only.lua --mysql-host=172.100.9.3 --mysql-port=3306 --mysql-user=test --mysql-password=test --create_secondary=off --report-interval=10 --time=0 --mysql-db=action_db --tables=2 --table_size=10 --events=20 run

6. compare dtle and dbz incr kafka messages

Output of ./dtle version:**

9.9.9.9-master-d24d240

mysql> create table a.a939 (id int primary key auto_increment,
    ->   val1 varchar(50) not null default 'aaa',
    ->   val2 varchar(50) not null default '',
    ->   val3 varchar(50) not null);
Query OK, 0 rows affected (0.04 sec)

mysql> show columns from a.a939;
+-------+-------------+------+-----+---------+----------------+
| Field | Type        | Null | Key | Default | Extra          |
+-------+-------------+------+-----+---------+----------------+
| id    | int(11)     | NO   | PRI | NULL    | auto_increment |
| val1  | varchar(50) | NO   |     | aaa     |                |
| val2  | varchar(50) | NO   |     |         |                |
| val3  | varchar(50) | NO   |     | NULL    |                |
+-------+-------------+------+-----+---------+----------------+
4 rows in set (0.02 sec)

• GetTableColumns()
  • 无法区别default=null和default=''
• GetTableColumnsSqle()
  • 可以区别default=null和default=''

在各类情况(全量增量)中具体用了哪一个解析尚不明.

kafkaColumnListToColDefs()会把default=''设成nil

func kafkaColumnListToColDefs(...) (valColDefs ColDefs, keyColDefs ColDefs) {
	cols := colList.ColumnList()
	for i, _ := range cols {
		var field *Schema
		defaultValue := cols[i].Default
		if defaultValue == "" { // **HERE**
			defaultValue = nil
		}
		optional := cols[i].Nullable

try to fix in d865a73

[asiroliu]

version:
9.9.9.9-master-458c807