BUG#39393 slave-skip-errors does not work when using ROW based replication

                              
RBR was not considering the option --slave-skip-errors.
                              
To fix the problem, we are reporting the ignored ERROR(s) as warnings thus avoiding 
stopping the SQL Thread. Besides, it fixes the output of "SHOW VARIABLES LIKE 
'slave_skip_errors'" which was showing nothing when the value "all" was assigned 
to --slave-skip-errors.
                  
@sql/log_event.cc
  skipped rbr errors when the option skip-slave-errors is set.
@sql/slave.cc
  fixed the output of for SHOW VARIABLES LIKE 'slave_skip_errors'"
@test-cases
  fixed the output of rpl.rpl_idempotency
  updated the test case rpl_skip_error

6 files changed, 316 insertions, 95 deletions

diff --git a/mysql-test/suite/rpl/r/rpl_idempotency.result b/mysql-test/suite/rpl/r/rpl_idempotency.result
index f39d0f1f37f..3341c03db0f 100644
--- a/mysql-test/suite/rpl/r/rpl_idempotency.result
+++ b/mysql-test/suite/rpl/r/rpl_idempotency.result
@@ -141,9 +141,9 @@ select * from ti1 order by b /* must be (2),(3) */;
 b
 2
 3
-*** slave must stop
+*** slave must stop (Trying to delete a referenced foreing key)
 Last_SQL_Error
-0
+1451
 select * from ti1 order by b /* must be (1),(2),(3) - not deleted */;
 b
 1
@@ -159,7 +159,7 @@ set global slave_exec_mode='STRICT';
 *** conspire future problem
 delete from ti1 where b=3;
 insert into ti2 set a=3, b=3 /* offending write event */;
-*** slave must stop
+*** slave must stop (Trying to insert an invalid foreign key)
 Last_SQL_Error
 1452
 select * from ti2 order by b /* must be (2,2) */;
@@ -179,7 +179,7 @@ a	b
 *** conspiring query
 insert into ti1 set b=1;
 insert into ti1 set b=1 /* offending write event */;
-*** slave must stop
+*** slave must stop (Trying to insert a dupliacte key)
 Last_SQL_Error
 1062
 set foreign_key_checks= 0;
@@ -195,32 +195,32 @@ INSERT INTO t2 VALUES (-1),(-2),(-3);
 DELETE FROM t1 WHERE a = -2;
 DELETE FROM t2 WHERE a = -2;
 DELETE FROM t1 WHERE a = -2;
-*** slave must stop
+*** slave must stop (Key was not found)
 Last_SQL_Error
 1032
 set global slave_exec_mode='IDEMPOTENT';
 start slave sql_thread;
 set global slave_exec_mode='STRICT';
 DELETE FROM t2 WHERE a = -2;
-*** slave must stop
+*** slave must stop (Key was not found)
 Last_SQL_Error
-0
+1032
 set global slave_exec_mode='IDEMPOTENT';
 start slave sql_thread;
 set global slave_exec_mode='STRICT';
 UPDATE t1 SET a = 1 WHERE a = -1;
 UPDATE t2 SET a = 1 WHERE a = -1;
 UPDATE t1 SET a = 1 WHERE a = -1;
-*** slave must stop
+*** slave must stop (Key was not found)
 Last_SQL_Error
 1032
 set global slave_exec_mode='IDEMPOTENT';
 start slave sql_thread;
 set global slave_exec_mode='STRICT';
 UPDATE t2 SET a = 1 WHERE a = -1;
-*** slave must stop
+*** slave must stop (Key was not found)
 Last_SQL_Error
-0
+1032
 set global slave_exec_mode='IDEMPOTENT';
 start slave sql_thread;
 SET @@global.slave_exec_mode= @old_slave_exec_mode;
diff --git a/mysql-test/suite/rpl/r/rpl_skip_error.result b/mysql-test/suite/rpl/r/rpl_skip_error.result
index b90d8113e8e..dc02bb783ae 100644
--- a/mysql-test/suite/rpl/r/rpl_skip_error.result
+++ b/mysql-test/suite/rpl/r/rpl_skip_error.result
@@ -74,19 +74,16 @@ Last_SQL_Error
 drop table t1;
 create table t1(a int primary key);
 insert into t1 values (1),(2);
