BUG#37613510: Ever growing GR Transactions Rows Validating after secondary joins the group

Group Replication start operation checks if there are partial
transactions on the `group_replication_applier` channel from a
previous group participation. If partial transactions are found,
`group_replication_applier` channel is stopped after applying all
complete transactions, its relay logs purged and then the channel is
restarted. After this step, distributed recovery kicks-in and
applies the missing data from a group member.
The Group Replication pipeline operation to stop the
`group_replication_applier` channel was incorrectly stopping the
periodic task from the certifier module, which was causing that some
periodic internal operations were not taking place.
One of the tasks that was not happening was the periodic send of the
committed transactions, which omission was preventing the
certification info garbage collection, which on its turn was causing
the continuous increase of the column
COUNT_TRANSACTIONS_ROWS_VALIDATING of the table
performance_schema.replication_group_member_stats.

To solve the above issue, the pipeline operation to stop the
`group_replication_applier` channel now does not interfere with the
certifier module.

Change-Id: I1a4c2f7a5b6d0ca65caf43eeae38103a17b2d5ec

Author: nacarvalho

mysql-test/suite/group_replication/r/gr_partial_trx_in_applier_relay_log.result

@@ -40,6 +40,7 @@ INSERT INTO t1 values (12);
 SET GLOBAL DEBUG="-d,stop_applier_channel_after_reading_write_rows_log_event";
 include/stop_group_replication.inc
 include/start_group_replication.inc
+include/assert.inc [Certifier broadcast thread must be running]
 ########################################################################
 # 5. On M1: Insert another tuple and do a diff tables with other nodes.
 #           (just to check that everything is working fine).

mysql-test/suite/group_replication/t/gr_partial_trx_in_applier_relay_log.test

@@ -93,6 +93,10 @@ SET GLOBAL DEBUG="-d,stop_applier_channel_after_reading_write_rows_log_event";
 --let $wait_timeout=120
 --source include/start_group_replication.inc
 
+--let $assert_text= Certifier broadcast thread must be running
+--let $assert_cond= [SELECT COUNT(*) FROM performance_schema.threads WHERE name = "thread/group_rpl/THD_certifier_broadcast"] = 1
+--source include/assert.inc
+
 --echo ########################################################################
 --echo # 5. On M1: Insert another tuple and do a diff tables with other nodes.
 --echo #           (just to check that everything is working fine).

plugin/group_replication/include/certifier.h

