MDEV-7145: Delayed slave.

Extend to work also for parallel replication.
Signed-off-by: Kristian Nielsen <knielsen@knielsen-hq.org>

sql/rpl_parallel.cc

@@ -2418,8 +2418,17 @@ rpl_parallel::do_event(rpl_group_info *serial_rgi, Log_event *ev,
       !(unlikely(rli->gtid_skip_flag != GTID_SKIP_NOT) && is_group_event))
     return -1;
 
-  /* ToDo: what to do with this lock?!? */
-  mysql_mutex_unlock(&rli->data_lock);
+  /* Note: rli->data_lock is released by sql_delay_event(). */
+  if (sql_delay_event(ev, rli->sql_driver_thd, serial_rgi))
+  {
+    /*
+      If sql_delay_event() returns non-zero, it means that the wait timed out
+      due to slave stop. We should not queue the event in this case, it must
+      not be applied yet.
+    */
+    delete ev;
+    return 1;
+  }
 
   if (unlikely(typ == FORMAT_DESCRIPTION_EVENT))
   {

sql/slave.cc

@@ -3424,7 +3424,8 @@ has_temporary_error(THD *thd)
 
   @retval nonzero If the delay was interrupted and the event shall be skipped.
 */
-static int sql_delay_event(Log_event *ev, THD *thd, rpl_group_info *rgi)
+int
+sql_delay_event(Log_event *ev, THD *thd, rpl_group_info *rgi)
 {
   Relay_log_info* rli= rgi->rli;
   long sql_delay= rli->get_sql_delay();

sql/slave.h

@@ -251,6 +251,7 @@ void set_slave_thread_options(THD* thd);
 void set_slave_thread_default_charset(THD *thd, rpl_group_info *rgi);
 int rotate_relay_log(Master_info* mi);
 int has_temporary_error(THD *thd);
+int sql_delay_event(Log_event *ev, THD *thd, rpl_group_info *rgi);
 int apply_event_and_update_pos(Log_event* ev, THD* thd,
                                struct rpl_group_info *rgi);
 int apply_event_and_update_pos_for_parallel(Log_event* ev, THD* thd,