-delete from t1 where @@server_id=1;
-Warnings:
-Warning	1592	Statement is not safe to log in statement format.
+SET SQL_LOG_BIN=0;
+delete from t1;
+SET SQL_LOG_BIN=1;
 set sql_mode=strict_trans_tables;
-insert into t1 values (7), (8), (9);
+insert into t1 values (1), (2), (3);
 [on slave]
 select * from t1;
 a
 1
 2
-7
-8
-9
 SHOW SLAVE STATUS;
 Slave_IO_State	#
 Master_Host	127.0.0.1
@@ -128,3 +125,84 @@ Last_SQL_Errno	0
 Last_SQL_Error	
 ==== Clean Up ====
 drop table t1;
+==== Using Innodb ====
+SET SQL_LOG_BIN=0;
+CREATE TABLE t1(id INT NOT NULL PRIMARY KEY, data INT) Engine=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) NOT NULL,
+  `data` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SET SQL_LOG_BIN=1;
+CREATE TABLE t1(id INT NOT NULL PRIMARY KEY, data INT) Engine=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) NOT NULL,
+  `data` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+INSERT INTO t1 VALUES(1, 1);
+INSERT INTO t1 VALUES(2, 1);
+INSERT INTO t1 VALUES(3, 1);
+INSERT INTO t1 VALUES(4, 1);
+SET SQL_LOG_BIN=0;
+DELETE FROM t1 WHERE id = 4;
+SET SQL_LOG_BIN=1;
+UPDATE t1 SET id= id + 3, data = 2;
+
+SELECT *, "INNODB SET SLAVE DATA" FROM t1 ORDER BY id;
+id	data	INNODB SET SLAVE DATA
+1	1	INNODB SET SLAVE DATA
+2	1	INNODB SET SLAVE DATA
+3	1	INNODB SET SLAVE DATA
+4	1	INNODB SET SLAVE DATA
+SELECT *, "INNODB SET MASTER DATA" FROM t1 ORDER BY id;
+id	data	INNODB SET MASTER DATA
+4	2	INNODB SET MASTER DATA
+5	2	INNODB SET MASTER DATA
+6	2	INNODB SET MASTER DATA
+==== Using MyIsam ====
+SET SQL_LOG_BIN=0;
+CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, data INT) Engine=MyIsam;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `id` int(11) NOT NULL,
+  `data` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SET SQL_LOG_BIN=1;
+CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, data INT) Engine=MyIsam;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `id` int(11) NOT NULL,
+  `data` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+INSERT INTO t2 VALUES(1, 1);
+INSERT INTO t2 VALUES(2, 1);
+INSERT INTO t2 VALUES(3, 1);
+INSERT INTO t2 VALUES(5, 1);
+SET SQL_LOG_BIN=0;
+DELETE FROM t2 WHERE id = 5;
+SET SQL_LOG_BIN=1;
+UPDATE t2 SET id= id + 3, data = 2;
+
+SELECT *, "MYISAM SET SLAVE DATA" FROM t2 ORDER BY id;
+id	data	MYISAM SET SLAVE DATA
+2	1	MYISAM SET SLAVE DATA
+3	1	MYISAM SET SLAVE DATA
+4	2	MYISAM SET SLAVE DATA
+5	1	MYISAM SET SLAVE DATA
+SELECT *, "MYISAM SET MASTER DATA" FROM t2 ORDER BY id;
+id	data	MYISAM SET MASTER DATA
+4	2	MYISAM SET MASTER DATA
+5	2	MYISAM SET MASTER DATA
+6	2	MYISAM SET MASTER DATA
+==== Clean Up ====
+DROP TABLE t1;
+DROP TABLE t2;
diff --git a/mysql-test/suite/rpl/t/rpl_idempotency.test b/mysql-test/suite/rpl/t/rpl_idempotency.test
index 1946aa100ab..bfd1860759e 100644
--- a/mysql-test/suite/rpl/t/rpl_idempotency.test
+++ b/mysql-test/suite/rpl/t/rpl_idempotency.test
@@ -208,7 +208,7 @@ select * from ti1 order by b /* must be (2),(3) */;
 
 # foreign key: row is referenced
 
