Bug #41183 rpl_ndb_circular, rpl_ndb_circular_simplex need maintenance, crash

            
The bug happened because filtering-out a STMT_END_F-flagged event so that
the transaction COMMIT finds traces of incomplete statement commit.
Such situation is only possible with ndb circular replication. The filtered-out
rows event is one that immediately preceeds the COMMIT query event.
      
Fixed with deploying an the rows-log-event statement commit at executing
of the transaction COMMIT event. 
Resources that were allocated by other than STMT_END_F-flagged event of
the last statement are clean up prior execution of the commit logics.


mysql-test/suite/rpl_ndb/t/disabled.def:
  re-enabling two tests.
sql/log_event.cc:
  Adding the statement cleanup to execute at the transaction commit time.
  The statement might not be ended with execution of STMT_END_F-flagged event because of
  the event was filtered out by SERVER_ID rules.
  Small refactoring for Rows_log_event::do_update_pos() to be split on two parts:
  the statement commit that releases its execution time allocated resources, and
  the relay log update.

mysql-test/suite/rpl_ndb/t/disabled.def

@@ -10,7 +10,4 @@
 #
 ##############################################################################
 
-rpl_ndb_circular         : Bug#41183 rpl_ndb_circular, rpl_ndb_circular_simplex need maintenance, crash
-rpl_ndb_circular_simplex : Bug#41183 rpl_ndb_circular, rpl_ndb_circular_simplex need maintenance, crash
-
 # the below testcase have been reworked to avoid the bug, test contains comment, keep bug open

sql/log_event.cc

@@ -51,6 +51,7 @@
 */
 #define FMT_G_BUFSIZE(PREC) (3 + (PREC) + 5 + 1)
 
+static int rows_event_stmt_cleanup(Relay_log_info const *rli, THD* thd);
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 static const char *HA_ERR(int i)
@@ -2894,7 +2895,37 @@ int Query_log_event::do_apply_event(Relay_log_info const *rli,
   DBUG_PRINT("info", ("log_pos: %lu", (ulong) log_pos));
 
   clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
+  if (strcmp("COMMIT", query) == 0 && rli->tables_to_lock)
+  {
+    /*
+      Cleaning-up the last statement context:
+      the terminal event of the current statement flagged with
+      STMT_END_F got filtered out in ndb circular replication.
+    */
+    int error;
+    char llbuff[22];
+    if ((error= rows_event_stmt_cleanup(const_cast<Relay_log_info*>(rli), thd)))
+    {
+      const_cast<Relay_log_info*>(rli)->report(ERROR_LEVEL, error,
+                  "Error in cleaning up after an event preceeding the commit; "
+                  "the group log file/position: %s %s",
+                  const_cast<Relay_log_info*>(rli)->group_master_log_name,
+                  llstr(const_cast<Relay_log_info*>(rli)->group_master_log_pos,
+                        llbuff));
+    }
+    /*
+      Executing a part of rli->stmt_done() logics that does not deal
+      with group position change. The part is redundant now but is 
+      future-change-proof addon, e.g if COMMIT handling will start checking
+      invariants like IN_STMT flag must be off at committing the transaction.
+    */
+    const_cast<Relay_log_info*>(rli)->inc_event_relay_log_pos();
+    const_cast<Relay_log_info*>(rli)->clear_flag(Relay_log_info::IN_STMT);
+  }
+  else
+  {
     const_cast<Relay_log_info*>(rli)->clear_tables_to_lock();
+  }
 
   /*
     Note:   We do not need to execute reset_one_shot_variables() if this
@@ -7403,16 +7434,20 @@ Rows_log_event::do_shall_skip(Relay_log_info *rli)
     return Log_event::do_shall_skip(rli);
 }
 
-int
-Rows_log_event::do_update_pos(Relay_log_info *rli)
-{
-  DBUG_ENTER("Rows_log_event::do_update_pos");
-  int error= 0;
+/**
+   The function is called at Rows_log_event statement commit time,
+   normally from Rows_log_event::do_update_pos() and possibly from
+   Query_log_event::do_apply_event() of the COMMIT.
+   The function commits the last statement for engines, binlog and
+   releases resources have been allocated for the statement.
   
-  DBUG_PRINT("info", ("flags: %s",
-                      get_flags(STMT_END_F) ? "STMT_END_F " : ""));
+   @retval  0         Ok.
+   @retval  non-zero  Error at the commit.
+ */
 
-  if (get_flags(STMT_END_F))
+static int rows_event_stmt_cleanup(Relay_log_info const *rli, THD * thd)
+{
+  int error;
   {
     /*
       This is the end of a statement or transaction, so close (and
@@ -7454,8 +7489,33 @@ Rows_log_event::do_update_pos(Relay_log_info *rli)
 
     thd->reset_current_stmt_binlog_row_based();
 
-    rli->cleanup_context(thd, 0);
-    if (error == 0)
+    const_cast<Relay_log_info*>(rli)->cleanup_context(thd, 0);
+  }
+  return error;
+}
+
+/**
+   The method either increments the relay log position or
+   commits the current statement and increments the master group 
+   possition if the event is STMT_END_F flagged and
+   the statement corresponds to the autocommit query (i.e replicated
+   without wrapping in BEGIN/COMMIT)
+
+   @retval 0         Success
+   @retval non-zero  Error in the statement commit
+ */
+int
+Rows_log_event::do_update_pos(Relay_log_info *rli)
+{
+  DBUG_ENTER("Rows_log_event::do_update_pos");
+  int error= 0;
+
+  DBUG_PRINT("info", ("flags: %s",
+                      get_flags(STMT_END_F) ? "STMT_END_F " : ""));
+
+  if (get_flags(STMT_END_F))
+  {
+    if ((error= rows_event_stmt_cleanup(rli, thd)) == 0)
     {
       /*
         Indicate that a statement is finished.
@@ -7475,12 +7535,14 @@ Rows_log_event::do_update_pos(Relay_log_info *rli)
       thd->clear_error();
     }
     else
+    {
       rli->report(ERROR_LEVEL, error,
                   "Error in %s event: commit of row events failed, "
                   "table `%s`.`%s`",
                   get_type_str(), m_table->s->db.str,
                   m_table->s->table_name.str);
     }
+  }
   else
   {
     rli->inc_event_relay_log_pos();