kafka full replication job: kafka message file and pos is empty

[asiroliu]

Description

kafka full replication job: kafka message file and pos is empty

Steps to reproduce the issue

1. create db and table, then insert some data

create database action_db_1;
use action_db_1

create table aggregation_source(id int unsigned AUTO_INCREMENT,name VARCHAR(100) NOT NULL, PRIMARY KEY(id))ENGINE=InnoDB DEFAULT CHARSET=utf8;

insert into aggregation_source (name) values("aggregation1");
insert into aggregation_source (name) values("aggregation2");
insert into aggregation_source (name) values("aggregation3");
insert into aggregation_source (name) values("aggregation4");
insert into aggregation_source (name) values("aggregation5");

2. create kafka job

{
  "Name": "kafka_aggregation_source_one",
  "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_1",
            "Tables": [
              {
                "TableName": "aggregation_source"
              }
            ]
          }
        ]
      }
    },
    {
      "Type": "Dest",
      "Driver": "Kafka",
      "NodeId": "8db2080b-2753-9ce6-a264-45d3383ef35a",
      "Config": {
        "Topic": "dtle",
        "Brokers": [
          "172.100.9.21:9092"
        ],
        "Converter": "json"
      }
    }
  ]
}

3. 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_1.aggregation_source

Describe the results you received

{
  "payload": {
    "after": {
      "id": 1,
      "name": "aggregation1"
    },
    "before": null,
    "op": "c",
    "source": {
      "db": "action_db_1",
      "file": "",
      "gtid": null,
      "name": "dtle",
      "pos": 0,
      "query": null,
      "row": 0,
      "server_id": 0,
      "snapshot": true,
      "table": "aggregation_source",
      "thread": null,
      "ts_sec": 0,
      "version": "0.0.1"
    },
    "ts_ms": 1602745869900
...
  },
...
}

Describe the results you expected

the file and pos should not empty such as follow value

{
  "payload": {
    "source": {
      "db": "action_db_1",
      "file": "mysql-bin.000003",
      "gtid": null,
      "name": "dtle",
      "pos": 4797,
      "query": null,
      "row": 0,
      "server_id": 0,
      "snapshot": true,
      "table": "aggregation_source",
      "thread": null,
      "ts_sec": 0,
      "version": "0.0.1"
    },
    "ts_ms": 1602745873915
...
  },
...
}

Output of ./dtle version:**

NOMAD_VERSION=<0.11.1>
DTLE_VERSION=<9.9.9.9-master-077b064>

Additional information

(e.g. issue happens only occasionally)

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