---echo *** slave must stop
+--echo *** slave must stop (Trying to delete a referenced foreing key)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
@@ -242,7 +242,7 @@ delete from ti1 where b=3;
 connection master;
 insert into ti2 set a=3, b=3 /* offending write event */;
 
---echo *** slave must stop
+--echo *** slave must stop (Trying to insert an invalid foreign key)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
@@ -281,7 +281,7 @@ insert into ti1 set b=1;
 connection master;
 insert into ti1 set b=1 /* offending write event */;
 
---echo *** slave must stop
+--echo *** slave must stop (Trying to insert a dupliacte key)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
@@ -316,7 +316,7 @@ DELETE FROM t2 WHERE a = -2;
 connection master;
 DELETE FROM t1 WHERE a = -2;
 
---echo *** slave must stop
+--echo *** slave must stop (Key was not found)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
@@ -333,8 +333,8 @@ sync_slave_with_master;
 set global slave_exec_mode='STRICT';
 
 connection master;
-DELETE FROM t2 WHERE a = -2;
---echo *** slave must stop
+DELETE FROM t2 WHERE a = -2; 
+--echo *** slave must stop (Key was not found)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
@@ -356,7 +356,7 @@ UPDATE t2 SET a = 1 WHERE a = -1;
 connection master;
 UPDATE t1 SET a = 1 WHERE a = -1;
 
---echo *** slave must stop
+--echo *** slave must stop (Key was not found)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
@@ -376,7 +376,7 @@ set global slave_exec_mode='STRICT';
 connection master;
 UPDATE t2 SET a = 1 WHERE a = -1;
 
---echo *** slave must stop
+--echo *** slave must stop (Key was not found)
 connection slave;
 source include/wait_for_slave_sql_to_stop.inc;
 
diff --git a/mysql-test/suite/rpl/t/rpl_skip_error.test b/mysql-test/suite/rpl/t/rpl_skip_error.test
index cac797d3797..c15d24be9e5 100644
--- a/mysql-test/suite/rpl/t/rpl_skip_error.test
+++ b/mysql-test/suite/rpl/t/rpl_skip_error.test
@@ -8,18 +8,23 @@
 # ==== Method ====
 #
 # We run the slave with --slave-skip-errors=1062 (the code for
-# duplicate key).  On slave, we insert value 1 in a table, and then,
-# on master, we insert value 1 in the table.  The error should be
-# ignored on slave.
-#
+# duplicate key).  Then we have two set of tests. In the first 
+# set, we insert value 1 in a table on the slave, and then, on 
+# master, we insert value 1 in the table. In the second set, we 
+# insert several values on the master, disable the binlog and
+# delete one of the values and re-enable the binlog. Right after,
+# we perform an update on the set of values in order to generate
+# a duplicate key on the slave. The errors should be ignored on
+# the slave.
+# 
 # ==== Related bugs ====
 #
 # BUG#28839: Errors in strict mode silently stop SQL thread if --slave-skip-errors exists
-# bug in this test: BUG#30594: rpl.rpl_skip_error is nondeterministic
+# bug in this test: BUG#30594: rpl.rpl_skip_error is nondeterministic:
+# BUG#39393: slave-skip-errors does not work when using ROW based replication
 
 source include/master-slave.inc;
-source include/have_binlog_format_statement.inc;
-
+source include/have_innodb.inc;
 
 --echo ==== Test Without sql_mode=strict_trans_tables ====
 
@@ -64,9 +69,11 @@ sync_slave_with_master;
 connection master;
 create table t1(a int primary key);
 insert into t1 values (1),(2);
-delete from t1 where @@server_id=1;
+SET SQL_LOG_BIN=0;
+delete from t1;
+SET SQL_LOG_BIN=1;
 set sql_mode=strict_trans_tables;
-insert into t1 values (7), (8), (9);
+insert into t1 values (1), (2), (3);
 
 --echo [on slave]
 sync_slave_with_master;
@@ -80,3 +87,93 @@ connection master;
 drop table t1;
 sync_slave_with_master;
 # End of 5.0 tests
