drainer: the sync order between transactions are not guaranteed

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do?

• 1. There are 20 transactions running insert statement from upstream, each transaction will execute 40 insert statements and each statement contains 10 rows. There are add/drop column options at the same time. The amend transaction is on, so the transactions above can both executed successfully.
• 2. When all the DDL options are done, commit the 20 transactions.
• 3. After the all the transaction are committed, insert a record into check point table(INSERT INTO check_point_t_114514 VALUES(1605785449)).
• 4. Waiting for binlog sync all the data into downstream, we take the existence of 1605785449 in the downstream as the evidence the above transactions are all synced.
• 5. Diff the upstream and down stream.

2. What did you expect to see?

The data of upstream and downstream are same in step 5.

3. What did you see instead?

Some records are not synced into downstream yet in step5. However SELECT * FROM t_114514 got 8,000 records from upstream, but only 7989 from downstream.

Here are some logs and simple analysis.

Upstream:

[2020/11/19 11:30:34.416 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=45] [user=root@172.16.4.209] [schemaVersion=808] [txnStartTS=420947016752300087] [forUpdateTS=420947016752300087] [isReadConsistency=false] [current_db=test] [txn_mode=PESSIMISTIC] [sql="INSERT INTO t_114514 VALUES ... (-6653172808726933946, \"flamboyant_noyce1\", 104608717171146891, -1648034004, \"optimistic_yalow0\", \"epic_euclid9\", \"3654-03-05 09:08:18\", \"9684-03-31 08:57:32\", \"elastic_cori\", 1143841086, -1832200570, \"quizzical_jones2\", \"6286-04-20 22:31:45\", \"1977-09-05 23:39:20\", \"ecstatic_goodall1\", \"determined_mclean9\", \"admiring_colden4\", 98, -2406631104964350243, \"1991-01-28 02:34:15\", \"1987-08-11 22:14:50\", -308463298, -113, \"4206-11-19\", \"silly_wozniak9\", -6, \"awesome_almeida6\", \"vibrant_goldwasser5\", \"3240-07-09\", \"sleepy_pike9\", -101, -415560878872419208, 6044286169729966138, 1253881188547516871, -988957040, -958445510818121878, -75, \"pedantic_leavitt9\", \"5750-02-15\", 5867313793019256156, 44, 841757531, -288052318, -918902326800907776, -7206061917876545496, \"2018-05-08 21:51:16\", \"pensive_brattain4\", 3952018417293608054, \"5499-03-23\", \"vigilant_jennings5\", \"4646-12-26 01:33:16\", -15, \"youthful_curran3\", \"7725-04-20 10:49:35\", 1004272094, -56, -3587768614121111799, \"2008-12-23 15:26:38\", 113, \"sleepy_darwin9\", -981600659, \"9025-02-03 03:48:08\", 82, 3338512954396141814, 292123362, -84, 64, \"hopeful_galileo5\", \"hopeful_fermat5\", \"5831-02-17\", 115, \"6030-04-04\", \"8977-07-03 22:06:35\")...)"]
...
[2020/11/19 11:30:49.181 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=45] [user=root@172.16.4.209] [schemaVersion=808] [txnStartTS=420947016752300087] [forUpdateTS=420947016
752300087] [isReadConsistency=false] [current_db=test] [txn_mode=PESSIMISTIC] [sql=COMMIT]
...
[2020/11/19 11:30:49.376 +00:00] [INFO] [2pc.go:1592] ["amend txn successfully for pessimistic commit"] [conn=45] [connID=45] ["txn startTS"=420947016752300087] [memAmended=false] [checkTS=420947020841746448] [startInfoSchemaVer=808] ["table ids"="[247]"] ["action types"="[96]"]
[2020/11/19 11:30:49.427 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=63] [user=root@172.16.4.209] [schemaVersion=996] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=PESSIMISTIC] [sql="INSERT INTO check_point_t_114514 VALUES(1605785449)"]
[2020/11/19 11:30:55.350 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=62] [user=root@172.16.4.209] [schemaVersion=996] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=PESSIMISTIC] [sql="ADMIN CHECK TABLE t_114514"]
[2020/11/19 11:30:55.351 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=63] [user=root@172.16.4.209] [schemaVersion=996] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=PESSIMISTIC] [sql="SHOW INDEXES FROM t_114514"]
[2020/11/19 11:30:55.353 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=62] [user=root@172.16.4.209] [schemaVersion=996] [txnStartTS=0] [forUpdateTS=0] [isReadConsistency=false] [current_db=test] [txn_mode=PESSIMISTIC] [sql="SELECT * FROM t_114514"]

Downstream:

