[BUG] streaming mode with TiKV sink fail with int type column

[shiyuhang0]

Describe the bug

• TiBigData： master
• streaming mode, tikv sink

TiKV sink fails to convert Flink row to TiDB row when the table contains any one of the following types column

• int
• bit
• tinyint
• smallint
• mediumint
• year

What did you do

1. create table

CREATE TABLE `test`.`t_src`  (
  `id` int(11) NOT NULL,
  `name` varchar(255) NULL DEFAULT NULL,
  PRIMARY KEY (`id`) 
) ;

CREATE TABLE `test`.`t_dst`  (
  `id` int(11) NOT NULL,
  `name` varchar(255) NULL DEFAULT NULL,
  PRIMARY KEY (`id`) 
) ;

2. Use Kafka source get change log from TiCDC
3. create catalog in sql-client

CREATE CATALOG `tidb` WITH (   'type' = 'tidb',   'tidb.database.url' = 'jdbc:mysql://127.0.0.1:4000?useSSL=false',   'tidb.username' = 'root',   'tidb.password' = '',   'tidb.streaming.source' = 'kafka',   'tidb.streaming.codec' = 'json',   'tidb.streaming.kafka.bootstrap.servers' = 'localhost:9092',   'tidb.streaming.kafka.topic' = 'cdc',   'tidb.streaming.kafka.group.id' = 'test_cdc_group',   'tidb.streaming.ignore-parse-errors' = 'true' ,'tidb.sink.impl'='TIKV')

4. execute SQL in flink sql-client

INSERT INTO `tidb`.`test`.`t_dst` SELECT id,name FROM `tidb`.`test`.`t_src`

6. Use JDBC to insert data into flink_sink

INSERT into test.flink_sink VALUES (1,1)

7. you will find the ClassCastException is thrown

Screenshots