+
+#
+# BUG#39393: slave-skip-errors does not work when using ROW based replication
+#
+--echo ==== Using Innodb ====
+
+connection master;
+
+SET SQL_LOG_BIN=0;
+CREATE TABLE t1(id INT NOT NULL PRIMARY KEY, data INT) Engine=InnoDB;
+SHOW CREATE TABLE t1;
+SET SQL_LOG_BIN=1;
+
+connection slave;
+
+CREATE TABLE t1(id INT NOT NULL PRIMARY KEY, data INT) Engine=InnoDB;
+SHOW CREATE TABLE t1;
+
+connection master;
+
+INSERT INTO t1 VALUES(1, 1);
+INSERT INTO t1 VALUES(2, 1);
+INSERT INTO t1 VALUES(3, 1);
+INSERT INTO t1 VALUES(4, 1);
+
+SET SQL_LOG_BIN=0;
+DELETE FROM t1 WHERE id = 4;
+SET SQL_LOG_BIN=1;
+UPDATE t1 SET id= id + 3, data = 2;
+
+sync_slave_with_master;
+
+let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
+echo $error;
+
+connection slave;
+
+SELECT *, "INNODB SET SLAVE DATA" FROM t1 ORDER BY id;
+
+connection master;
+
+SELECT *, "INNODB SET MASTER DATA" FROM t1 ORDER BY id;
+
+--echo ==== Using MyIsam ====
+
+connection master;
+
+SET SQL_LOG_BIN=0;
+CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, data INT) Engine=MyIsam;
+SHOW CREATE TABLE t2;
+SET SQL_LOG_BIN=1;
+
+connection slave;
+
+CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, data INT) Engine=MyIsam;
+SHOW CREATE TABLE t2;
+
+connection master;
+
+INSERT INTO t2 VALUES(1, 1);
+INSERT INTO t2 VALUES(2, 1);
+INSERT INTO t2 VALUES(3, 1);
+INSERT INTO t2 VALUES(5, 1);
+
+SET SQL_LOG_BIN=0;
+DELETE FROM t2 WHERE id = 5;
+SET SQL_LOG_BIN=1;
+UPDATE t2 SET id= id + 3, data = 2;
+
+sync_slave_with_master;
+
+let $error= query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
+echo $error;
+
+connection slave;
+
+SELECT *, "MYISAM SET SLAVE DATA" FROM t2 ORDER BY id;
+
+connection master;
+
+SELECT *, "MYISAM SET MASTER DATA" FROM t2 ORDER BY id;
+
+--echo ==== Clean Up ====
+
+connection master;
+
+DROP TABLE t1;
+DROP TABLE t2;
+
+sync_slave_with_master;
diff --git a/sql/log_event.cc b/sql/log_event.cc
index aa7e8c84a0f..423174f77d5 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -277,6 +277,47 @@ static void clear_all_errors(THD *thd, Relay_log_info *rli)
   rli->clear_error();
 }
 
