MDEV-21675: Data inconsistency after multirow insert rollback (#1474)

* Remove dead code

* MDEV-21675 Data inconsistency after multirow insert rollback

This patch fixes data inconsistencies that happen after rollback of
multirow inserts, with binlog disabled.
For example, statements such as `INSERT INTO t1 VALUES (1,'a'),(1,'b')`
that fail with duplicate key error. In such cases the whole statement
is rolled back. However, with wsrep_emulate_binlog in effect, the
IO_CACHE would not be truncated, and the pending rows events would be
replicated to the rest of the cluster. In the above example, it would
result in row (1,'a') being replicated, whereas locally the statement
is rolled back entirely. Making the cluster inconsistent.
The patch changes the code so that prior to statement rollback,
pending rows event are removed and the stmt cache reset.
That patch also introduces MTR tests that excercise multirow insert
statements for regular, and streaming replication.

12 files changed, 484 insertions, 67 deletions

diff --git a/mysql-test/suite/galera/r/galera_multirow_rollback.result b/mysql-test/suite/galera/r/galera_multirow_rollback.result
new file mode 100644
index 00000000000..13502d0d3dc
--- /dev/null
+++ b/mysql-test/suite/galera/r/galera_multirow_rollback.result
@@ -0,0 +1,73 @@
+connection node_2;
+connection node_1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+START TRANSACTION;
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+COMMIT;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_2;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+DROP TABLE t1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+START TRANSACTION;
+INSERT INTO t1 VALUES (1, 'a');
+INSERT INTO t1 VALUES (2, 'b');
+INSERT INTO t1 (f2) VALUES ('c'), ('d');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+COMMIT;
+expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+f1	f2
+1	a
+2	b
+connection node_2;
+expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+f1	f2
+1	a
+2	b
+DROP TABLE t1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+INSERT INTO t1 (f2) VALUES ('a'),('b');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_2;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+DROP TABLE t1;
+connection node_1;
+CREATE TABLE p(id int primary key, j int) ENGINE=InnoDB;
+CREATE TABLE c(id int primary key, fk1 int) ENGINE=InnoDB;
+ALTER TABLE c ADD FOREIGN KEY (fk1) references p(id);
+INSERT INTO p VALUES(1, 0);
+START TRANSACTION;
+INSERT INTO c VALUES (3,1);
+INSERT INTO c VALUES (1,1), (2,2);
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails (`test`.`c`, CONSTRAINT `c_ibfk_1` FOREIGN KEY (`fk1`) REFERENCES `p` (`id`))
+COMMIT;
+SELECT * FROM p;
+id	j
+1	0
+SELECT * FROM c;
+id	fk1
+3	1
+connection node_2;
+SELECT * FROM p;
+id	j
+1	0
+SELECT * FROM c;
+id	fk1
+3	1
+DROP TABLE c;
+DROP TABLE p;
diff --git a/mysql-test/suite/galera/t/galera_multirow_rollback.combinations b/mysql-test/suite/galera/t/galera_multirow_rollback.combinations
new file mode 100644
index 00000000000..1ce3b45aa1a
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_multirow_rollback.combinations
@@ -0,0 +1,4 @@
+[binlogon]
+log-bin
+
+[binlogoff]
diff --git a/mysql-test/suite/galera/t/galera_multirow_rollback.test b/mysql-test/suite/galera/t/galera_multirow_rollback.test
new file mode 100644
index 00000000000..a5aaedd8bd4
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_multirow_rollback.test
@@ -0,0 +1,89 @@
+#
+# Test multirow insert rollback
+#
+
+--source include/galera_cluster.inc
+
+#
+# Case 1: error on multirow insert results in empty transaction
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+START TRANSACTION;
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+COMMIT;
+
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 2: error on multirow insert does not affect previous statements
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+START TRANSACTION;
+INSERT INTO t1 VALUES (1, 'a');
+INSERT INTO t1 VALUES (2, 'b');
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('c'), ('d');
+COMMIT;
+
+--echo expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+
+--connection node_2
+--echo expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 3: error on autocommit multirow insert
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('a'),('b');
+
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 4: FK constraint violation on multirow insert
+#
+--connection node_1
+CREATE TABLE p(id int primary key, j int) ENGINE=InnoDB;
+CREATE TABLE c(id int primary key, fk1 int) ENGINE=InnoDB;
+ALTER TABLE c ADD FOREIGN KEY (fk1) references p(id);
+INSERT INTO p VALUES(1, 0);
+
+START TRANSACTION;
+INSERT INTO c VALUES (3,1);
+--error ER_NO_REFERENCED_ROW_2
+INSERT INTO c VALUES (1,1), (2,2);
+COMMIT;
+
+SELECT * FROM p;
+SELECT * FROM c;
+
+--connection node_2
+SELECT * FROM p;
+SELECT * FROM c;
+
+DROP TABLE c;
+DROP TABLE p;
diff --git a/mysql-test/suite/galera_sr/r/galera_sr_multirow_rollback.result b/mysql-test/suite/galera_sr/r/galera_sr_multirow_rollback.result
new file mode 100644
index 00000000000..1795d16bbfa
--- /dev/null
+++ b/mysql-test/suite/galera_sr/r/galera_sr_multirow_rollback.result
@@ -0,0 +1,127 @@
+connection node_2;
+connection node_1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+SET SESSION wsrep_trx_fragment_size = 1;
+START TRANSACTION;
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+COMMIT;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_2;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+DROP TABLE t1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+SET SESSION wsrep_trx_fragment_size = 1000;
+START TRANSACTION;
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+COMMIT;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_2;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+DROP TABLE t1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+SET SESSION wsrep_trx_fragment_size = 1000;
+START TRANSACTION;
+INSERT INTO t1 VALUES (1, 'a');
+INSERT INTO t1 VALUES (2, 'b');
+INSERT INTO t1 (f2) VALUES ('c'), ('d');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+COMMIT;
+expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+f1	f2
+1	a
+2	b
+connection node_2;
+expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+f1	f2
+1	a
+2	b
+DROP TABLE t1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+connection node_1;
+SET SESSION wsrep_trx_fragment_size = 1;
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_2;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_1;
+SET SESSION wsrep_trx_fragment_size = 1000;
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+ERROR 23000: Duplicate entry '0' for key 'PRIMARY'
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+connection node_2;
+SELECT COUNT(*) AS expect_0 FROM t1;
+expect_0
+0
+DROP TABLE t1;
+connection node_1;
+CREATE TABLE p(id int primary key, j int) ENGINE=InnoDB;
+CREATE TABLE c(id int primary key, fk1 int) ENGINE=InnoDB;
+ALTER TABLE c ADD FOREIGN KEY (fk1) references p(id);
+INSERT INTO p VALUES(1, 0);
+SET SESSION wsrep_trx_fragment_size=1;
+START TRANSACTION;
+INSERT INTO c VALUES (3,1);
+INSERT INTO c VALUES (1,1), (2,2);
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+COMMIT;
+SELECT * FROM p;
+id	j
+1	0
+SELECT * FROM c;
+id	fk1
+connection node_2;
+SELECT * FROM p;
+id	j
+1	0
+SELECT * FROM c;
+id	fk1
+DROP TABLE c;
+DROP TABLE p;
+connection node_1;
+CREATE TABLE p(id int primary key, j int) ENGINE=InnoDB;
+CREATE TABLE c(id int primary key, fk1 int) ENGINE=InnoDB;
+ALTER TABLE c ADD FOREIGN KEY (fk1) references p(id);
+INSERT INTO p VALUES(1, 0);
+SET SESSION wsrep_trx_fragment_size=1000;
+START TRANSACTION;
+INSERT INTO c VALUES (3,1);
+INSERT INTO c VALUES (1,1), (2,2);
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails (`test`.`c`, CONSTRAINT `c_ibfk_1` FOREIGN KEY (`fk1`) REFERENCES `p` (`id`))
+COMMIT;
+SELECT * FROM p;
+id	j
+1	0
+SELECT * FROM c;
+id	fk1
+3	1
+connection node_2;
+SELECT * FROM p;
+id	j
+1	0
+SELECT * FROM c;
+id	fk1
+3	1
+DROP TABLE c;
+DROP TABLE p;
diff --git a/mysql-test/suite/galera_sr/t/galera_sr_multirow_rollback.combinations b/mysql-test/suite/galera_sr/t/galera_sr_multirow_rollback.combinations
new file mode 100644
index 00000000000..1ce3b45aa1a
--- /dev/null
+++ b/mysql-test/suite/galera_sr/t/galera_sr_multirow_rollback.combinations
@@ -0,0 +1,4 @@
+[binlogon]
+log-bin
+
+[binlogoff]
diff --git a/mysql-test/suite/galera_sr/t/galera_sr_multirow_rollback.test b/mysql-test/suite/galera_sr/t/galera_sr_multirow_rollback.test
new file mode 100644
index 00000000000..9004d332393
--- /dev/null
+++ b/mysql-test/suite/galera_sr/t/galera_sr_multirow_rollback.test
@@ -0,0 +1,156 @@
+#
+# Test multirow insert rollback with streaming replication
+#
+
+--source include/galera_cluster.inc
+
+
+#
+# Case 1: multirow insert results full rollback if a fragment
+#         managed to replicate
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+SET SESSION wsrep_trx_fragment_size = 1;
+START TRANSACTION;
+# With fragment size 1 we expect full rollback
+# because a fragment is already replicated.
+# Therefore, expect ER_LOCK_DEADLOCK instead of ER_DUP_ENTRY
+--error ER_LOCK_DEADLOCK
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+COMMIT;
+
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 2: error on multirow insert results in empty commit
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+SET SESSION wsrep_trx_fragment_size = 1000;
+START TRANSACTION;
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+COMMIT;
+
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 3: error on multirow insert does not affect previous statements
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+SET SESSION wsrep_trx_fragment_size = 1000;
+START TRANSACTION;
+INSERT INTO t1 VALUES (1, 'a');
+INSERT INTO t1 VALUES (2, 'b');
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('c'), ('d');
+COMMIT;
+
+--echo expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+
+--connection node_2
+--echo expect (1,'a'), (2, 'b')
+SELECT * FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 4: error on autocommit multirow insert
+#
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY DEFAULT 0, f2 char(12));
+
+--connection node_1
+SET SESSION wsrep_trx_fragment_size = 1;
+
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+ --connection node_1
+SET SESSION wsrep_trx_fragment_size = 1000;
+--error ER_DUP_ENTRY
+INSERT INTO t1 (f2) VALUES ('a'), ('b');
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+--connection node_2
+SELECT COUNT(*) AS expect_0 FROM t1;
+
+DROP TABLE t1;
+
+
+#
+# Case 5: FK constraint violation on multirow insert results
+#         full rollback if a fragment has already replicated
+#
+--connection node_1
+CREATE TABLE p(id int primary key, j int) ENGINE=InnoDB;
+CREATE TABLE c(id int primary key, fk1 int) ENGINE=InnoDB;
+ALTER TABLE c ADD FOREIGN KEY (fk1) references p(id);
+INSERT INTO p VALUES(1, 0);
+
+SET SESSION wsrep_trx_fragment_size=1;
+START TRANSACTION;
+INSERT INTO c VALUES (3,1);
+--error ER_LOCK_DEADLOCK
+INSERT INTO c VALUES (1,1), (2,2);
+COMMIT;
+
+SELECT * FROM p;
+SELECT * FROM c;
+
+--connection node_2
+SELECT * FROM p;
+SELECT * FROM c;
+
+DROP TABLE c;
+DROP TABLE p;
+
+
+#
+# Case 6: FK constraint violation on multirow insert results
+#         stmt rollback if no fragments have replicated
+#
+--connection node_1
+CREATE TABLE p(id int primary key, j int) ENGINE=InnoDB;
+CREATE TABLE c(id int primary key, fk1 int) ENGINE=InnoDB;
+ALTER TABLE c ADD FOREIGN KEY (fk1) references p(id);
+INSERT INTO p VALUES(1, 0);
+
+SET SESSION wsrep_trx_fragment_size=1000;
+START TRANSACTION;
+INSERT INTO c VALUES (3,1);
+--error ER_NO_REFERENCED_ROW_2
+INSERT INTO c VALUES (1,1), (2,2);
+COMMIT;
+
+SELECT * FROM p;
+SELECT * FROM c;
+
+--connection node_2
+SELECT * FROM p;
+SELECT * FROM c;
+
+DROP TABLE c;
+DROP TABLE p;
diff --git a/sql/log.cc b/sql/log.cc
index ffeb3661783..1da73ab25df 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -10702,7 +10702,6 @@ maria_declare_plugin(binlog)
 maria_declare_plugin_end;
 
 #ifdef WITH_WSREP
