BUG#25055190: FLOW-CONTROL APPLIER QUEUE SIZE INCORRECT IN JOINS ON BUSY GROUPS

When a new member joins a group that is busy, one of the statistics
used by the flow-control - applier queue size - may be incorrectly
calculated. This comes from the fact that the queue size is
incremented once the transaction is queued into the relay log, but
it is only decremented when executing if the member is considered
ONLINE. If the member is on RECOVERING status, the queue size is
not decremented.
To fix the above issue, the applier queue counters are only
incremented and decremented when member status is ONLINE.
The certifier queue counter it is always sent, to avoid the
throttling to reach to small values when one or more members are on
RECOVERING, we cap the throttling to 5% of the queues threshold
specified by DBA on
group_replication_flow_control_certifier_threshold and
group_replication_flow_control_applier_threshold options.

A second issue was observed on flow control disable, which was not
enforcing its immediate disable. Now when the low control is disable,
it does reset its used quota which ensures immediate throttling
disable.

(cherry picked from commit d78bb00532032355fc29af70012422d6ede9e992)

Author: nacarvalho

rapid/plugin/group_replication/src/certifier.cc

@@ -1604,8 +1604,11 @@ void Certifier::update_certified_transaction_count(bool result)
   else
     negative_cert++;
 
-  applier_module->get_pipeline_stats_member_collector()
-      ->increment_transactions_certified();
+  if (local_member_info->get_recovery_status() == Group_member_info::MEMBER_ONLINE)
+  {
+    applier_module->get_pipeline_stats_member_collector()
+        ->increment_transactions_certified();
+  }
 }
 
 ulonglong Certifier::get_positive_certified()

rapid/plugin/group_replication/src/handlers/applier_handler.cc

@@ -157,7 +157,8 @@ int Applier_handler::handle_event(Pipeline_event *event,Continuation *cont)
   {
     error= channel_interface.queue_packet((const char*)p->payload, p->len);
 
-    if (event->get_event_type() == binary_log::GTID_LOG_EVENT)
+    if (event->get_event_type() == binary_log::GTID_LOG_EVENT &&
+        local_member_info->get_recovery_status() == Group_member_info::MEMBER_ONLINE)
     {
       applier_module->get_pipeline_stats_member_collector()
           ->increment_transactions_waiting_apply();

rapid/plugin/group_replication/src/pipeline_stats.cc

@@ -205,14 +205,8 @@ Pipeline_stats_member_message::decode_payload(const unsigned char *buffer,
 }
 
 
-/*
-  m_transactions_waiting_apply is initialized with -1 to take
-  View_change_log_event transaction into account, that despite
-  being queued on applier channel is applied through recovery
-  channel.
-*/
 Pipeline_stats_member_collector::Pipeline_stats_member_collector()
-  : m_transactions_waiting_apply(-1), m_transactions_certified(0),
+  : m_transactions_waiting_apply(0), m_transactions_certified(0),
     m_transactions_applied(0), m_transactions_local(0)
 {}
 
@@ -455,7 +449,7 @@ Flow_control_module::flow_control_step()
       if (holds > 0)
       {
         uint num_writing_members= 0;
-        int64 min_certifier_capacity= MAXTPS, min_applier_capacity= MAXTPS;
+        int64 min_certifier_capacity= MAXTPS, min_applier_capacity= MAXTPS, safe_capacity= MAXTPS;
 
         Flow_control_module_info::iterator it= m_info.begin();
         while (it != m_info.end())
@@ -471,15 +465,23 @@ Flow_control_module::flow_control_step()
           else
           {
             if (flow_control_certifier_threshold_var > 0
+                && it->second.get_delta_transactions_certified() > 0
                 && it->second.get_transactions_waiting_certification() - flow_control_certifier_threshold_var > 0
                 && min_certifier_capacity > it->second.get_delta_transactions_certified())
               min_certifier_capacity= it->second.get_delta_transactions_certified();
 
+            if (it->second.get_delta_transactions_certified() > 0)
+              safe_capacity= std::min(safe_capacity, it->second.get_delta_transactions_certified());
+
             if (flow_control_applier_threshold_var > 0
+                && it->second.get_delta_transactions_applied() > 0
                 && it->second.get_transactions_waiting_apply() - flow_control_applier_threshold_var > 0
                 && min_applier_capacity > it->second.get_delta_transactions_applied())
               min_applier_capacity= it->second.get_delta_transactions_applied();
 
+            if (it->second.get_delta_transactions_applied() > 0)
+              safe_capacity= std::min(safe_capacity, it->second.get_delta_transactions_applied());
+
             if (it->second.get_delta_transactions_local() > 0)
               num_writing_members++;
 
@@ -492,6 +494,10 @@ Flow_control_module::flow_control_step()
         int64 min_capacity= (min_certifier_capacity > 0 && min_certifier_capacity < min_applier_capacity)
                              ? min_certifier_capacity : min_applier_capacity;
 
+        // Minimum capacity will never be less than lim_throttle.
+        int64 lim_throttle= 0.05 * std::min(flow_control_certifier_threshold_var,
+                                            flow_control_applier_threshold_var);
+        min_capacity= std::max(std::min(min_capacity, safe_capacity), lim_throttle);
         quota_size= (min_capacity * HOLD_FACTOR) / num_writing_members - extra_quota;
         my_atomic_store64(&m_quota_size, quota_size > 1 ? quota_size : 1);
       }
@@ -510,6 +516,8 @@ Flow_control_module::flow_control_step()
     }
 
     case FCM_DISABLED:
+      my_atomic_store64(&m_quota_size, 0);
+      my_atomic_store64(&m_quota_used, 0);
       break;
 
     default: