1 files changed, 163 insertions, 126 deletions

diff --git a/sql/slave.cc b/sql/slave.cc
index 749aad5d683..34324a3f2a2 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2020, MariaDB Corporation
+   Copyright (c) 2009, 2020, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -53,6 +53,9 @@
                                                 // Create_file_log_event,
                                                 // Format_description_log_event
 #include "wsrep_mysqld.h"
+#ifdef WITH_WSREP
+#include "wsrep_trans_observer.h"
+#endif
 
 #ifdef HAVE_REPLICATION
 
@@ -360,6 +363,29 @@ end:
   return err;
 }
 
+static THD *new_bg_THD()
+{
+  THD *thd= new THD(next_thread_id());
+  thd->thread_stack= (char*) &thd;
+  thd->store_globals();
+  thd->system_thread = SYSTEM_THREAD_SLAVE_BACKGROUND;
+  thd->security_ctx->skip_grants();
+  thd->set_command(COM_DAEMON);
+  thd->variables.wsrep_on= 0;
+  return thd;
+}
+
+static void bg_gtid_delete_pending(void *)
+{
+  THD *thd= new_bg_THD();
+
+  rpl_slave_state::list_element *list;
+  list= rpl_global_gtid_slave_state->gtid_grab_pending_delete_list();
+  rpl_global_gtid_slave_state->gtid_delete_pending(thd, &list);
+  if (list)
+    rpl_global_gtid_slave_state->put_back_list(list);
+  delete thd;
+}
 
 static void bg_gtid_pos_auto_create(void *hton)
 {
@@ -394,8 +420,8 @@ static void bg_gtid_pos_auto_create(void *hton)
 
   /* Find the entry for the table to auto-create. */
   mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
-  entry= (rpl_slave_state::gtid_pos_table *)
-    rpl_global_gtid_slave_state->gtid_pos_tables;
+  entry= rpl_global_gtid_slave_state->
+         gtid_pos_tables.load(std::memory_order_relaxed);
   while (entry)
   {
     if (entry->table_hton == hton &&
@@ -419,13 +445,7 @@ static void bg_gtid_pos_auto_create(void *hton)
   table_name.str= loc_table_name.c_ptr_safe();
   table_name.length= loc_table_name.length();
 
-  thd= new THD(next_thread_id());
-  thd->thread_stack= (char*) &thd;           /* Set approximate stack start */
-  thd->system_thread = SYSTEM_THREAD_SLAVE_BACKGROUND;
-  thd->store_globals();
-  thd->security_ctx->skip_grants();
-  thd->set_command(COM_DAEMON);
-  thd->variables.wsrep_on= 0;
+  thd= new_bg_THD();
   err= gtid_pos_table_creation(thd, engine, &table_name);
   if (err)
   {
@@ -438,8 +458,8 @@ static void bg_gtid_pos_auto_create(void *hton)
 
   /* Now enable the entry for the auto-created table. */
   mysql_mutex_lock(&rpl_global_gtid_slave_state->LOCK_slave_state);
-  entry= (rpl_slave_state::gtid_pos_table *)
-    rpl_global_gtid_slave_state->gtid_pos_tables;
+  entry= rpl_global_gtid_slave_state->
+         gtid_pos_tables.load(std::memory_order_relaxed);
   while (entry)
   {
     if (entry->table_hton == hton &&
@@ -458,19 +478,11 @@ end:
     plugin_unlock(NULL, engine);
 }
 
-
 static bool slave_background_thread_gtid_loaded;
 
 static void bg_rpl_load_gtid_slave_state(void *)
 {
-  THD *thd= new THD(next_thread_id());
-  thd->thread_stack= (char*) &thd;           /* Set approximate stack start */
-  thd->system_thread = SYSTEM_THREAD_SLAVE_BACKGROUND;
-  thd->store_globals();
-  thd->security_ctx->skip_grants();
-  thd->set_command(COM_DAEMON);
-  thd->variables.wsrep_on= 0;
-
+  THD *thd= new_bg_THD();
   thd_proc_info(thd, "Loading slave GTID position from table");
   if (rpl_load_gtid_slave_state(thd))
     sql_print_warning("Failed to load slave replication state from table "
@@ -510,8 +522,7 @@ void slave_background_kill_request(THD *to_kill)
   to ensure that the table_entry will not go away before we can lock the
   LOCK_slave_state.
 */
-void
-slave_background_gtid_pos_create_request(
+void slave_background_gtid_pos_create_request(
         rpl_slave_state::gtid_pos_table *table_entry)
 {
   if (table_entry->state != rpl_slave_state::GTID_POS_AUTO_CREATE)
@@ -529,6 +540,16 @@ slave_background_gtid_pos_create_request(
 }
 
 
+/*
+  Request the manager thread to delete no longer used rows from the
+  mysql.gtid_slave_pos* tables.
+*/
+void slave_background_gtid_pending_delete_request(void)
+{
+  mysql_manager_submit(bg_gtid_delete_pending, NULL);
+}
+
+
 /* Initialize slave structures */
 
 int init_slave()
@@ -833,6 +854,8 @@ static void make_slave_transaction_retry_errors_printable(void)
 }
 
 
+#define DEFAULT_SLAVE_RETRY_ERRORS 9
+
 bool init_slave_transaction_retry_errors(const char* arg)
 {
   const char *p;
@@ -844,7 +867,7 @@ bool init_slave_transaction_retry_errors(const char* arg)
   if (!arg)
     arg= "";
 
-  slave_transaction_retry_error_length= 2;
+  slave_transaction_retry_error_length= DEFAULT_SLAVE_RETRY_ERRORS;
   for (;my_isspace(system_charset_info,*arg);++arg)
     /* empty */;
   for (p= arg; *p; )
@@ -867,15 +890,22 @@ bool init_slave_transaction_retry_errors(const char* arg)
     currently, InnoDB deadlock detected by InnoDB or lock
     wait timeout (innodb_lock_wait_timeout exceeded
   */
-  slave_transaction_retry_errors[0]= ER_LOCK_DEADLOCK;
-  slave_transaction_retry_errors[1]= ER_LOCK_WAIT_TIMEOUT;
+  slave_transaction_retry_errors[0]= ER_NET_READ_ERROR;
+  slave_transaction_retry_errors[1]= ER_NET_READ_INTERRUPTED;
+  slave_transaction_retry_errors[2]= ER_NET_ERROR_ON_WRITE;
+  slave_transaction_retry_errors[3]= ER_NET_WRITE_INTERRUPTED;
+  slave_transaction_retry_errors[4]= ER_LOCK_WAIT_TIMEOUT;
+  slave_transaction_retry_errors[5]= ER_LOCK_DEADLOCK;
+  slave_transaction_retry_errors[6]= ER_CONNECT_TO_FOREIGN_DATA_SOURCE;
+  slave_transaction_retry_errors[7]= 2013; /* CR_SERVER_LOST */
+  slave_transaction_retry_errors[8]= 12701; /* ER_SPIDER_REMOTE_SERVER_GONE_AWAY_NUM */
 
   /* Add user codes after this */
-  for (p= arg, i= 2; *p; )
+  for (p= arg, i= DEFAULT_SLAVE_RETRY_ERRORS; *p; )
   {
     if (!(p= str2int(p, 10, 0, LONG_MAX, &err_code)))
       break;
-    if (err_code > 0 && err_code < ER_ERROR_LAST)
+    if (err_code > 0)
       slave_transaction_retry_errors[i++]= (uint) err_code;
     while (!my_isdigit(system_charset_info,*p) && *p)
       p++;
@@ -1040,6 +1070,7 @@ terminate_slave_thread(THD *thd,
     DBUG_PRINT("loop", ("killing slave thread"));
 
     mysql_mutex_lock(&thd->LOCK_thd_kill);
+    mysql_mutex_lock(&thd->LOCK_thd_data);
 #ifndef DONT_USE_THR_ALARM
     /*
       Error codes from pthread_kill are:
@@ -1052,6 +1083,7 @@ terminate_slave_thread(THD *thd,
     thd->awake_no_mutex(NOT_KILLED);
 
     mysql_mutex_unlock(&thd->LOCK_thd_kill);
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
 
     /*
       There is a small chance that slave thread might miss the first
@@ -1304,7 +1336,7 @@ static bool io_slave_killed(Master_info* mi)
   DBUG_ENTER("io_slave_killed");
 
   DBUG_ASSERT(mi->slave_running); // tracking buffer overrun
-  DBUG_RETURN(mi->abort_slave || abort_loop || mi->io_thd->killed);
+  DBUG_RETURN(mi->abort_slave || mi->io_thd->killed);
 }
 
 /**
@@ -1329,7 +1361,7 @@ static bool sql_slave_killed(rpl_group_info *rgi)
 
   DBUG_ASSERT(rli->sql_driver_thd == thd);
   DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun
-  if (abort_loop || rli->sql_driver_thd->killed || rli->abort_slave)
+  if (rli->sql_driver_thd->killed || rli->abort_slave)
   {
     /*
       The transaction should always be binlogged if OPTION_KEEP_LOG is
@@ -2555,10 +2587,7 @@ static bool wait_for_relay_log_space(Relay_log_info* rli)
                   &rli->log_space_lock,
                   &stage_waiting_for_relay_log_space,
                   &old_stage);
-  while (rli->log_space_limit <
-         (ulonglong)my_atomic_load64_explicit((volatile int64*)
-                                              (&rli->log_space_total),
-                                              MY_MEMORY_ORDER_RELAXED) &&
+  while (rli->log_space_limit < rli->log_space_total &&
          !(slave_killed=io_slave_killed(mi)) &&
          !rli->ignore_log_space_limit)
     mysql_cond_wait(&rli->log_space_cond, &rli->log_space_lock);
@@ -2599,7 +2628,7 @@ static bool wait_for_relay_log_space(Relay_log_info* rli)
       DBUG_PRINT("info", ("log_space_limit=%llu log_space_total=%llu "
                           "ignore_log_space_limit=%d "
                           "sql_force_rotate_relay=%d", 
-                        rli->log_space_limit, rli->log_space_total,
+                        rli->log_space_limit, uint64(rli->log_space_total),
                         (int) rli->ignore_log_space_limit,
                         (int) rli->sql_force_rotate_relay));
     }
@@ -3121,10 +3150,7 @@ static bool send_show_master_info_data(THD *thd, Master_info *mi, bool full,
     protocol->store(mi->rli.last_error().message, &my_charset_bin);
     protocol->store((uint32) mi->rli.slave_skip_counter);
     protocol->store((ulonglong) mi->rli.group_master_log_pos);
-    protocol->store((ulonglong)
-                     my_atomic_load64_explicit((volatile int64*)
-                                               (&mi->rli.log_space_total),
-                                               MY_MEMORY_ORDER_RELAXED));
+    protocol->store((ulonglong) mi->rli.log_space_total);
 
     protocol->store(
       mi->rli.until_condition==Relay_log_info::UNTIL_NONE ? "None":
@@ -3251,22 +3277,15 @@ static bool send_show_master_info_data(THD *thd, Master_info *mi, bool full,
     // Slave_SQL_Running_State
     protocol->store(slave_sql_running_state, &my_charset_bin);
 
-    uint64 events;
-    events= (uint64)my_atomic_load64_explicit((volatile int64 *)
-              &mi->total_ddl_groups, MY_MEMORY_ORDER_RELAXED);
-    protocol->store(events);
-    events= (uint64)my_atomic_load64_explicit((volatile int64 *)
-              &mi->total_non_trans_groups, MY_MEMORY_ORDER_RELAXED);
-    protocol->store(events);
-    events= (uint64)my_atomic_load64_explicit((volatile int64 *)
-              &mi->total_trans_groups, MY_MEMORY_ORDER_RELAXED);
-    protocol->store(events);
+    protocol->store(mi->total_ddl_groups);
+    protocol->store(mi->total_non_trans_groups);
+    protocol->store(mi->total_trans_groups);
 
     if (full)
     {
       protocol->store((uint32)    mi->rli.retried_trans);
       protocol->store((ulonglong) mi->rli.max_relay_log_size);
-      protocol->store((uint32)    mi->rli.executed_entries);
+      protocol->store(mi->rli.executed_entries);
       protocol->store((uint32)    mi->received_heartbeats);
       protocol->store((double)    mi->heartbeat_period, 3, &tmp);
       protocol->store(gtid_pos->ptr(), gtid_pos->length(), &my_charset_bin);
@@ -3418,7 +3437,6 @@ static int init_slave_thread(THD* thd, Master_info *mi,
 
   thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
     SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO;
-  thread_safe_increment32(&service_thread_count);
 
   /* We must call store_globals() before doing my_net_init() */
   if (init_thr_lock() || thd->store_globals() ||
@@ -3789,32 +3807,26 @@ apply_event_and_update_pos_apply(Log_event* ev, THD* thd, rpl_group_info *rgi,
     exec_res= ev->apply_event(rgi);
 
 #ifdef WITH_WSREP
-  if (exec_res)
-  {
-    switch (thd->wsrep_conflict_state) {
-    case NO_CONFLICT: break;
-    case MUST_REPLAY:
-      WSREP_DEBUG("SQL apply failed for MUST_REPLAY, res %d", exec_res);
+  if (WSREP(thd)) {
+
+    if (exec_res) {
       mysql_mutex_lock(&thd->LOCK_thd_data);
-      wsrep_replay_transaction(thd);
-      switch (thd->wsrep_conflict_state) {
-      case NO_CONFLICT:
-        exec_res = 0; /* replaying succeeded, and slave may continue */
+      switch(thd->wsrep_trx().state()) {
+      case wsrep::transaction::s_must_replay:
+        /* this transaction will be replayed,
+           so not raising slave error here */
+        WSREP_DEBUG("SQL apply failed for MUST_REPLAY, res %d", exec_res);
+	exec_res = 0;
         break;
-      case ABORTED: break; /* replaying has failed, trx is rolled back */
       default:
-	WSREP_WARN("unexpected result of slave transaction replaying: %lld, %d",
-		   thd->thread_id, thd->wsrep_conflict_state);
+          WSREP_DEBUG("SQL apply failed, res %d conflict state: %s",
+                      exec_res, wsrep_thd_transaction_state_str(thd));
+          rli->abort_slave= 1;
+          rli->report(ERROR_LEVEL, ER_UNKNOWN_COM_ERROR, rgi->gtid_info(),
+                      "Node has dropped from cluster");
+          break;
       }
       mysql_mutex_unlock(&thd->LOCK_thd_data);
-      break;
-    default:
-      WSREP_DEBUG("SQL apply failed, res %d conflict state: %d",
-                  exec_res, thd->wsrep_conflict_state);
-      rli->abort_slave= 1;
-      rli->report(ERROR_LEVEL, ER_UNKNOWN_COM_ERROR, rgi->gtid_info(),
-                  "Node has dropped from cluster");
-      break;
     }
   }
 #endif
@@ -4109,6 +4121,15 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
   }
   if (ev)
   {
+#ifdef WITH_WSREP
+    if (wsrep_before_statement(thd))
+    {
+      mysql_mutex_unlock(&rli->data_lock);
+      delete ev;
+      WSREP_INFO("Wsrep before statement error");
+      DBUG_RETURN(1);
+    }
+#endif /* WITH_WSREP */
     int exec_res;
     Log_event_type typ= ev->get_type_code();
 
@@ -4149,6 +4170,9 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
       rli->abort_slave= 1;
       rli->stop_for_until= true;
       mysql_mutex_unlock(&rli->data_lock);
+#ifdef WITH_WSREP
+      wsrep_after_statement(thd);
+#endif /* WITH_WSREP */
       delete ev;
       DBUG_RETURN(1);
     }
@@ -4186,7 +4210,12 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
       if (res == 0)
         rli->event_relay_log_pos= rli->future_event_relay_log_pos;
       if (res >= 0)
+      {
+#ifdef WITH_WSREP
+	wsrep_after_statement(thd);
+#endif /* WITH_WSREP */
         DBUG_RETURN(res);
+      }
       /*
         Else we proceed to execute the event non-parallel.
         This is the case for pre-10.0 events without GTID, and for handling
@@ -4221,6 +4250,9 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
                         "aborted because of out-of-memory error");
         mysql_mutex_unlock(&rli->data_lock);
         delete ev;
+#ifdef WITH_WSREP
+	  wsrep_after_statement(thd);
+#endif /* WITH_WSREP */
         DBUG_RETURN(1);
       }
 
@@ -4235,6 +4267,9 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
                           "thread aborted because of out-of-memory error");
           mysql_mutex_unlock(&rli->data_lock);
           delete ev;
+#ifdef WITH_WSREP
+          wsrep_after_statement(thd);
+#endif /* WITH_WSREP */
           DBUG_RETURN(1);
         }
         /*
@@ -4263,13 +4298,17 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
       retry.
     */
     if (unlikely(exec_res == 2))
+    {
+#ifdef WITH_WSREP
+      wsrep_after_statement(thd);
+#endif /* WITH_WSREP */
       DBUG_RETURN(1);
-
+    }
 #ifdef WITH_WSREP
     mysql_mutex_lock(&thd->LOCK_thd_data);
-    if (thd->wsrep_conflict_state == NO_CONFLICT)
-    {
-      mysql_mutex_unlock(&thd->LOCK_thd_data);
+    enum wsrep::client_error wsrep_error= thd->wsrep_cs().current_error();
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+    if (wsrep_error == wsrep::e_success)
 #endif /* WITH_WSREP */
     if (slave_trans_retries)
     {
@@ -4282,8 +4321,8 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
           We were in a transaction which has been rolled back because of a
           temporary error;
           let's seek back to BEGIN log event and retry it all again.
-	  Note, if lock wait timeout (innodb_lock_wait_timeout exceeded)
-	  there is no rollback since 5.0.13 (ref: manual).
+          Note, if lock wait timeout (innodb_lock_wait_timeout exceeded)
+          there is no rollback since 5.0.13 (ref: manual).
           We have to not only seek but also
 
           a) init_master_info(), to seek back to hot relay log's start
@@ -4344,13 +4383,11 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli,
                             serial_rgi->trans_retries));
       }
     }
+
+    rli->executed_entries++;
 #ifdef WITH_WSREP
-    }
-    else
-      mysql_mutex_unlock(&thd->LOCK_thd_data);
+    wsrep_after_statement(thd);
 #endif /* WITH_WSREP */
-
-    thread_safe_increment64(&rli->executed_entries);
     DBUG_EXECUTE_IF(
         "pause_sql_thread_on_fde",
         if (ev && typ == FORMAT_DESCRIPTION_EVENT) {
@@ -4529,7 +4566,7 @@ pthread_handler_t handle_slave_io(void *arg)
     goto err_during_init;
   }
   thd->system_thread_info.rpl_io_info= &io_info;
-  add_to_active_threads(thd);
+  server_threads.insert(thd);
   mi->slave_running = MYSQL_SLAVE_RUN_NOT_CONNECT;
   mi->abort_slave = 0;
   mysql_mutex_unlock(&mi->run_lock);
@@ -4822,6 +4859,7 @@ Stopping slave I/O thread due to out-of-memory error from master");
           not cause the slave IO thread to stop, and the error messages are
           already reported.
         */
+        DBUG_EXECUTE_IF("simulate_delay_semisync_slave_reply", my_sleep(800000););
         (void)repl_semisync_slave.slave_reply(mi);
       }
 
@@ -4857,15 +4895,13 @@ Stopping slave I/O thread due to out-of-memory error from master");
       {
         DBUG_PRINT("info", ("log_space_limit=%llu log_space_total=%llu "
                             "ignore_log_space_limit=%d",
-                            rli->log_space_limit, rli->log_space_total,
+                            rli->log_space_limit, uint64(rli->log_space_total),
                             (int) rli->ignore_log_space_limit));
       }
 #endif
 
       if (rli->log_space_limit && rli->log_space_limit <
-          (ulonglong) my_atomic_load64_explicit((volatile int64*)
-                                                (&rli->log_space_total),
-                                                MY_MEMORY_ORDER_RELAXED) &&
+          rli->log_space_total &&
           !rli->ignore_log_space_limit)
         if (wait_for_relay_log_space(rli))
         {
@@ -4919,7 +4955,7 @@ err:
     flush_master_info(mi, TRUE, TRUE);
   THD_STAGE_INFO(thd, stage_waiting_for_slave_mutex_on_exit);
   thd->add_status_to_global();
-  unlink_not_visible_thd(thd);
+  server_threads.erase(thd);
   mysql_mutex_lock(&mi->run_lock);
 
 err_during_init:
@@ -4931,8 +4967,6 @@ err_during_init:
 
   thd->assert_not_linked();
   delete thd;
-  thread_safe_decrement32(&service_thread_count);
-  signal_thd_deleted();
 
   mi->abort_slave= 0;
   mi->slave_running= MYSQL_SLAVE_NOT_RUN;
@@ -5207,7 +5241,7 @@ pthread_handler_t handle_slave_sql(void *arg)
   /* Ensure that slave can exeute any alter table it gets from master */
   thd->variables.alter_algorithm= (ulong) Alter_info::ALTER_TABLE_ALGORITHM_DEFAULT;
 
-  add_to_active_threads(thd);
+  server_threads.insert(thd);
   /*
     We are going to set slave_running to 1. Assuming slave I/O thread is
     alive and connected, this is going to make Seconds_Behind_Master be 0
@@ -5294,13 +5328,6 @@ pthread_handler_t handle_slave_sql(void *arg)
   }
 #endif
 
-#ifdef WITH_WSREP
-  thd->wsrep_exec_mode= LOCAL_STATE;
-  wsrep_thd_set_query_state(thd, QUERY_EXEC);
-  /* synchronize with wsrep replication */
-  if (WSREP_ON)
-    wsrep_ready_wait();
-#endif
   DBUG_PRINT("master_info",("log_file_name: %s  position: %llu",
                             rli->group_master_log_name,
                             rli->group_master_log_pos));
@@ -5401,7 +5428,14 @@ pthread_handler_t handle_slave_sql(void *arg)
     goto err;
   }
   mysql_mutex_unlock(&rli->data_lock);
-
+#ifdef WITH_WSREP
+  wsrep_open(thd);
+  if (wsrep_before_command(thd))
+  {
+    WSREP_WARN("Slave SQL wsrep_before_command() failed");
+    goto err;
+  }
+#endif /* WITH_WSREP */
   /* Read queries from the IO/THREAD until this thread is killed */
 
   thd->set_command(COM_SLAVE_SQL);
@@ -5438,10 +5472,16 @@ pthread_handler_t handle_slave_sql(void *arg)
     if (exec_relay_log_event(thd, rli, serial_rgi))
     {
 #ifdef WITH_WSREP
-      if (thd->wsrep_conflict_state != NO_CONFLICT)
+      if (WSREP(thd))
       {
-        wsrep_node_dropped= TRUE;
-        rli->abort_slave= TRUE;
+        mysql_mutex_lock(&thd->LOCK_thd_data);
+
+        if (thd->wsrep_cs().current_error())
+        {
+          wsrep_node_dropped = TRUE;
+          rli->abort_slave   = TRUE;
+        }
+        mysql_mutex_unlock(&thd->LOCK_thd_data);
       }
 #endif /* WITH_WSREP */
 
@@ -5494,6 +5534,10 @@ pthread_handler_t handle_slave_sql(void *arg)
                           rli->group_master_log_pos, tmp.c_ptr_safe());
     sql_print_information("master was %s:%d", mi->host, mi->port);
   }
+#ifdef WITH_WSREP
+  wsrep_after_command_before_result(thd);
+  wsrep_after_command_after_result(thd);
+#endif /* WITH_WSREP */
 
  err_before_start:
 
@@ -5567,7 +5611,7 @@ pthread_handler_t handle_slave_sql(void *arg)
   }
   THD_STAGE_INFO(thd, stage_waiting_for_slave_mutex_on_exit);
   thd->add_status_to_global();
-  unlink_not_visible_thd(thd);
+  server_threads.erase(thd);
   mysql_mutex_lock(&rli->run_lock);
 
 err_during_init:
@@ -5612,17 +5656,17 @@ err_during_init:
                  "SQL slave will continue");
       wsrep_node_dropped= FALSE;
       mysql_mutex_unlock(&rli->run_lock);
-      WSREP_DEBUG("wsrep_conflict_state now: %d", thd->wsrep_conflict_state);
-      WSREP_INFO("slave restart: %d", thd->wsrep_conflict_state);
-      thd->wsrep_conflict_state= NO_CONFLICT;
       goto wsrep_restart_point;
-    } else {
+    }
+    else
+    {
       WSREP_INFO("Slave error due to node going non-primary");
       WSREP_INFO("wsrep_restart_slave was set and therefore slave will be "
-                 "automatically restarted when node joins back to cluster.");
+                 "automatically restarted when node joins back to cluster");
       wsrep_restart_slave_activated= TRUE;
     }
   }
+  wsrep_close(thd);
 #endif /* WITH_WSREP */
 
  /*
@@ -5639,8 +5683,6 @@ err_during_init:
 
   delete serial_rgi;
   delete thd;
-  thread_safe_decrement32(&service_thread_count);
-  signal_thd_deleted();
 
   DBUG_LEAVE;                                   // Must match DBUG_ENTER()
   my_thread_end();
@@ -7693,10 +7735,7 @@ static Log_event* next_event(rpl_group_info *rgi, ulonglong *event_size)
              is are able to rotate and purge sometime soon.
          */
         if (rli->log_space_limit && 
-            rli->log_space_limit <
-            (ulonglong) my_atomic_load64_explicit((volatile int64*)
-                                                  (&rli->log_space_total),
-                                                  MY_MEMORY_ORDER_RELAXED))
+            rli->log_space_limit < rli->log_space_total)
         {
           /* force rotation if not in an unfinished group */
           rli->sql_force_rotate_relay= !rli->is_in_group();
@@ -7931,7 +7970,7 @@ enum Log_event_type wsrep_peak_event(rpl_group_info *rgi, ulonglong* event_size)
     ev_type= (ev) ? ev->get_type_code() : UNKNOWN_EVENT;
     delete ev;
     future_pos+= *event_size;
-  } while (ev_type == ANNOTATE_ROWS_EVENT);
+  } while (ev_type == ANNOTATE_ROWS_EVENT || ev_type == XID_EVENT);
 
   /* scan the log back and re-set the positions to original values */
   rgi->rli->event_relay_log_pos= event_pos;
@@ -8015,8 +8054,8 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
 {
   struct st_version_range_for_one_bug {
     uint        bug_id;
-    const uchar introduced_in[3]; // first version with bug
-    const uchar fixed_in[3];      // first version with fix
+    Version introduced_in; // first version with bug
+    Version fixed_in;      // first version with fix
   };
   static struct st_version_range_for_one_bug versions_for_all_bugs[]=
   {
@@ -8026,19 +8065,17 @@ bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
     {33029, { 5, 1,  0 }, { 5, 1, 12 } },
     {37426, { 5, 1,  0 }, { 5, 1, 26 } },
   };
-  const uchar *master_ver=
-    rli->relay_log.description_event_for_exec->server_version_split.ver;
-
-  DBUG_ASSERT(sizeof(rli->relay_log.description_event_for_exec->server_version_split.ver) == 3);
+  const Version &master_ver=
+    rli->relay_log.description_event_for_exec->server_version_split;
 
   for (uint i= 0;
        i < sizeof(versions_for_all_bugs)/sizeof(*versions_for_all_bugs);i++)
   {
-    const uchar *introduced_in= versions_for_all_bugs[i].introduced_in,
-      *fixed_in= versions_for_all_bugs[i].fixed_in;
+    const Version &introduced_in= versions_for_all_bugs[i].introduced_in;
+    const Version &fixed_in= versions_for_all_bugs[i].fixed_in;
     if ((versions_for_all_bugs[i].bug_id == bug_id) &&
-        (memcmp(introduced_in, master_ver, 3) <= 0) &&
-        (memcmp(fixed_in,      master_ver, 3) >  0) &&
+        introduced_in <= master_ver &&
+        fixed_in > master_ver &&
         (pred == NULL || (*pred)(param)))
     {
       if (!report)