-#include "wsrep_trans_observer.h"
 #include "wsrep_mysqld.h"
 
 IO_CACHE *wsrep_get_trans_cache(THD * thd)
@@ -10725,33 +10724,33 @@ void wsrep_thd_binlog_trx_reset(THD * thd)
   /*
     todo: fix autocommit select to not call the caller
   */
-  if (thd_get_ha_data(thd, binlog_hton) != NULL)
+  binlog_cache_mngr *const cache_mngr=
+    (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
+  if (cache_mngr)
   {
-    binlog_cache_mngr *const cache_mngr=
-      (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
-    if (cache_mngr)
+    cache_mngr->reset(false, true);
+    if (!cache_mngr->stmt_cache.empty())
     {
-      cache_mngr->reset(false, true);
-      if (!cache_mngr->stmt_cache.empty())
-      {
-        WSREP_DEBUG("pending events in stmt cache, sql: %s", thd->query());
-        cache_mngr->stmt_cache.reset();
-      }
+      WSREP_DEBUG("pending events in stmt cache, sql: %s", thd->query());
+      cache_mngr->stmt_cache.reset();
     }
   }
   thd->clear_binlog_table_maps();
   DBUG_VOID_RETURN;
 }
 
-
-void thd_binlog_rollback_stmt(THD * thd)
+void wsrep_thd_binlog_stmt_rollback(THD * thd)
 {
-  WSREP_DEBUG("thd_binlog_rollback_stmt connection: %llu",
-	      thd->thread_id);
+  DBUG_ENTER("wsrep_thd_binlog_stmt_rollback");
+  WSREP_DEBUG("wsrep_thd_binlog_stmt_rollback");
   binlog_cache_mngr *const cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
   if (cache_mngr)
-    cache_mngr->trx_cache.set_prev_position(MY_OFF_T_UNDEF);
+  {
+    thd->binlog_remove_pending_rows_event(TRUE, TRUE);
+    cache_mngr->stmt_cache.reset();
+  }
+  DBUG_VOID_RETURN;
 }
 
 bool wsrep_stmt_rollback_is_safe(THD* thd)
diff --git a/sql/log.h b/sql/log.h
index eef81c46ac4..4b80bdfd81f 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -1222,6 +1222,7 @@ static inline TC_LOG *get_tc_log_implementation()
 #ifdef WITH_WSREP
 IO_CACHE* wsrep_get_trans_cache(THD *);
 void wsrep_thd_binlog_trx_reset(THD * thd);
+void wsrep_thd_binlog_stmt_rollback(THD * thd);
 #endif /* WITH_WSREP */
 
 class Gtid_list_log_event;
diff --git a/sql/wsrep_binlog.cc b/sql/wsrep_binlog.cc
index 84fce7c2b2d..787ebc042ae 100644
--- a/sql/wsrep_binlog.cc
+++ b/sql/wsrep_binlog.cc
@@ -228,40 +228,6 @@ void wsrep_dump_rbr_buf(THD *thd, const void* rbr_buf, size_t buf_len)
   free(filename);
 }
 
