MDEV-18497 : CTAS async replication from mariadb master crashes galera nodes (#1410)

In MariaDB 10.2 master could have been configured so that there
is extra annotate events. When we peak next event type for CTAS we
need to skip annotate events.

mysql-test/suite/galera/r/galera_as_slave_ctas.result

+connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2;
+connect node_3, 127.0.0.1, root, , test, $NODE_MYPORT_3;
+connection node_2;
 START SLAVE;
+connection node_1;
 SHOW VARIABLES LIKE 'binlog_format';
 Variable_name	Value
 binlog_format	ROW
+connection node_1;
 CREATE TABLE source (f1 INTEGER PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE target AS SELECT * FROM source;
+connection node_2;
+connection node_3;
+connection node_1;
 DROP TABLE target;
 INSERT INTO source VALUES(1);
 CREATE TABLE target AS SELECT * FROM source;
+connection node_2;
+connection node_3;
+connection node_1;
 DROP TABLE source;
 DROP TABLE target;
+connection node_3;
+connection node_2;
 STOP SLAVE;
 RESET SLAVE ALL;
+connection node_1;
 RESET MASTER;

sql/slave.cc

@@ -7397,24 +7397,30 @@ event(errno: %d  cur_log->error: %d)",
 #ifdef WITH_WSREP
 enum Log_event_type wsrep_peak_event(rpl_group_info *rgi, ulonglong* event_size)
 {
+  enum Log_event_type ev_type;
+
   mysql_mutex_lock(&rgi->rli->data_lock);
 
   unsigned long long event_pos= rgi->event_relay_log_pos;
+  unsigned long long orig_future_pos= rgi->future_event_relay_log_pos;
   unsigned long long future_pos= rgi->future_event_relay_log_pos;
 
-  /* scan the log to read next event */
-  my_b_seek(rgi->rli->cur_log, future_pos);
-  rgi->rli->event_relay_log_pos= future_pos;
-  rgi->event_relay_log_pos= future_pos;
-
-  Log_event* ev = next_event(rgi, event_size);
-  enum Log_event_type ev_type= (ev) ? ev->get_type_code() : UNKNOWN_EVENT;
-  delete ev;
+  /* scan the log to read next event and we skip
+     annotate events. */
+  do {
+    my_b_seek(rgi->rli->cur_log, future_pos);
+    rgi->rli->event_relay_log_pos= future_pos;
+    rgi->event_relay_log_pos= future_pos;
+    Log_event* ev= next_event(rgi, event_size);
+    ev_type= (ev) ? ev->get_type_code() : UNKNOWN_EVENT;
+    delete ev;
+    future_pos+= *event_size;
+  } while (ev_type == ANNOTATE_ROWS_EVENT);
 
   /* scan the log back and re-set the positions to original values */
   rgi->rli->event_relay_log_pos= event_pos;
   rgi->event_relay_log_pos= event_pos;
-  my_b_seek(rgi->rli->cur_log, future_pos);
+  my_b_seek(rgi->rli->cur_log, orig_future_pos);
 
   mysql_mutex_unlock(&rgi->rli->data_lock);