timestamp incorrect after sync by binlog

[3AceShowHand]

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. deploy and start upstream v5.0.3, enable binlog

2. start pump v6.1.0 manually

3. deploy and start downtream v6.1.0, change the time zone by set global time_zone = "UTC"

4. start drainer v6.1.0.

5. change the downtream time zone by set global time_zone = "Asia/Shanghai"

6. run the sql file

7. What did you expect to see?

select * from test.t should display the same data, but different.

upstream v5.0.3:

downstream v6.1.0:

9. What did you see instead?

upstream and downtream should display as the same.

10. Please provide the relate downstream type and version of drainer. (run drainer -V in terminal to get drainer's version)

v6.1.0

[lichunzhu]

check the error log and find that drainer's generated update values are correct, maybe tidb has some bug which causes this problem. This problem can be reproduced only when drainer tryed to batch dmls execution, that is to say, log at markSuccess length is bigger than 1.

[2022/08/17 17:51:47.514 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.514 +08:00] [DEBUG] [translator.go:120] ["get insert row pk: [KindInt64 2], datums: map[1:KindInt64 2 2:KindInt64 2 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.520 +08:00] [INFO] [load.go:342] ["refresh table info"] [schema=test] [table=t]
[2022/08/17 17:51:47.529 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":1,\"b\":1,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 1)`test`.`t`\"]"]
[2022/08/17 17:51:47.529 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":2,\"b\":2,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 2)`test`.`t`\"]"]
[2022/08/17 17:51:47.530 +08:00] [DEBUG] [executor.go:430] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":1,\"b\":1,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[1,1,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.537 +08:00] [DEBUG] [executor.go:430] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":2,\"b\":2,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[2,2,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.540 +08:00] [DEBUG] [load.go:321] ["markSuccess txns"] ["txns len"=2]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380870631425, commitTS: 435350380870631426, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380883738625, commitTS: 435350380883738626, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380870631425, commitTS: 435350380870631426, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [translator.go:120] ["get insert row pk: [KindInt64 3], datums: map[1:KindInt64 3 2:KindInt64 3 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380883738625, commitTS: 435350380883738626, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.615 +08:00] [DEBUG] [translator.go:120] ["get insert row pk: [KindInt64 4], datums: map[1:KindInt64 4 2:KindInt64 4 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.621 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":3,\"b\":3,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 3)`test`.`t`\"]"]
[2022/08/17 17:51:47.621 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":4,\"b\":4,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 4)`test`.`t`\"]"]
[2022/08/17 17:51:47.623 +08:00] [DEBUG] [executor.go:430] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":3,\"b\":3,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[3,3,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.627 +08:00] [DEBUG] [executor.go:430] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":4,\"b\":4,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[4,4,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.631 +08:00] [DEBUG] [load.go:321] ["markSuccess txns"] ["txns len"=2]
[2022/08/17 17:51:47.715 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380896845825, commitTS: 435350380896845826, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.715 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380909953025, commitTS: 435350380909953026, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380896845825, commitTS: 435350380896845826, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [translator.go:120] ["get insert row pk: [KindInt64 5], datums: map[1:KindInt64 5 2:KindInt64 5 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380909953025, commitTS: 435350380909953026, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [translator.go:327] ["get update oldRow: map[1:KindInt64 1 2:KindInt64 1 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47], newRow: map[1:KindInt64 1 2:KindInt64 10 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.716 +08:00] [DEBUG] [mysql.go:82] ["get update oldVals: [1 1 2022-08-17 09:51:47 2022-08-17 09:51:47], newVals: [1 10 2022-08-17 09:51:47 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.721 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":5,\"b\":5,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 5)`test`.`t`\"]"]
[2022/08/17 17:51:47.721 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":1,\"b\":1,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":1,\"b\":10,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 1)`test`.`t`\",\"(a: 1)`test`.`t`\"]"]
[2022/08/17 17:51:47.723 +08:00] [DEBUG] [executor.go:430] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":1,\"OldValues\":null,\"Values\":{\"a\":5,\"b\":5,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[5,5,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.723 +08:00] [DEBUG] [executor.go:416] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":1,\"b\":1,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":1,\"b\":10,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="DELETE FROM `test`.`t` WHERE `a` = ? LIMIT 1"] [args="[1]"]
[2022/08/17 17:51:47.725 +08:00] [DEBUG] [executor.go:423] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":1,\"b\":1,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":1,\"b\":10,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[1,10,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.728 +08:00] [DEBUG] [load.go:321] ["markSuccess txns"] ["txns len"=2]
[2022/08/17 17:51:47.760 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380923060225, commitTS: 435350380923060226, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380936167426, commitTS: 435350380936167427, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380923060225, commitTS: 435350380923060226, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [translator.go:327] ["get update oldRow: map[1:KindInt64 2 2:KindInt64 2 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47], newRow: map[1:KindInt64 2 2:KindInt64 20 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [mysql.go:82] ["get update oldVals: [2 2 2022-08-17 09:51:47 2022-08-17 09:51:47], newVals: [2 20 2022-08-17 09:51:47 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380936167426, commitTS: 435350380936167427, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [translator.go:327] ["get update oldRow: map[1:KindInt64 3 2:KindInt64 3 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47], newRow: map[1:KindInt64 3 2:KindInt64 30 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.816 +08:00] [DEBUG] [mysql.go:82] ["get update oldVals: [3 3 2022-08-17 09:51:47 2022-08-17 09:51:47], newVals: [3 30 2022-08-17 09:51:47 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.821 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":2,\"b\":2,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":2,\"b\":20,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 2)`test`.`t`\",\"(a: 2)`test`.`t`\"]"]
[2022/08/17 17:51:47.822 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":3,\"b\":3,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":3,\"b\":30,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 3)`test`.`t`\",\"(a: 3)`test`.`t`\"]"]
[2022/08/17 17:51:47.823 +08:00] [DEBUG] [executor.go:416] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":3,\"b\":3,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":3,\"b\":30,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="DELETE FROM `test`.`t` WHERE `a` = ? LIMIT 1"] [args="[3]"]
[2022/08/17 17:51:47.823 +08:00] [DEBUG] [executor.go:416] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":2,\"b\":2,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":2,\"b\":20,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="DELETE FROM `test`.`t` WHERE `a` = ? LIMIT 1"] [args="[2]"]
[2022/08/17 17:51:47.824 +08:00] [DEBUG] [executor.go:423] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":3,\"b\":3,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":3,\"b\":30,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[3,30,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.825 +08:00] [DEBUG] [executor.go:423] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":2,\"b\":2,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":2,\"b\":20,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[2,20,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.828 +08:00] [DEBUG] [load.go:321] ["markSuccess txns"] ["txns len"=2]
[2022/08/17 17:51:47.916 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350380949274625, commitTS: 435350380962381825, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.916 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350380949274625, commitTS: 435350380962381825, node: lingjin-0:8250}"]
[2022/08/17 17:51:47.916 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:47.916 +08:00] [DEBUG] [translator.go:327] ["get update oldRow: map[1:KindInt64 4 2:KindInt64 4 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47], newRow: map[1:KindInt64 4 2:KindInt64 40 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.916 +08:00] [DEBUG] [mysql.go:82] ["get update oldVals: [4 4 2022-08-17 09:51:47 2022-08-17 09:51:47], newVals: [4 40 2022-08-17 09:51:47 2022-08-17 09:51:47]"]
[2022/08/17 17:51:47.922 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":4,\"b\":4,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":4,\"b\":40,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 4)`test`.`t`\",\"(a: 4)`test`.`t`\"]"]
[2022/08/17 17:51:47.923 +08:00] [DEBUG] [executor.go:416] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":4,\"b\":4,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":4,\"b\":40,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="DELETE FROM `test`.`t` WHERE `a` = ? LIMIT 1"] [args="[4]"]
[2022/08/17 17:51:47.925 +08:00] [DEBUG] [executor.go:423] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":4,\"b\":4,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":4,\"b\":40,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[4,40,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:47.927 +08:00] [DEBUG] [load.go:321] ["markSuccess txns"] ["txns len"=1]
[2022/08/17 17:51:48.319 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350381080346625, commitTS: 435350381080346626, node: lingjin-0:8250}"]
[2022/08/17 17:51:48.319 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350381080346625, commitTS: 435350381080346626, node: lingjin-0:8250}"]
[2022/08/17 17:51:48.319 +08:00] [DEBUG] [syncer.go:387] ["get DML"] [SchemaVersion=55]
[2022/08/17 17:51:48.320 +08:00] [DEBUG] [translator.go:327] ["get update oldRow: map[1:KindInt64 5 2:KindInt64 5 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47], newRow: map[1:KindInt64 5 2:KindInt64 50 3:KindMysqlTime 2022-08-17 09:51:47 4:KindMysqlTime 2022-08-17 09:51:47]"]
[2022/08/17 17:51:48.320 +08:00] [DEBUG] [mysql.go:82] ["get update oldVals: [5 5 2022-08-17 09:51:47 2022-08-17 09:51:47], newVals: [5 50 2022-08-17 09:51:47 2022-08-17 09:51:47]"]
[2022/08/17 17:51:48.325 +08:00] [DEBUG] [load.go:536] ["get keys"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":5,\"b\":5,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":5,\"b\":50,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [keys="[\"(a: 5)`test`.`t`\",\"(a: 5)`test`.`t`\"]"]
[2022/08/17 17:51:48.326 +08:00] [DEBUG] [executor.go:416] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":5,\"b\":5,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":5,\"b\":50,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="DELETE FROM `test`.`t` WHERE `a` = ? LIMIT 1"] [args="[5]"]
[2022/08/17 17:51:48.328 +08:00] [DEBUG] [executor.go:423] ["get sql for dml"] [dml="{\"Database\":\"test\",\"Table\":\"t\",\"Tp\":2,\"OldValues\":{\"a\":5,\"b\":5,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"Values\":{\"a\":5,\"b\":50,\"create_at\":\"2022-08-17 09:51:47\",\"update_at\":\"2022-08-17 09:51:47\"},\"UpColumnsInfoMap\":null,\"DestDBType\":1}"] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(?,?,?,?)"] [args="[5,50,\"2022-08-17 09:51:47\",\"2022-08-17 09:51:47\"]"]
[2022/08/17 17:51:48.331 +08:00] [DEBUG] [load.go:321] ["markSuccess txns"] ["txns len"=1]
[2022/08/17 17:51:49.759 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:51:51.758 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:51:53.092 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350382325530626, commitTS: 435350382325530626, node: lingjin-0:8250}"]
[2022/08/17 17:51:53.092 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350382325530626, commitTS: 435350382325530626, node: lingjin-0:8250}"]
[2022/08/17 17:51:53.092 +08:00] [INFO] [syncer.go:278] ["write save point"] [ts=435350382325530626] [version=55]
[2022/08/17 17:51:53.759 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:51:55.758 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:51:56.106 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350383111962625, commitTS: 435350383111962625, node: lingjin-0:8250}"]
[2022/08/17 17:51:56.106 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350383111962625, commitTS: 435350383111962625, node: lingjin-0:8250}"]
[2022/08/17 17:51:57.758 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:51:59.118 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350383898394626, commitTS: 435350383898394626, node: lingjin-0:8250}"]
[2022/08/17 17:51:59.118 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350383898394626, commitTS: 435350383898394626, node: lingjin-0:8250}"]
[2022/08/17 17:51:59.118 +08:00] [INFO] [syncer.go:278] ["write save point"] [ts=435350383898394626] [version=55]
[2022/08/17 17:51:59.758 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:52:01.759 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:52:02.087 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350384684826625, commitTS: 435350384684826625, node: lingjin-0:8250}"]
[2022/08/17 17:52:02.087 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350384684826625, commitTS: 435350384684826625, node: lingjin-0:8250}"]
[2022/08/17 17:52:03.758 +08:00] [DEBUG] [kv.go:547] ["update safeTS failed"] [error="rpc error: code = Unimplemented desc = "] [errorVerbose="rpc error: code = Unimplemented desc = \ngithub.com/tikv/client-go/v2/tikvrpc.CallRPC\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikvrpc/tikvrpc.go:1019\ngithub.com/tikv/client-go/v2/internal/client.(*RPCClient).SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client.go:439\ngithub.com/tikv/client-go/v2/internal/client.interceptedClient.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_interceptor.go:42\ngithub.com/tikv/client-go/v2/internal/client.reqCollapse.SendRequest\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/internal/client/client_collapse.go:74\ngithub.com/tikv/client-go/v2/tikv.(*KVStore).updateSafeTS.func1\n\t/root/go/pkg/mod/github.com/tikv/client-go/v2@v2.0.1-0.20220311043619-5042c6f2aaa6/tikv/kv.go:540\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1571"] [store-id=1]
[2022/08/17 17:52:05.099 +08:00] [DEBUG] [syncer.go:715] ["receive publish binlog item"] [item="{startTS: 435350385471258626, commitTS: 435350385471258626, node: lingjin-0:8250}"]
[2022/08/17 17:52:05.099 +08:00] [DEBUG] [syncer.go:357] ["consume binlog item"] [item="{startTS: 435350385471258626, commitTS: 435350385471258626, node: lingjin-0:8250}"]
[2022/08/17 17:52:05.099 +08:00] [INFO] [syncer.go:278] ["write save point"] [ts=435350385471258626] [version=55]

[3AceShowHand]

3 / 5 shows that the time is 18:54:52, but 1 / 2 / 4 is 10:54:52.

From the TiDB's general log:

3 / 5 is inserted by the conn = 5329260292062839387

consider that 5329260292062839387 is associated with session.time_zone=utf+0. drainer think the tz=utf, it insert 10:54:52 associate with utf-0 to the downtream tidb, since tidb global time_zone is utf-8, it will be convert to 18:54:52, data correct.

[2022/08/17 18:54:16.552 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=165] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=PESSIMISTIC] [sql="SET NAMES utf8mb4"]
[2022/08/17 18:54:16.553 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=165] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=PESSIMISTIC] [sql="SET allow_auto_random_explicit_insert='1',tidb_txn_mode='optimistic',tidb_placement_mode='ignore'"]
[2022/08/17 18:54:16.554 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=165] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="SELECT cast(TIMEDIFF(NOW(6), UTC_TIMESTAMP(6)) as time);"]
[2022/08/17 18:54:53.198 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=165] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="START TRANSACTION"]
[2022/08/17 18:54:53.200 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=165] [txnStartTS=435351373259472897] [forUpdateTS=435351373259472897] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="use `test`;"]
[2022/08/17 18:54:53.201 +08:00] [INFO] [session.go:3274] ["CRUCIAL OPERATION"] [conn=5329260292062839387] [schemaVersion=165] [cur_db=test] [sql="create table t(\n    a int primary key AUTO_INCREMENT, \n    b int,\n    create_at TIMESTAMP not null default current_timestamp,\n    update_at TIMESTAMP not null default current_timestamp\n)"] [user=root@%]
[2022/08/17 18:54:53.201 +08:00] [INFO] [session.go:2560] ["Try to create a new txn inside a transaction auto commit"] [conn=5329260292062839387] [schemaVersion=165] [txnStartTS=435351373259472897] [txnScope=global]
[2022/08/17 18:54:53.296 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql=COMMIT]
[2022/08/17 18:54:53.296 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="SHOW MASTER STATUS"]
[2022/08/17 18:54:53.303 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="SELECT column_name, extra FROM information_schema.columns WHERE table_schema = 'test' AND table_name = 't';"]
[2022/08/17 18:54:53.307 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="SELECT non_unique, index_name, seq_in_index, column_name  FROM information_schema.statistics WHERE table_schema = 'test' AND table_name = 't' ORDER BY seq_in_index ASC;"]
[2022/08/17 18:54:53.311 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="START TRANSACTION"]
[2022/08/17 18:54:53.312 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(3,3,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.318 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(5,5,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.319 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="DELETE FROM `test`.`t` WHERE `a` = 3 LIMIT 1"]
[2022/08/17 18:54:53.320 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(3,30,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.321 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="DELETE FROM `test`.`t` WHERE `a` = 5 LIMIT 1"]
[2022/08/17 18:54:53.322 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(5,50,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.323 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839387] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532353] [forUpdateTS=435351373298532353] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql=COMMIT]

1 / 2 / 4 is inserted by conn=5329260292062839391

consider that 5329260292062839391 is associated with session.time_zone=utf+8. drainer think the tz=utf, it insert 10:54:52 associate with utf-8 to the downtream tidb, since tidb global time_zone is utf-8, it won't be converted, so the data is 10:54:52, which is incorrect.

[2022/08/17 18:54:53.314 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=PESSIMISTIC] [sql="SET NAMES utf8mb4"]
[2022/08/17 18:54:53.315 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=PESSIMISTIC] [sql="SET allow_auto_random_explicit_insert='1',tidb_txn_mode='optimistic',tidb_placement_mode='ignore'"]
[2022/08/17 18:54:53.315 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="START TRANSACTION"]
[2022/08/17 18:54:53.316 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(1,1,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.318 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(2,2,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.320 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(4,4,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.321 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="DELETE FROM `test`.`t` WHERE `a` = 1 LIMIT 1"]
[2022/08/17 18:54:53.322 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(1,10,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.323 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="DELETE FROM `test`.`t` WHERE `a` = 2 LIMIT 1"]
[2022/08/17 18:54:53.324 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(2,20,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.324 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="DELETE FROM `test`.`t` WHERE `a` = 4 LIMIT 1"]
[2022/08/17 18:54:53.325 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t`(`a`,`b`,`create_at`,`update_at`) VALUES(4,40,'2022-08-17 10:54:52','2022-08-17 10:54:52')"]
[2022/08/17 18:54:53.325 +08:00] [INFO] [session.go:3300] [GENERAL_LOG] [conn=5329260292062839391] [user=root@%] [schemaVersion=166] [txnStartTS=435351373298532357] [forUpdateTS=435351373298532357] [isReadConsistency=false] [current_db=] [txn_mode=OPTIMISTIC] [sql=COMMIT]

[3AceShowHand]

The above situation may happen when start a new transaction by https://github.com/pingcap/tidb-binlog/blob/master/pkg/loader/executor.go#L234

[lichunzhu]

https://github.com/pingcap/tidb-binlog/blob/master/pkg/loader/executor.go#L234

It sames that it's a connection pool problem. The root cause is still the undefined behavior. I think we can close this issue now.

Maybe what we can do later is to check downstream database's global.time_zone in an interval, when it diffes we can report it in drainer's log. However, users may ignore the warning in log and the can't be printed if we change time_zones two times in an interval.

So maybe we should add a warning in the official document to warn our users.

[3AceShowHand]

https://github.com/pingcap/tidb-binlog/blob/master/pkg/loader/executor.go#L234

It sames that it's a connection pool problem. The root cause is still the undefined behavior. I think we can close this issue now.

Maybe what we can do later is to check downstream database's global.time_zone in an interval, when it diffes we can report it in drainer's log. However, users may ignore the warning in log and the can't be printed if we change time_zones two times in an interval.

So maybe we should add a warning in the official document to warn our users.

Yes, for the solution:

• For all replication tools, make sure users know that time_zone should be equal between the upstream and the downstream database. This can be achieved by mentioning this in the official document.
• The replication tool should check time_zone before starting replication, else report the error to the user.
• When the task is running, and then the time_zone is changed incautiously, task should stop, and report error to the user, this would be hard to perform since check the time_zone periodically introduce other problems.