-/*
-  wsrep exploits binlog's caches even if binlogging itself is not
-  activated. In such case connection close needs calling
-  actual binlog's method.
-  Todo: split binlog hton from its caches to use ones by wsrep
-  without referring to binlog's stuff.
-*/
-int wsrep_binlog_close_connection(THD* thd)
-{
-  DBUG_ENTER("wsrep_binlog_close_connection");
-  if (thd_get_ha_data(thd, binlog_hton) != NULL)
-    binlog_hton->close_connection (binlog_hton, thd);
-  DBUG_RETURN(0);
-}
-
-int wsrep_binlog_savepoint_set(THD *thd,  void *sv)
-{
-  if (!wsrep_emulate_bin_log) return 0;
-  int rcode= binlog_hton->savepoint_set(binlog_hton, thd, sv);
-  return rcode;
-}
-
-int wsrep_binlog_savepoint_rollback(THD *thd, void *sv)
-{
-  if (!wsrep_emulate_bin_log) return 0;
-  int rcode= binlog_hton->savepoint_rollback(binlog_hton, thd, sv);
-  return rcode;
-}
-
-void thd_binlog_flush_pending_rows_event(THD *thd, bool stmt_end)
-{
-  thd->binlog_flush_pending_rows_event(stmt_end);
-}
-
 /* Dump replication buffer along with header to a file. */
 void wsrep_dump_rbr_buf_with_header(THD *thd, const void *rbr_buf,
                                     size_t buf_len)
