After an invalid connection error occurs, drainer's processing logic isn't working as expected

[zhenjiaogao]

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do?
   When doing DDL operations, network connection errors will appear in drainer's log and stderr log as follow:
   (1) drainer's log

[2020/09/05 17:42:06.316 +08:00] [ERROR] [load.go:414] ["Rollback failed"] [sql="ALTER TABLE `table_test` CHANGE `content` `content` text NULL COMMENT '操作内容'"] [error="invalid connection"]
[2020/09/05 17:42:08.381 +08:00] [INFO] [main.go:63] ["got signal to exit."] [signal=interrupt]
[2020/09/05 17:42:08.381 +08:00] [INFO] [server.go:451] ["begin to close drainer server"]
[2020/09/05 17:42:08.387 +08:00] [INFO] [server.go:416] ["has already update status"] [id=192.x.x.128:8249]
[2020/09/05 17:42:08.387 +08:00] [INFO] [server.go:455] ["commit status done"]
[2020/09/05 17:42:08.387 +08:00] [INFO] [pump.go:77] ["pump is closing"] [id=192.x.x.135:8250]
[2020/09/05 17:42:08.387 +08:00] [INFO] [pump.go:77] ["pump is closing"] [id=192.x.x.134:8250]
[2020/09/05 17:42:08.387 +08:00] [INFO] [util.go:72] [Exit] [name=heartbeat]
[2020/09/05 17:42:08.387 +08:00] [INFO] [collector.go:135] ["publishBinlogs quit"]
[2020/09/05 17:42:08.387 +08:00] [INFO] [util.go:72] [Exit] [name=collect]
[2020/09/05 17:42:08.387 +08:00] [INFO] [merge.go:245] ["Merger is closed successfully"]
[2020/09/05 17:43:07.318 +08:00] [ERROR] [load.go:414] ["Rollback failed"] [sql="ALTER TABLE `table_test` CHANGE `content` `content` text NULL COMMENT '操作内容'"] [error="invalid connection"]
[2020/09/05 17:44:08.319 +08:00] [ERROR] [load.go:414] ["Rollback failed"] [sql="ALTER TABLE `table_test` CHANGE `content` `content` text NULL COMMENT '操作内容'"] [error="invalid connection"]
[2020/09/05 17:44:11.321 +08:00] [ERROR] [load.go:776] ["exec failed"] [sql="ALTER TABLE `table_test` CHANGE `content` `content` text NULL COMMENT '操作内容'"] [metadata="commit ts: 419236934156288071"] [error="invalid connection"] [errorVerbose="invalid connection\ngithub.com/pingcap/errors.AddStack\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/pkg/mod/github.com/pingcap/[email protected]/errors.go:174\ngithub.com/pingcap/errors.Trace\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/pkg/mod/github.com/pingcap/[email protected]/juju_adaptor.go:15\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*loaderImpl).execDDL\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:431\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*batchManager).execDDL\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:748\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*batchManager).put\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:770\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*loaderImpl).Run\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:602\ngithub.com/pingcap/tidb-binlog/drainer/sync.(*MysqlSyncer).run\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/sync/mysql.go:237\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1357"]
[2020/09/05 17:44:11.321 +08:00] [INFO] [load.go:876] ["txnManager has been closed"]
[2020/09/05 17:44:11.321 +08:00] [INFO] [load.go:566] ["{16 20 0xc000dfe310 0xc000e100a0 false 1 true true false}"]
[2020/09/05 17:44:11.321 +08:00] [INFO] [load.go:567] ["Run()... in Loader quit"]
[2020/09/05 17:44:11.321 +08:00] [INFO] [mysql.go:233] ["Successes chan quit"]
[2020/09/05 17:44:11.321 +08:00] [INFO] [load.go:820] ["run()... in txnManager quit"]
[2020/09/05 17:44:11.321 +08:00] [INFO] [syncer.go:257] ["write save point"] [ts=419236934116966409]
[2020/09/05 17:44:11.321 +08:00] [ERROR] [syncer.go:457] ["Failed to close syncer"] [error="invalid connection"] [errorVerbose="invalid connection\ngithub.com/pingcap/errors.AddStack\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/pkg/mod/github.com/pingcap/[email protected]/errors.go:174\ngithub.com/pingcap/errors.Trace\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/pkg/mod/github.com/pingcap/[email protected]/juju_adaptor.go:15\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*loaderImpl).execDDL\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:431\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*batchManager).execDDL\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:748\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*batchManager).put\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:770\ngithub.com/pingcap/tidb-binlog/pkg/loader.(*loaderImpl).Run\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/pkg/loader/load.go:602\ngithub.com/pingcap/tidb-binlog/drainer/sync.(*MysqlSyncer).run\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/sync/mysql.go:237\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1357"]
[2020/09/05 17:44:11.322 +08:00] [FATAL] [syncer.go:260] ["save checkpoint failed"] [ts=419236934116966409] [error="query sql failed: replace into tidb_binlog.checkpoint values(6763676702800952729, '{\"consistent\":false,\"commitTS\":419236934116966409,\"ts-map\":{}}'): invalid connection"] [errorVerbose="invalid connection\nquery sql failed: replace into tidb_binlog.checkpoint values(6763676702800952729, '{\"consistent\":false,\"commitTS\":419236934116966409,\"ts-map\":{}}')\ngithub.com/pingcap/tidb-binlog/drainer/checkpoint.(*MysqlCheckPoint).Save\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/checkpoint/mysql.go:153\ngithub.com/pingcap/tidb-binlog/drainer.(*Syncer).savePoint\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/syncer.go:258\ngithub.com/pingcap/tidb-binlog/drainer.(*Syncer).handleSuccess\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/syncer.go:232\ngithub.com/pingcap/tidb-binlog/drainer.(*Syncer).run.func1\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/syncer.go:278\nruntime.goexit\n\t/usr/local/go/src/runtime/asm_amd64.s:1357"] [stack="github.com/pingcap/log.Fatal\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/pkg/mod/github.com/pingcap/[email protected]/global.go:59\ngithub.com/pingcap/tidb-binlog/drainer.(*Syncer).savePoint\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/syncer.go:260\ngithub.com/pingcap/tidb-binlog/drainer.(*Syncer).handleSuccess\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/syncer.go:232\ngithub.com/pingcap/tidb-binlog/drainer.(*Syncer).run.func1\n\t/home/jenkins/agent/workspace/_binlog_multi_branch_release-4.0/go/src/github.com/pingcap/tidb-binlog/drainer/syncer.go:278"]

(2) drainer's sdterr log

[mysql] 2020/09/05 17:44:08 packets.go:36: read tcp 192.x.x.128:35456->192.x.x.106:3306: i/o timeout
[mysql] 2020/09/05 17:44:09 packets.go:36: unexpected EOF
[mysql] 2020/09/05 17:44:10 packets.go:36: unexpected EOF
[mysql] 2020/09/05 17:44:11 packets.go:36: unexpected EOF

2. What did you expect to see?
   After an invalid connection error occurs, it is recommended that drainer uses the following processing logic:
   (1) Reconnect after invalid connection error occurs
   (2) If the DDL operation has been executed in downstream such as mysql or tidb, then skip this DDL
   (3) If it is a DML operation, enable safe-mode to ensure data consistency

3. What did you see instead?

4. Please provide the relate downstream type and version of drainer.
   (run drainer -V in terminal to get drainer's version)
   Release Version: v4.0.3-1-g09a8cfd
   Git Commit Hash: 09a8cfd