@@ -169,12 +169,8 @@ class Certifier_broadcast_thread {
 
   /**
     Terminate broadcast thread.
-
-    @return the operation status
-      @retval 0      OK
-      @retval !=0    Error
   */
-  int terminate();
+  void terminate();
 
   /**
     Broadcast thread worker method.
@@ -268,15 +264,6 @@ class Certifier : public Certifier_interface {
   */
   int initialize(ulonglong gtid_assignment_block_size);
 
-  /**
-    Terminate certifier.
-
-    @return the operation status
-      @retval 0      OK
-      @retval !=0    Error
-  */
-  int terminate();
-
   /**
     Handle view changes on certifier.
    */

plugin/group_replication/src/certifier.cc

@@ -92,8 +92,11 @@ int Certifier_broadcast_thread::initialize() {
   if ((mysql_thread_create(key_GR_THD_cert_broadcast, &broadcast_pthd,
                            get_connection_attrib(), launch_broadcast_thread,
                            (void *)this))) {
-    mysql_mutex_unlock(&broadcast_run_lock); /* purecov: inspected */
-    return 1;                                /* purecov: inspected */
+    /* purecov: begin inspected */
+    mysql_mutex_unlock(&broadcast_run_lock);
+    LogPluginErr(ERROR_LEVEL, ER_GRP_RPL_CERT_BROADCAST_THREAD_CREATE_FAILED);
+    return 1;
+    /* purecov: end */
   }
   broadcast_thd_state.set_created();
 
@@ -108,13 +111,13 @@ int Certifier_broadcast_thread::initialize() {
   return 0;
 }
 
-int Certifier_broadcast_thread::terminate() {
+void Certifier_broadcast_thread::terminate() {
   DBUG_TRACE;
 
   mysql_mutex_lock(&broadcast_run_lock);
   if (broadcast_thd_state.is_thread_dead()) {
     mysql_mutex_unlock(&broadcast_run_lock);
-    return 0;
+    return;
   }
 
   aborted = true;
@@ -135,8 +138,6 @@ int Certifier_broadcast_thread::terminate() {
     mysql_cond_timedwait(&broadcast_run_cond, &broadcast_run_lock, &abstime);
   }
   mysql_mutex_unlock(&broadcast_run_lock);
-
-  return 0;
 }
 
 void Certifier_broadcast_thread::dispatcher() {
@@ -156,6 +157,8 @@ void Certifier_broadcast_thread::dispatcher() {
   mysql_cond_broadcast(&broadcast_run_cond);
   mysql_mutex_unlock(&broadcast_run_lock);
 
+  LogPluginErr(SYSTEM_LEVEL, ER_GRP_RPL_CERT_BROADCAST_THREAD_STARTED);
+
   while (!aborted) {
     // Increase Group Replication feature usage every 10 minutes.
     if (broadcast_counter % 600 == 0 ||
@@ -216,6 +219,8 @@ void Certifier_broadcast_thread::dispatcher() {
   mysql_cond_broadcast(&broadcast_run_cond);
   mysql_mutex_unlock(&broadcast_run_lock);
 
+  LogPluginErr(SYSTEM_LEVEL, ER_GRP_RPL_CERT_BROADCAST_THREAD_STOPPED);
+
   my_thread_exit(nullptr);
 }
 
@@ -324,6 +329,10 @@ Certifier::Certifier()
 Certifier::~Certifier() {
   mysql_mutex_lock(&LOCK_certification_info);
   initialized = false;
+
+  broadcast_thread->terminate();
+  delete broadcast_thread;
+
   clear_certification_info();
   delete certification_info_tsid_map;
 
@@ -334,7 +343,6 @@ Certifier::~Certifier() {
   delete group_gtid_extracted;
   delete group_gtid_tsid_map;
   mysql_mutex_unlock(&LOCK_certification_info);
-  delete broadcast_thread;
 
   mysql_mutex_lock(&LOCK_members);
   clear_members();
@@ -566,15 +574,6 @@ int Certifier::initialize(ulonglong gtid_assignment_block_size) {
   return error;
 }
 
-int Certifier::terminate() {
-  DBUG_TRACE;
-  int error = 0;
-
-  if (is_initialized()) error = broadcast_thread->terminate();
-
-  return error;
-}
-
 void Certifier::update_parallel_applier_indexes(
     bool update_parallel_applier_last_committed_global,
     bool increment_parallel_applier_sequence_number) {

plugin/group_replication/src/handlers/certification_handler.cc

@@ -100,8 +100,6 @@ int Certification_handler::handle_action(Pipeline_action *action) {
     Handler_THD_setup_action *thd_conf_action =
         (Handler_THD_setup_action *)action;
     applier_module_thd = thd_conf_action->get_THD_object();
-  } else if (action_type == HANDLER_STOP_ACTION) {
-    error = cert_module->terminate();
   }
 
   if (error) return error;

share/messages_to_error_log.txt

@@ -12330,6 +12330,15 @@ ER_CHECK_TABLE_FUNCTIONS
 ER_CHECK_TABLE_FUNCTIONS_DETAIL
   eng "%s"
 
+ER_GRP_RPL_CERT_BROADCAST_THREAD_CREATE_FAILED
+  eng "Failed to create the Group Replication certifier broadcast thread (THD_certifier_broadcast)."
+
+ER_GRP_RPL_CERT_BROADCAST_THREAD_STARTED
+  eng "The Group Replication certifier broadcast thread (THD_certifier_broadcast) started."
+
+ER_GRP_RPL_CERT_BROADCAST_THREAD_STOPPED
+  eng "The Group Replication certifier broadcast thread (THD_certifier_broadcast) stopped."
+
 #
 # End of 8.0 error messages intended to be written to the server error log.
 #