@@ -343,8 +309,6 @@ cleanup1:
   DBUG_VOID_RETURN;
 }
 
-#include "log_event.h"
-
 int wsrep_write_skip_event(THD* thd)
 {
   DBUG_ENTER("wsrep_write_skip_event");
diff --git a/sql/wsrep_binlog.h b/sql/wsrep_binlog.h
index 4e29b30baca..252fbe602d2 100644
--- a/sql/wsrep_binlog.h
+++ b/sql/wsrep_binlog.h
@@ -50,8 +50,6 @@ void wsrep_dump_rbr_buf(THD *thd, const void* rbr_buf, size_t buf_len);
 void wsrep_dump_rbr_buf_with_header(THD *thd, const void *rbr_buf,
                                     size_t buf_len);
 
-int wsrep_binlog_close_connection(THD* thd);
-
 /**
    Write a skip event into binlog.
 
diff --git a/sql/wsrep_mysqld.h b/sql/wsrep_mysqld.h
index 71cbc875b91..a5da8e3bc44 100644
--- a/sql/wsrep_mysqld.h
+++ b/sql/wsrep_mysqld.h
@@ -405,11 +405,6 @@ extern void
 wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                           MDL_ticket *ticket,
                           const MDL_key *key);
-IO_CACHE * get_trans_log(THD * thd);
-bool wsrep_trans_cache_is_empty(THD *thd);
-void thd_binlog_flush_pending_rows_event(THD *thd, bool stmt_end);
-void thd_binlog_rollback_stmt(THD * thd);
-void thd_binlog_trx_reset(THD * thd);
 
 enum wsrep_thread_type {
   WSREP_APPLIER_THREAD=1,
diff --git a/sql/wsrep_trans_observer.h b/sql/wsrep_trans_observer.h
index 6bb26c40064..98fc63cf783 100644
--- a/sql/wsrep_trans_observer.h
+++ b/sql/wsrep_trans_observer.h
@@ -335,15 +335,22 @@ static inline int wsrep_before_rollback(THD* thd, bool all)
   int ret= 0;
   if (wsrep_is_active(thd))
   {
-    if (!all && thd->in_active_multi_stmt_transaction() &&
-        thd->wsrep_trx().is_streaming() &&
-        !wsrep_stmt_rollback_is_safe(thd))
+    if (!all && thd->in_active_multi_stmt_transaction())
     {
-      /* Non-safe statement rollback during SR multi statement
-         transasction. Self abort the transaction, the actual rollback
-         and error handling will be done in after statement phase. */
-      wsrep_thd_self_abort(thd);
-      ret= 0;
+      if (wsrep_emulate_bin_log)
+      {
+        wsrep_thd_binlog_stmt_rollback(thd);
+      }
+
+      if (thd->wsrep_trx().is_streaming() &&
+          !wsrep_stmt_rollback_is_safe(thd))
+      {
+        /* Non-safe statement rollback during SR multi statement
+           transasction. Self abort the transaction, the actual rollback
+           and error handling will be done in after statement phase. */
+        wsrep_thd_self_abort(thd);
+        ret= 0;
+      }
     }
     else if (wsrep_is_real(thd, all) &&
              thd->wsrep_trx().state() != wsrep::transaction::s_aborted)