BUG#33405699: After secondary node is killed, it can not rejoined

It was detected that when a group was running with
`group_replication_consistency = AFTER`, when a secondary failed due
to external conditions (like unstable network), when it rejoined
sometimes the secondary could face the error
  Plugin group_replication reported: 'Transaction
  'GTID' does not exist on Group Replication consistency manager
  while receiving remote transaction prepare.'

The root cause of this issue is that the primary may log the
View_change_log_event on which the secondary rejoined out of order,
which when the secondary uses the primary as the group donor may
cause the secondary to improperly catch-up with the group and
eventually generate incorrect GTIDs for the group transactions.

The primary on Group Replication does ensure that the
View_change_log_event is logged after all precedent transactions,
but there is a window on which transactions ordered after the
View_change_log_event on the group global order can be logged before
the View_change_log_event.

In order to solve this issue, the primary must ensure:
 1) transactions ordered before a view are always logged into binary
    log before the View_change_log_event;
 2) transactions ordered after a view are always logged into binary
    log after the View_change_log_event;

That guarantee is now provided by the Binlog Ticket Manager, which
ensures a given order on which transactions on the Binlog Group
Commit do commit.
Components on the server, in this case Group Replication, do assign
a commit ticket order to each transaction, the binlog ordered commit
does follow that order. Example:
 * The ticket manager is committing ticket 5, all transactions will
   be assigned to ticket 5, meaning that there is no guarantee in
   the commit order among them.
 * When a new member joins, the ticket is incremented to 6 and
   assigned to the view. The ticket is incremented again, so the new
   transactions will be assigned to ticket 7.
 * This will ensure that the order between previous transactions,
   view and future transactions is respected.
   This means that transactions with ticket 7 cannot be committed
   until all previous transactions are committed.

   ------------------------->
   | T1  T2 |  V  |  T4  T3 |
   +--------+-----+---------+
   |  5   5 |  6  |   7   7 | tickets
   +--------+-----+---------+

Change-Id: Ifa881894a65626ae8102761f23c6ddfe205e80bc

Author: nacarvalho

include/mysql/group_replication_priv.h

@@ -29,12 +29,14 @@
 
 #include "my_sys.h"
 #include "my_thread.h"
+#include "sql/binlog/group_commit/bgc_ticket_manager.h"
 #include "sql/binlog_ostream.h"
 #include "sql/binlog_reader.h"
 #include "sql/debug_sync.h"
 #include "sql/log_event.h"
 #include "sql/replication.h"
 #include "sql/rpl_channel_service_interface.h"
+#include "sql/rpl_commit_stage_manager.h"
 #include "sql/rpl_gtid.h"
 #include "sql/rpl_write_set_handler.h"
 

libbinlogevents/include/control_events.h