[2020/11/19 11:30:55.296 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=21] [user=root@172.16.4.170] [schemaVersion=998] [txnStartTS=420947022390493194] [forUpdateTS=420947022390493194] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t_114514`(`col_0`,`col_1`,`col_10`,`col_11`,`col_12`,`col_13`,`col_14`,`col_15`,`col_16`,`col_17`,`col_18`,`col_19`,`col_2`,`col_20`,`col_21`,`col_22`,`col_23`,`col_24`,`col_25`,`col_26`,`col_27`,`col_28`,`col_29`,`col_3`,`col_30`,`col_31`,`col_32`,`col_33`,`col_34`,`col_35`,`col_4`,`col_5`,`col_6`,`col_7`,`col_8`,`col_9`,`new_col_0`,`new_col_1`,`new_col_2`,`new_col_3`,`new_col_4`,`new_col_5`,`new_col_6`,`new_col_7`,`new_col_8`,`new_col_9`) VALUES(2546959307161802209,'priceless_mcclintock0',742515745,'tender_colden8','3757-10-02 01:06:25','2018-05-02 01:56:40','ecstatic_curran6','elegant_galileo2','brave_raman3',-53,1368153934125735671,'1989-07-10 21:04:52',-2224887784145052415,'2033-05-12 03:35:08',1340479397,46,'9382-02-26','quirky_bhaskara4',37,'awesome_cori3','keen_einstein3','2721-03-14','elated_yalow0',-924781239,-71,-783947825996253583,14669304578822526,-1216249593400291093,1608868357,7751877400394979278,'pedantic_jennings8','inspiring_poincare1','1660-09-24 19:51:31','4321-04-04 16:08:08','cranky_haibt6',1133957375,'0000-00-00 00:00:00','0000-00-00',NULL,NULL,'0',NULL,NULL,NULL,NULL,'0')"]
[2020/11/19 11:30:55.296 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=20] [user=root@172.16.4.170] [schemaVersion=998] [txnStartTS=420947022390493188] [forUpdateTS=420947022390493188] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`check_point_t_114514`(`id`) VALUES(1605785449)"]
[2020/11/19 11:30:55.296 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=20] [user=root@172.16.4.170] [schemaVersion=998] [txnStartTS=420947022390493188] [forUpdateTS=420947022390493188] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql=COMMIT]
[2020/11/19 11:30:55.297 +00:00] [INFO] [session.go:2191] [GENERAL_LOG] [conn=13] [user=root@172.16.4.170] [schemaVersion=998] [txnStartTS=420947022390493192] [forUpdateTS=420947022390493192] [isReadConsistency=false] [current_db=test] [txn_mode=OPTIMISTIC] [sql="REPLACE INTO `test`.`t_114514`(`col_0`,`col_1`,`col_10`,`col_11`,`col_12`,`col_13`,`col_14`,`col_15`,`col_16`,`col_17`,`col_18`,`col_19`,`col_2`,`col_20`,`col_21`,`col_22`,`col_23`,`col_24`,`col_25`,`col_26`,`col_27`,`col_28`,`col_29`,`col_3`,`col_30`,`col_31`,`col_32`,`col_33`,`col_34`,`col_35`,`col_4`,`col_5`,`col_6`,`col_7`,`col_8`,`col_9`,`new_col_0`,`new_col_1`,`new_col_2`,`new_col_3`,`new_col_4`,`new_col_5`,`new_col_6`,`new_col_7`,`new_col_8`,`new_col_9`) VALUES(-6653172808726933946,'flamboyant_noyce1',-1832200570,'quizzical_jones2','6286-04-20 22:31:45','1977-09-05 23:39:20','ecstatic_goodall1','determined_mclean9','admiring_colden4',98,-2406631104964350243,'1991-01-28 02:34:15',104608717171146891,'1987-08-11 22:14:50',-308463298,-113,'4206-11-19','silly_wozniak9',-6,'awesome_almeida6','vibrant_goldwasser5','3240-07-09','sleepy_pike9',-1648034004,-101,-415560878872419208,6044286169729966138,1253881188547516871,-988957040,-958445510818121878,'optimistic_yalow0','epic_euclid9','3654-03-05 09:08:18','9684-03-31 08:57:32','elastic_cori',1143841086,'0000-00-00 00:00:00','0000-00-00',NULL,NULL,'0',NULL,NULL,NULL,NULL,'0')"]

From the log of downstream, there still some replace-into statements, trace the next replace-into statement after writing check point in downstream. This record is written by first line captured from upstream's log, the transaction's startTs is 420947016752300087, also it's successcully committed before we writing check point in upstream.

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.8-1-g336b519
Git Commit Hash: 336b51969ce563fc46e9c8d422ef71a1c40d5376
Build TS: 2020-11-19 02:38:44
Go Version: go1.15.2
Go OS/Arch: linux/amd64

Start by reproducing the reported drainer scenario with 20 concurrent transactions, concurrent DDL/amend operations, and the checkpoint insert using drainer v4.0.8. Trace whether downstream preserves transaction commit order before the checkpoint; done means the checkpoint is not observed until all earlier committed rows are synchronized and the upstream and downstream counts match.