+inline int idempotent_error_code(int err_code)
+{
+  int ret= 0;
+
+  switch (err_code)
+  {
+    case 0:
+      ret= 1;
+    break;
+    /*
+      The following list of "idempotent" errors
+      means that an error from the list might happen
+      because of idempotent (more than once)
+      applying of a binlog file.
+      Notice, that binlog has a  ddl operation its
+      second applying may cause
+
+      case HA_ERR_TABLE_DEF_CHANGED:
+      case HA_ERR_CANNOT_ADD_FOREIGN:
+
+      which are not included into to the list.
+
+      Note that HA_ERR_RECORD_DELETED is not in the list since
+      do_exec_row() should not return that error code.
+    */
+    case HA_ERR_RECORD_CHANGED:
+    case HA_ERR_KEY_NOT_FOUND:
+    case HA_ERR_END_OF_FILE:
+    case HA_ERR_FOUND_DUPP_KEY:
+    case HA_ERR_FOUND_DUPP_UNIQUE:
+    case HA_ERR_FOREIGN_DUPLICATE_KEY:
+    case HA_ERR_NO_REFERENCED_ROW:
+    case HA_ERR_ROW_IS_REFERENCED:
+      ret= 1;
+    break;
+    default:
+      ret= 0;
+    break;
+  }
+  return (ret);
+}
 
 /**
   Ignore error code specified on command line.
@@ -301,14 +342,37 @@ inline int ignored_error_code(int err_code)
   return ((err_code == ER_SLAVE_IGNORED_TABLE) ||
           (use_slave_mask && bitmap_is_set(&slave_error_mask, err_code)));
 }
-#endif
 
+/*
+  This function converts an engine's error to a server error.
+   
+  If the thread does not have an error already reported, it tries to 
+  define it by calling the engine's method print_error. However, if a 
+  mapping is not found, it uses the ER_UNKNOWN_ERROR and prints out a 
+  warning message.
+*/ 
+int convert_handler_error(int error, THD* thd, TABLE *table)
+{
+  uint actual_error= (thd->is_error() ? thd->main_da.sql_errno() :
+                           0);
+
+  if (actual_error == 0)
+  {
+    table->file->print_error(error, MYF(0));
+    actual_error= (thd->is_error() ? thd->main_da.sql_errno() :
+                        ER_UNKNOWN_ERROR);
+    if (actual_error == ER_UNKNOWN_ERROR)
+      if (global_system_variables.log_warnings)
+        sql_print_warning("Unknown error detected %d in handler", error);
+  }
+
+  return (actual_error);
+}
 
 /*
   pretty_print_str()
 */
 