@@ -1042,6 +1042,18 @@ class Gtid_event : public Binary_log_event {
   void print_event_info(std::ostream &) override {}
   void print_long_info(std::ostream &) override {}
 #endif
+  /*
+    Commit group ticket consists of: 1st bit, used internally for
+    synchronization purposes ("is in use"),  followed by 63 bits for
+    the ticket value.
+  */
+  static constexpr int COMMIT_GROUP_TICKET_LENGTH = 8;
+  /*
+    Default value of commit_group_ticket, which means it is not
+    being used.
+  */
+  static constexpr std::uint64_t kGroupTicketUnset = 0;
+
  protected:
   static const int ENCODED_FLAG_LENGTH = 1;
   static const int ENCODED_SID_LENGTH = 16;  // Uuid::BYTE_LENGTH;
@@ -1115,9 +1127,10 @@ class Gtid_event : public Binary_log_event {
     On the originating master, the event has only one timestamp as the two
     timestamps are equal. On every other server we have two timestamps.
   */
-  static const int MAX_DATA_LENGTH = FULL_COMMIT_TIMESTAMP_LENGTH +
-                                     TRANSACTION_LENGTH_MAX_LENGTH +
-                                     FULL_SERVER_VERSION_LENGTH;
+  static const int MAX_DATA_LENGTH =
+      FULL_COMMIT_TIMESTAMP_LENGTH + TRANSACTION_LENGTH_MAX_LENGTH +
+      FULL_SERVER_VERSION_LENGTH +
+      COMMIT_GROUP_TICKET_LENGTH; /* 64-bit unsigned integer */
 
   static const int MAX_EVENT_LENGTH =
       LOG_EVENT_HEADER_LEN + POST_HEADER_LENGTH + MAX_DATA_LENGTH;
@@ -1137,6 +1150,28 @@ class Gtid_event : public Binary_log_event {
   uint32_t original_server_version;
   /** The version of the immediate server */
   uint32_t immediate_server_version;
+
+  /** Ticket number used to group sessions together during the BGC. */
+  std::uint64_t commit_group_ticket{kGroupTicketUnset};
+
+  /**
+    Returns the length of the packed `commit_group_ticket` field. It may be
+    8 bytes or 0 bytes, depending on whether or not the value is
+    instantiated.
+
+    @return The length of the packed `commit_group_ticket` field
+  */
+  int get_commit_group_ticket_length() const;
+
+  /**
+   Set the commit_group_ticket and update the transaction length if
+   needed, that is, if the commit_group_ticket was not set already
+   account it on the transaction size.
+
+   @param value The commit_group_ticket value.
+  */
+  void set_commit_group_ticket_and_update_transaction_length(
+      std::uint64_t value);
 };
 
 /**

libbinlogevents/src/control_events.cpp

@@ -448,6 +448,8 @@ Gtid_event::Gtid_event(const char *buf, const Format_description_event *fde)
     +------------+
     |   4/8 bytes| original/immediate_server_version (see timestamps*)
     +------------+
+    |     8 bytes| Commit group ticket
+    +------------+
 
     The 'Flags' field contains gtid flags.
 
@@ -558,6 +560,10 @@ Gtid_event::Gtid_event(const char *buf, const Format_description_event *fde)
                            ORIGINAL_SERVER_VERSION_LENGTH);
           } else
             original_server_version = immediate_server_version;
+
+          if (READER_CALL(can_read, COMMIT_GROUP_TICKET_LENGTH)) {
+            READER_TRY_SET(this->commit_group_ticket, read<uint64_t>);
+          }
         }
       }
     }
@@ -567,6 +573,26 @@ Gtid_event::Gtid_event(const char *buf, const Format_description_event *fde)
   BAPI_VOID_RETURN;
 }
 
+int Gtid_event::get_commit_group_ticket_length() const {
+  if (kGroupTicketUnset != commit_group_ticket) {
+    return COMMIT_GROUP_TICKET_LENGTH;
+  }
+  return 0;
+}
+
+void Gtid_event::set_commit_group_ticket_and_update_transaction_length(
+    std::uint64_t value) {
+  /*
+    Add the commit_group_ticket length to the transaction length if
+    it was not yet considered.
+  */
+  assert(value > 0);
+  set_trx_length(transaction_length + (kGroupTicketUnset == commit_group_ticket
+                                           ? COMMIT_GROUP_TICKET_LENGTH
+                                           : 0));
+  commit_group_ticket = value;
+}
+
 Previous_gtids_event::Previous_gtids_event(const char *buffer,
                                            const Format_description_event *fde)
     : Binary_log_event(&buffer, fde) {

mysql-test/include/execute_from_conditional_timestamp_sync_point.inc

@@ -0,0 +1,169 @@
+# ==== Purpose ====
+#
+# Resume a statement that was previously executed up until a conditional
+# timestamp sync point.
+#
+# ==== About conditional sync points for timestamp ====
+#
+# Conditional timestamp sync points are used to control the execution flow
+# of a statement for which the session variable `TIMESTAMP` as been set to
+# a given value, meaning, the given named sync point will only be activated
+# when the transaction being executed has a given session `TIMESTAMP` value
+# associated.
+#
+# Note that the usage of the `TIMESTAMP` session variable doens't imply
+# that the conditional execution is dependent on time measuring or the
+# measured value of any clock. The `TIMESTAMP` variable is used for mere
+# convenience and due to low impact on server behavior.
+#
+# Usage example for executing a statement and triggering a conditional sync
+# point for timestamp on a single server:
+#
+# - On the server code, setup the conditional timestamp using
+#   `CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP`, as an example, just before the
+#   transaction is committed in the storage engines:
+#
+#     bool trx_coordinator::commit_in_engines(THD *thd, bool all,
+#                                             bool run_after_commit) {
+#       if (all) CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP("before_commit_in_engines");
+#       ...
+#
+# - On the MTR test, include this script, passing the connection used to
+#   execute the statement and the connection used to activate the sync
+#   point.
+#
+#     --connect(con1, localhost, root,,)
+#     --connection con1
+#
+#     --let $timestamp = 100
+#     --let $auxiliary_connection = default
+#     --let $statement_connection = con1
+#     --let $statement = INSERT INTO t VALUES (100);
+#     --let $sync_point = before_commit_in_engines
+#     --source include/execute_to_conditional_timestamp_sync_point.inc
+#     ...
+#
+# Usage example for executing a statement on the source server and
+# triggering a conditional sync point for timestamp on a replica:
+#
+# - On the server code, setup the conditional timestamp using
+#   `CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP`, as an example, just before the
+#   transaction is committed in the storage engines:
+#
+#     bool trx_coordinator::commit_in_engines(THD *thd, bool all,
+#                                             bool run_after_commit) {
+#       if (all) CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP("before_commit_in_engines");
+#       ...
+#
+# - On the MTR test, include this script, passing the connection used to
+#   execute the statement and the connection used to activate the sync
+#   point.
+#
+#     --let $timestamp = 100
+#     --let $auxiliary_connection = slave
+#     --let $statement_connection = master
+#     --let $statement = INSERT INTO t VALUES (100);
+#     --let $sync_point = before_commit_in_engines
+#     --source include/execute_to_conditional_timestamp_sync_point.inc
+#     ...
+#
+# ==== Usage ====
+#
+# --let $statement_connection = {connection name}
+# --let $statement = {statement to execute}
+# --let $sync_point = {name of the sync point}
+# [--let $timestamp = {timestamp to set}]
+# [--let $auxiliary_connection = {connection name}]
+# [--let $reap_statement_connection = 0|1]
+# --source include/execute_from_conditional_timestamp_sync_point.inc
+#
+#   $statement_connection
+#     The connection on which $statement is to be executed.
+#     This must be different from $auxiliary_connection.
+#
+#   $statement
+#     The statement that will be paused on the sync point.
+#
+#   $sync_point
+#     The name of the sync point (not quoted).
+#
+#   $timestamp
+#     By default, the current unix timestamp, it's the value used in `SET
+#     TIMESTAMP = ...` in order to be used with the conditional timestamp
+#     sync point.
+#
+#   $auxiliary_connection
+#     The connection used to set the debug sync points and
+#     symbols. $auxiliary_connection is set to the currenct active
+#     connection, if no value is passed. It must be different from the
+#     $statement connection.
+#
+#   $reap_statement_connection
+#     If the statement was executed using `--send`, reap the
+#     $statement_connection.
+#
+# Side effects:
+#   If $auxiliary_connection and $statement_connection target the same
+#   server, the $statement is executed using `--send` and without invoking
+#   `--reap`. Default value is 0.
+#
+
+if ($sync_point == '') {
+  --die ERROR IN TEST: You must set $sync_point before sourcing execute_to_conditional_timestamp_sync_point.inc
+}
+
+if ($timestamp  == '') {
+  --let $timestamp = 1
+}
+
+--let $include_filename= execute_from_conditional_timestamp_sync_point.inc [$timestamp@$sync_point]
+--source include/begin_include_file.inc
+
+if ($statement_connection == '') {
+  --die ERROR IN TEST: You must set $statement_connection before sourcing execute_to_conditional_timestamp_sync_point.inc
+}
+
+--let $_eftsp_auxiliary_connection = $CURRENT_CONNECTION
+if ($auxiliary_connection != '') {
+  --let $_eftsp_auxiliary_connection = $auxiliary_connection
+}
+
+if ($statement_connection == $_eftsp_auxiliary_connection) {
+  --echo statement_connection=$statement_connection
+  --echo auxiliary_connection=$_eftsp_auxiliary_connection
+  --die ERROR IN TEST: You must set $statement_connection to something else than $auxiliary_connection
+}
+
+--let $underscore = _
+
+--connection $_eftsp_auxiliary_connection
+if ($to_sync_point != '') {
+  --let $debug_point_silent = 1
+  --let $debug_point = syncpoint_$to_sync_point$underscore$timestamp
+  --source include/add_debug_point.inc
+}
+
+--disable_query_log
+--eval SET @@SESSION.DEBUG_SYNC = "now SIGNAL continue_$sync_point$underscore$timestamp"
+--enable_query_log
+
+if ($to_sync_point != '') {
+  --connection $_eftsp_auxiliary_connection
+  --disable_query_log
+  --eval SET @@SESSION.DEBUG_SYNC = "now WAIT_FOR reached_$to_sync_point$underscore$timestamp"
+  --enable_query_log
+
+  --source include/remove_debug_point.inc
+  --let $debug_point_silent = 0
+}
+
+if ($reap_statement_connection == 1) {
+  --connection $statement_connection
+  if ($_ectsp_server_aux == $_ectsp_server_stmt) {
+    --reap
+  }
+}
+
+--let $skip_restore_connection= 1
+--let $include_filename= execute_from_conditional_timestamp_sync_point.inc
+--source include/end_include_file.inc

mysql-test/suite/binlog/r/binlog_bgc_ticket_manager_stress.result

@@ -0,0 +1,15 @@
+RESET MASTER;
+CREATE TABLE t (
+c1 INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+c2 INT,
+c3 DATETIME DEFAULT CURRENT_TIMESTAMP);
+# Configuration:
+# · MYSQL_SLAP theads: 100
+# · MYSQL_SLAP iterations: 200
+# · BGC tickets: 100
+# · Expected table count: 20100
+# · Sleep: 0.201
+# Adding debug point 'begin_new_bgc_ticket' to @@SESSION.debug
+# Removing debug point 'begin_new_bgc_ticket' from @@SESSION.debug
+DROP TABLE t;
+RESET MASTER;

mysql-test/suite/binlog/t/binlog_bgc_ticket_manager_stress.test

@@ -0,0 +1,57 @@
+--source include/big_test.inc
+--source include/have_debug.inc
+--source include/have_binlog_format_row.inc
+
+RESET MASTER;
+CREATE TABLE t (
+  c1 INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+  c2 INT,
+  c3 DATETIME DEFAULT CURRENT_TIMESTAMP);
+
+--let $time_unit = 0.00001
+--let $threads = 100
+--let $iterations = 200
+--let $n_tickets = 100
+--expr $records = $threads * $iterations
+--expr $records = $records + $n_tickets
+--expr $sleep = $records * $time_unit
+
+--echo # Configuration:
+--echo # · MYSQL_SLAP theads: $threads
+--echo # · MYSQL_SLAP iterations: $iterations
+--echo # · BGC tickets: $n_tickets
+--echo # · Expected table count: $records
+--echo # · Sleep: $sleep
+
+exec_in_background $MYSQL_SLAP
+  --silent
+  --delimiter=";"
+  --create-schema="test"
+  -uroot
+  --password=""
+  --query="INSERT INTO t (c2) VALUES (0)"
+  --concurrency=$threads
+  --iterations=$iterations
+  --commit=1;
+
+--let $debug_point = begin_new_bgc_ticket
+--let $debug_type = SESSION
+--source include/add_debug_point.inc
+
+--let $new_ticket = 0
+while ($new_ticket != $n_tickets) {
+  --disable_query_log
+  --eval INSERT INTO t (c2) VALUES ($new_ticket + 1)
+  --enable_query_log
+  --inc $new_ticket
+  --sleep $sleep
+}
+
+--source include/remove_debug_point.inc
+
+--let $wait_condition= SELECT COUNT(1) = $records FROM t
+--let $wait_timeout= 3000
+--source include/wait_condition.inc
+
+DROP TABLE t;
+RESET MASTER;

mysql-test/suite/binlog_gtid/r/binlog_gtid_bgc_ticket_manager.result

@@ -0,0 +1,19 @@
+include/only_with_option.inc [GLOBAL.binlog_order_commits = 1]
+CREATE TABLE t (
+c1 INT NOT NULL PRIMARY KEY,
+c2 INT
+);
+include/save_binlog_position.inc
+include/execute_to_conditional_timestamp_sync_point.inc [before_assign_session_to_bgc_ticket]
+SET gtid_next = "435aeb3e-cc91-11ec-a715-6ba04eb9487e:1"; INSERT INTO t VALUES (1, 1);
+include/execute_to_conditional_timestamp_sync_point.inc [before_wait_on_ticket]
+SET gtid_next = "435aeb3e-cc91-11ec-a715-6ba04eb9487e:2"; INSERT INTO t VALUES (2, 2);
+include/execute_to_conditional_timestamp_sync_point.inc [before_wait_on_ticket]
+SET gtid_next = "435aeb3e-cc91-11ec-a715-6ba04eb9487e:3"; INSERT INTO t VALUES (3, 3);
+include/execute_from_conditional_timestamp_sync_point.inc [1@before_assign_session_to_bgc_ticket]
+include/execute_from_conditional_timestamp_sync_point.inc [1@inside_wait_on_ticket]
+include/execute_from_conditional_timestamp_sync_point.inc [2@before_wait_on_ticket]
+include/execute_from_conditional_timestamp_sync_point.inc [3@before_wait_on_ticket]
+include/assert_binlog_events.inc [Gtid/.*GTID_NEXT= '435aeb3e-cc91-11ec-a715-6ba04eb9487e:2' # !Begin # !Insert # !Commit # Gtid/.*GTID_NEXT= '435aeb3e-cc91-11ec-a715-6ba04eb9487e:3' # !Begin # !Insert # !Commit # Gtid/.*GTID_NEXT= '435aeb3e-cc91-11ec-a715-6ba04eb9487e:1' # !Begin # !Insert # !Commit]
+DROP TABLE t;
+RESET MASTER;