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

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/errors@v0.11.5-0.20190809092503-95897b64e011/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/errors@v0.11.5-0.20190809092503-95897b64e011/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/errors@v0.11.5-0.20190809092503-95897b64e011/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/errors@v0.11.5-0.20190809092503-95897b64e011/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/log@v0.0.0-20200117041106-d28c14d3b1cd/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: 09a8cfda8a305416f11d56bfcc939210eeea804e

Start with drainer/sync/mysql.go and the loader paths in pkg/loader/load.go referenced by the logs, then trace checkpoint handling in drainer/checkpoint/mysql.go and drainer/syncer.go. Reproduce the invalid-connection case and determine how reconnect, already-applied DDL, and DML safe-mode behavior should be verified. Done means the drainer handles the three requested cases without losing consistency.