CDC Stream API deserialize column type like `bigint(20) unsigned` Covert to wrong type

[guixin]

Describe the bug
my table defined some column type like bigint(20) unsigned. I got wrong value by JsonDebeziumDeserializationSchema

Environment :

• Flink version : flink 1.13
• Flink CDC version: 2.2-snapshot
• Database and version: mysql 5.7

To Reproduce
use com.ververica.cdc.connectors.mysql.source.MySqlSource to build source

Additional Description
If applicable, add screenshots to help explain your problem.

this is my test table，and dataset

my map operator get string like this. order_id column and fee column is wrong value

[guixin]

Hi, @leonardBang, i want to fix it

[yaoboxu]

        case BIGINT_UNSIGNED:
            return DataTypes.DECIMAL(20, 0);

IN CDC SOURCE CODE DESCRIBE

[guixin]

add debeziumProperties bigint.unsigned.handling.mode = long can fixed it;

• exmaple

        Properties p = new Properties();
        p.put("bigint.unsigned.handling.mode", "long");

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("127.0.0.1")
                .port(3307)
                .databaseList("")
                .tableList("")
                .username("")
                .password("")
                .serverId("")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .debeziumProperties(p)
                .build();

default bigint.unsigned.handling.mode configure is precise. it will set column type is bytes, so JsonDebeziumDeserializationSchema class deserialize wrong.

[wuchong]

I think it's also fine to introduce bigint.unsigned.handling.mode option for mysql cdc connector.
cc @fsk119

[PatrickRen]

Closing this issue because it was created before version 2.3.0 (2022-11-10). Please try the latest version of Flink CDC to see if the issue has been resolved. If the issue is still valid, kindly report it on Apache Jira under project Flink with component tag Flink CDC. Thank you!