kafka full / incr replication job column type double/double_unsigned/float/float_unsigned：if insert value is int, it's no decimal part in the kafka message return value

[asiroliu]

Description

kafka full / incr replication job column type double/double_unsigned/float/float_unsigned：if insert value is int, it's no decimal part in the kafka message return value

Steps to reproduce the issue

1. create db and table

create database action_db;
use action_db

create table float_columns(id int(11) not null primary key,c_float float(7,4) DEFAULT 111)ENGINE=InnoDB DEFAULT CHARSET=utf8;
create table float_unsigned_columns(id int(11) not null primary key,c_float float(7,4) unsigned DEFAULT 12.23)ENGINE=InnoDB DEFAULT CHARSET=utf8;
create table double_columns(id int(11) not null primary key,c_double double(7,4) DEFAULT 10.13)ENGINE=InnoDB DEFAULT CHARSET=utf8;
create table double_unsigned_columns(id int(11) not null primary key,c_double double(7,4) unsigned DEFAULT 12.22)ENGINE=InnoDB DEFAULT CHARSET=utf8;

2. create kafka job

{
  "Name": "column_types_default_full_columns",
  "Datacenters": [
    "dc1"
  ],
  "Tasks": [
    {
      "Type": "Src",
      "Driver": "MySQL",
      "NodeId": "da1c766d-b945-9ad4-a434-458d637f4420",
      "Config": {
        "ConnectionConfig": {
          "Host": "172.100.9.2",
          "Port": 3306,
          "User": "test_src",
          "Password": "test_src"
        },
        "ReplicateDoDb": [
          {
            "TableSchema": "action_db"
          }
        ]
      }
    },
    {
      "Type": "Dest",
      "Driver": "Kafka",
      "NodeId": "8db2080b-2753-9ce6-a264-45d3383ef35a",
      "Config": {
        "Topic": "dtle",
        "Brokers": [
          "172.100.9.21:9092"
        ],
        "Converter": "json"
      }
    }
  ]
}

3. insert some data

use action_db
insert into float_columns values (1,100);
insert into float_unsigned_columns values (1,100);
insert into double_columns values (1,100);
insert into double_unsigned_columns values (1,100);

4. get message from kafka

/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.100.9.21:9092 --from-beginning --property print.key=true --topic dtle.action_db.float_columns
/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.100.9.21:9092 --from-beginning --property print.key=true --topic dtle.action_db.float_unsigned_columns
/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.100.9.21:9092 --from-beginning --property print.key=true --topic dtle.action_db.double_columns
/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.100.9.21:9092 --from-beginning --property print.key=true --topic dtle.action_db.double_unsigned_columns

Describe the results you received

{
  "payload": {
    "after": {
      "c_double": 100,
      "id": 1
    },

Describe the results you expected

{
  "payload": {
    "after": {
      "c_double": 100.0,
      "id": 1
    },

Output of ./dtle version:**

NOMAD_VERSION=<0.11.1>
DTLE_VERSION=<9.9.9.9-master-f8a8a93>

Additional information

(e.g. issue happens only occasionally)

Additional details (log, config, job config etc):