-#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 static char *pretty_print_str(char *packet, const char *str, int len)
 {
   const char *end= str + len;
@@ -7158,7 +7222,9 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
       {
         /*
           Error reporting borrowed from Query_log_event with many excessive
-          simplifications (we don't honour --slave-skip-errors)
+          simplifications. 
+          We should not honour --slave-skip-errors at this point as we are
+          having severe errors which should not be skiped.
         */
         rli->report(ERROR_LEVEL, actual_error,
                     "Error '%s' on opening tables",
@@ -7184,6 +7250,10 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
       {
         if (ptr->m_tabledef.compatible_with(rli, ptr->table))
         {
+          /*
+            We should not honour --slave-skip-errors at this point as we are
+            having severe errors which should not be skiped.
+          */
           mysql_unlock_tables(thd, thd->lock);
           thd->lock= 0;
           thd->is_slave_error= 1;
@@ -7275,7 +7345,6 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
 
     // Do event specific preparations 
     error= do_before_row_operations(rli);
-
     // row processing loop
 
     while (error == 0 && m_curr_row < m_rows_end)
@@ -7291,48 +7360,27 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
       DBUG_ASSERT(error != HA_ERR_RECORD_DELETED);
 
       table->in_use = old_thd;
-      switch (error)
-      {
-      case 0:
-	break;
-      /*
-        The following list of "idempotent" errors
-        means that an error from the list might happen
-        because of idempotent (more than once) 
-        applying of a binlog file.
-        Notice, that binlog has a  ddl operation its
-        second applying may cause
-
-        case HA_ERR_TABLE_DEF_CHANGED:
-        case HA_ERR_CANNOT_ADD_FOREIGN:
-
-        which are not included into to the list.
-
-        Note that HA_ERR_RECORD_DELETED is not in the list since
-        do_exec_row() should not return that error code.
-      */
-      case HA_ERR_RECORD_CHANGED:
-      case HA_ERR_KEY_NOT_FOUND:
-      case HA_ERR_END_OF_FILE:
-      case HA_ERR_FOUND_DUPP_KEY:
-      case HA_ERR_FOUND_DUPP_UNIQUE:
-      case HA_ERR_FOREIGN_DUPLICATE_KEY:
-      case HA_ERR_NO_REFERENCED_ROW:
-      case HA_ERR_ROW_IS_REFERENCED:
 
-        if (bit_is_set(slave_exec_mode, SLAVE_EXEC_MODE_IDEMPOTENT) == 1)
+      if (error)
+      {
+        int actual_error= convert_handler_error(error, thd, table);
+        bool idempotent_error= (idempotent_error_code(error) &&
+                                ((bit_is_set(slave_exec_mode, 
+                                SLAVE_EXEC_MODE_IDEMPOTENT)) == 1));
+        bool ignored_error= (idempotent_error == 0 ?
+                             ignored_error_code(actual_error) : 0);
+
+        if (idempotent_error || ignored_error)
         {
           if (global_system_variables.log_warnings)
             slave_rows_error_report(WARNING_LEVEL, error, rli, thd, table,
                                     get_type_str(),
                                     RPL_LOG_NAME, (ulong) log_pos);
+          clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
           error= 0;
+          if (idempotent_error == 0)
+            break;
         }
-        break;
-        
-      default:
-	thd->is_slave_error= 1;
-	break;
       }
 
       /*
@@ -7346,7 +7394,7 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
                           (ulong) m_curr_row, (ulong) m_curr_row_end, (ulong) m_rows_end));
 
       if (!m_curr_row_end && !error)
-        unpack_current_row(rli);
+        error= unpack_current_row(rli);
   
       // at this moment m_curr_row_end should be set
       DBUG_ASSERT(error || m_curr_row_end != NULL); 
@@ -7359,7 +7407,19 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
 
     DBUG_EXECUTE_IF("STOP_SLAVE_after_first_Rows_event",
                     const_cast<Relay_log_info*>(rli)->abort_slave= 1;);
-    error= do_after_row_operations(rli, error);
+
+    if ((error= do_after_row_operations(rli, error)) &&
+        ignored_error_code(convert_handler_error(error, thd, table)))
+    {
+
+      if (global_system_variables.log_warnings)
+        slave_rows_error_report(WARNING_LEVEL, error, rli, thd, table,
+                                get_type_str(),
+                                RPL_LOG_NAME, (ulong) log_pos);
+      clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
+      error= 0;
+    }
+
     if (!cache_stmt)
     {
       DBUG_PRINT("info", ("Marked that we need to keep log"));
@@ -7375,36 +7435,21 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
     const_cast<Relay_log_info*>(rli)->clear_tables_to_lock();
   
   if (error)
-  {                     /* error has occured during the transaction */
-    slave_rows_error_report(ERROR_LEVEL, error, rli, thd, table,
-                            get_type_str(), RPL_LOG_NAME, (ulong) log_pos);
-  }
-  if (error)
   {
-    /*
-      If one day we honour --skip-slave-errors in row-based replication, and
-      the error should be skipped, then we would clear mappings, rollback,
-      close tables, but the slave SQL thread would not stop and then may
-      assume the mapping is still available, the tables are still open...
-      So then we should clear mappings/rollback/close here only if this is a
-      STMT_END_F.
-      For now we code, knowing that error is not skippable and so slave SQL
-      thread is certainly going to stop.
-      rollback at the caller along with sbr.
-    */
+    slave_rows_error_report(ERROR_LEVEL, error, rli, thd, table,
+                             get_type_str(),
+                             RPL_LOG_NAME, (ulong) log_pos);
     thd->reset_current_stmt_binlog_row_based();
     const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
     thd->is_slave_error= 1;
-    DBUG_RETURN(error);
   }
-
   /*
     This code would ideally be placed in do_update_pos() instead, but
     since we have no access to table there, we do the setting of
     last_event_start_time here instead.
   */
-  if (table && (table->s->primary_key == MAX_KEY) &&
-      !cache_stmt && get_flags(STMT_END_F) == RLE_NO_FLAGS)
+  else if (table && (table->s->primary_key == MAX_KEY) &&
+           !cache_stmt && get_flags(STMT_END_F) == RLE_NO_FLAGS)
   {
     /*
       ------------ Temporary fix until WL#2975 is implemented ---------
@@ -7425,7 +7470,7 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
     const_cast<Relay_log_info*>(rli)->last_event_start_time= my_time(0);
   }
 
-  DBUG_RETURN(0);
+  DBUG_RETURN(error);
 }
 
 Log_event::enum_skip_reason
diff --git a/sql/slave.cc b/sql/slave.cc
index 8c29cb8a72f..fed53cf605e 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -361,6 +361,7 @@ void init_slave_skip_errors(const char* arg)
   if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const uchar*)"all",4))
   {
     bitmap_set_all(&slave_error_mask);
+    print_slave_skip_errors();
     DBUG_VOID_RETURN;
   }
   for (p= arg ; *p; )