kafka incr replication job: file is empty and server id is not correct

[asiroliu]

Description

kafka incr replication job: file is empty and server id is not correct

Steps to reproduce the issue

1. create db and table and check server_id

create database action_db;
use action_db

create table char_columns(id int(11) not null primary key,c_char char(255)) ENGINE=InnoDB DEFAULT CHARSET=utf8;

2. check src mysql server id

mysql> SELECT @@server_id;
+-------------+
| @@server_id |
+-------------+
|        3306 |
+-------------+
1 row in set (0.00 sec)

3. create kafka job

{
  "Name": "column_types_incr_char_columns",
  "Datacenters": [
    "dc1"
  ],
  "Tasks": [
    {
      "Type": "Src",
      "Driver": "MySQL",
      "NodeId": "93f3d165-3bb3-535d-e7dd-70a38df2d85a",
      "Config": {
        "ConnectionConfig": {
          "Host": "172.100.9.1",
          "Port": 3306,
          "User": "test_src",
          "Password": "test_src"
        },
        "ReplicateDoDb": [
          {
            "TableSchema": "action_db",
            "Tables": [
              {
                "TableName": "char_columns"
              }
            ]
          }
        ]
      }
    },
    {
      "Type": "Dest",
      "Driver": "Kafka",
      "NodeId": "08ff253f-ad31-2da8-f505-3a5418cd1a94",
      "Config": {
        "Topic": "dtle",
        "Brokers": [
          "172.100.9.21:9092"
        ],
        "Converter": "json"
      }
    }
  ]
}

4. insert some data

use action_db
insert into char_columns values (0,NULL);
insert into char_columns values (1,'');
insert into char_columns values (2,' ');
insert into char_columns values (3,'a a a ');
insert into char_columns values (4,'abcdefghijklmnopqrstuvwxyz');
insert into char_columns values (5,'ABCDEFGHIJKLMNOPQRSTUVWXYZ');

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.char_columns

Describe the results you received

{
  "payload": {
    "after": {
      "c_char": null,
      "id": 0
    },
    "before": null,
    "op": "c",
    "source": {
      "db": "action_db",
      "file": "",
      "gtid": "863b5aa5-8dd8-11eb-beae-0242ac640901:4",
      "name": "dtle",
      "pos": 1004,
      "query": null,
      "row": 0,
      "server_id": 1,
      "table": "char_columns",
    },
  },
...
}

Describe the results you expected

the payload "file" should be not null. the payload "server_id" should be same as src mysql server id(3306)

{
  "payload": {
    "after": {
      "c_char": null,
      "id": 0
    },
    "before": null,
    "op": "c",
    "source": {
      "db": "action_db",
      "file": "mysql-bin.000001",
      "gtid": "863b5aa5-8dd8-11eb-beae-0242ac640901:4",
      "name": "dtle",
      "pos": 1004,
      "query": null,
      "row": 0,
      "server_id": 3306,
      "table": "char_columns",
    },
  },

Output of ./dtle version:**

NOMAD_VERSION=<0.11.1>
DTLE_VERSION=<3.21.03.0-3.21.03.x-9a4d6ec>

Additional information

(e.g. issue happens only occasionally)

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

[asiroliu]

fixed file is empty.

[asiroliu]

从binlog中可以获取server id信息