MySQL-kafka full / incr stage: data type datetime - Some data is inconsistent with dbz

[asiroliu]

Description

MySQL-kafka full / incr stage: data type datetime - Some data is inconsistent with dbz

Steps to reproduce the issue

1. inser data on src MySQL

mysql> CREATE DATABASE IF NOT EXISTS `action_db`;
mysql> create table datetime_columns(id int(11) not null primary key,c_datetime '
 "datetime DEFAULT '2019-03-02 04:06:09')ENGINE=InnoDB DEFAULT CHARSET=utf8;
mysql> insert into datetime_columns values (0,'1000-01-01 00:00:00');
mysql> insert into datetime_columns values (1,'9999-12-31 23:59:59');
mysql> insert into datetime_columns values (2,'1000-01-01 00:00:00.000000');
mysql> insert into datetime_columns values (3,'1900-03-02 04:06:09');
mysql> insert into datetime_columns values (4,now());
mysql> insert into datetime_columns values (5,CURRENT_TIMESTAMP);
mysql> insert into datetime_columns values (6,NULL);
mysql> insert into datetime_columns values (7,default);

2. create MySQL-kafka job

{
  "job_id": "column_types_default_full_datetime_columns",
  "is_password_encrypted": false,
  "task_step_name": "all",
  "failover": true,
  "retry": 2,
  "src_task": {
    "task_name": "src",
    "node_id": "0117aa99-17f9-2d2d-3427-4e7338fde7a9",
    "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": "datetime_columns"
          }
        ]
      }
    ]
  },
  "dest_task": {
    "task_name": "dest",
    "node_id": "3908d648-0f51-c45c-9f7f-e12abf914b69",
    "parallel_workers": 1,
    "kafka_topic": "dtle",
    "kafka_broker_addrs": [
      "172.100.9.21:9092"
    ]
  }
}

3. create dbz job

{
  "name": "column_types_default_full_datetime_columns",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "172.100.9.3",
    "database.port": 3306,
    "database.user": "test",
    "database.password": "test",
    "database.server.id": "1",
    "database.server.name": "dbz",
    "database.history.kafka.bootstrap.servers": "172.100.9.21:9092",
    "database.history.kafka.topic": "dbz.history",
    "database.serverTimezone": "UTC",
    "table.whitelist": "action_db.datetime_columns"
  }
}

4. get kafka message and compare
   insert into datetime_columns values (1,'9999-12-31 23:59:59');

insert into datetime_columns values (3,'1900-03-02 04:06:09');

Output of ./dtle version:**

9.9.9.9-master-a269c29

[asiroliu]

verified version:

9.9.9.9-master-3cca6a1

与dbz 0.8对比不一致，与dbz 1.9对比一致。