MDEV-25114 Crash: WSREP: invalid state ROLLED_BACK (FATAL)

This patch is the plan D variant for fixing potetial mutex locking
order exercised by BF aborting and KILL command execution.

In this approach, KILL command is replicated as TOI operation.
This guarantees total isolation for the KILL command execution
in the first node: there is no concurrent replication applying
and no concurrent DDL executing. Therefore there is no risk of
BF aborting to happen in parallel with KILL command execution
either. Potential mutex deadlocks between the different mutex
access paths with KILL command execution and BF aborting cannot
therefore happen.

TOI replication is used, in this approach,  purely as means
to provide isolated KILL command execution in the first node.
KILL command should not (and must not) be applied in secondary
nodes. In this patch, we make this sure by skipping KILL
execution in secondary nodes, in applying phase, where we
bail out if applier thread is trying to execute KILL command.
This is effective, but skipping the applying of KILL command
could happen much earlier as well.

This patch also fixes mutex locking order and unprotected
THD member accesses on bf aborting case. We try to hold
THD::LOCK_thd_data during bf aborting. Only case where it
is not possible is at wsrep_abort_transaction before
call wsrep_innobase_kill_one_trx where we take InnoDB
mutexes first and then THD::LOCK_thd_data.

This will also fix possible race condition during
close_connection and while wsrep is disconnecting
connections.

Added wsrep_bf_kill_debug test case

Reviewed-by: Jan Lindström <jan.lindstrom@mariadb.com>

18 files changed, 725 insertions, 151 deletions

diff --git a/mysql-test/suite/galera/r/galera_UK_conflict.result b/mysql-test/suite/galera/r/galera_UK_conflict.result
index 76649f1b268..2795a86d6a6 100644
--- a/mysql-test/suite/galera/r/galera_UK_conflict.result
+++ b/mysql-test/suite/galera/r/galera_UK_conflict.result
@@ -68,6 +68,9 @@ f1	f2	f3
 10	10	0
 INSERT INTO t1 VALUES (7,7,7);
 INSERT INTO t1 VALUES (8,8,8);
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+7
 SELECT * FROM t1;
 f1	f2	f3
 1	1	0
@@ -78,6 +81,9 @@ f1	f2	f3
 8	8	8
 10	10	0
 connection node_1;
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+7
 SELECT * FROM t1;
 f1	f2	f3
 1	1	0
@@ -85,5 +91,6 @@ f1	f2	f3
 4	4	2
 5	5	2
 7	7	7
+8	8	8
 10	10	0
 DROP TABLE t1;
diff --git a/mysql-test/suite/galera/r/galera_bf_kill_debug.result b/mysql-test/suite/galera/r/galera_bf_kill_debug.result
new file mode 100644
index 00000000000..c20bb80131a
--- /dev/null
+++ b/mysql-test/suite/galera/r/galera_bf_kill_debug.result
@@ -0,0 +1,163 @@
+#
+# Case 1: We execute bf kill to wsrep_innobase_kill_one_trx
+# function just before wsrep_thd_LOCK(thd) call. Then we
+# try to kill victim transaction by KILL QUERY
+#
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+SET DEBUG_SYNC='wsrep_before_BF_victim_lock SIGNAL bf_kill WAIT_FOR bf_continue';
+ALTER TABLE t1 ADD UNIQUE KEY b1(b);;
+connection node_1;
+SET DEBUG_SYNC='now WAIT_FOR bf_kill';
+connection node_1b;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `b1` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+id	b
+1	1
+2	2
+3	3
+4	4
+5	5
+connection node_1;
+SET DEBUG_SYNC= 'RESET';
+DROP TABLE t1;
+disconnect node_1a;
+disconnect node_1b;
+disconnect node_1c;
+#
+# Case 2: We execute bf kill to wsrep_innobase_kill_one_trx
+# function just after wsrep_thd_LOCK(thd) call. Then we
+# try to kill victim transaction by KILL QUERY
+#
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+SET DEBUG_SYNC='wsrep_after_BF_victim_lock SIGNAL bf_kill WAIT_FOR bf_continue';
+ALTER TABLE t1 ADD UNIQUE KEY b1(b);;
+connection node_1;
+SET DEBUG_SYNC='now WAIT_FOR bf_kill';
+connection node_1b;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `b1` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+id	b
+1	1
+2	2
+3	3
+4	4
+5	5
+connection node_1;
+SET DEBUG_SYNC= 'RESET';
+DROP TABLE t1;
+disconnect node_1a;
+disconnect node_1b;
+disconnect node_1c;
+#
+# Case 3: Create victim transaction and try to send user KILL
+# from several threads
+#
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1d, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connection node_1b;
+connection node_1c;
+connection node_1d;
+connection node_1;
+disconnect node_1a;
+disconnect node_1b;
+disconnect node_1c;
+disconnect node_1d;
+DROP TABLE t1;
+#
+# Case 4: MDL-conflict, we execute ALTER until we hit gap in
+# wsrep_abort_transaction, while we are there we try to
+# manually KILL conflicting transaction (UPDATE) and
+# send conflicting transaction from other node to be executed
+# in this node by applier. As ALTER and KILL are TOI they
+# are not executed concurrently. Similarly UPDATE from other
+# node will wait for certification.
+#
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+SET DEBUG_SYNC='wsrep_abort_victim_unlocked SIGNAL bf_kill_unlocked WAIT_FOR bf_continue';
+ALTER TABLE t1 ADD UNIQUE KEY b1(b);;
+connection node_1;
+SET DEBUG_SYNC='now WAIT_FOR bf_kill_unlocked';
+connection node_1b;
+connection node_2;
+update t1 set b = b + 1000 where id between 2 and 4;;
+connection node_1;
+SET DEBUG_SYNC='now SIGNAL bf_continue';
+connection node_1c;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `b1` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT * FROM t1;
+id	b
+1	1
+5	5
+2	1002
+3	1003
+4	1004
+connection node_1b;
+connection node_1;
+SET DEBUG_SYNC= 'RESET';
+SELECT * FROM t1;
+id	b
+1	1
+5	5
+2	1002
+3	1003
+4	1004
+connection node_2;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `id` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `b1` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT * FROM t1;
+id	b
+1	1
+5	5
+2	1002
+3	1003
+4	1004
+DROP TABLE t1;
+disconnect node_1a;
+disconnect node_1c;
diff --git a/mysql-test/suite/galera/r/galera_toi_ddl_fk_insert.result b/mysql-test/suite/galera/r/galera_toi_ddl_fk_insert.result
index 94752ed7c76..a972394f87c 100644
--- a/mysql-test/suite/galera/r/galera_toi_ddl_fk_insert.result
+++ b/mysql-test/suite/galera/r/galera_toi_ddl_fk_insert.result
@@ -21,22 +21,6 @@ connection node_1a;
 connection node_1b;
 connection node_2;
 connection node_2a;
-connection node_1;
-SET SESSION wsrep_sync_wait=15;
-SELECT COUNT(*) FROM parent;
-COUNT(*)
-20001
-SELECT COUNT(*) FROM child;
-COUNT(*)
-10000
-connection node_2;
-SET SESSION wsrep_sync_wait=15;
-SELECT COUNT(*) FROM parent;
-COUNT(*)
-20001
-SELECT COUNT(*) FROM child;
-COUNT(*)
-10000
 DROP TABLE child;
 DROP TABLE parent;
 DROP TABLE ten;
diff --git a/mysql-test/suite/galera/t/galera_UK_conflict.test b/mysql-test/suite/galera/t/galera_UK_conflict.test
index 57bafbf8ae0..83d0e47dc3d 100644
--- a/mysql-test/suite/galera/t/galera_UK_conflict.test
+++ b/mysql-test/suite/galera/t/galera_UK_conflict.test
@@ -140,9 +140,13 @@ SELECT * FROM t1;
 # original state in node 1
 INSERT INTO t1 VALUES (7,7,7);
 INSERT INTO t1 VALUES (8,8,8);
+SELECT COUNT(*) FROM t1;
 SELECT * FROM t1;
 
 --connection node_1
+--let $wait_condition = SELECT COUNT(*) = 7 FROM t1
+--source include/wait_condition.inc
+SELECT COUNT(*) FROM t1;
 SELECT * FROM t1;
 
 DROP TABLE t1;
diff --git a/mysql-test/suite/galera/t/galera_bf_kill_debug.cnf b/mysql-test/suite/galera/t/galera_bf_kill_debug.cnf
new file mode 100644
index 00000000000..77bb6af9f35
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_bf_kill_debug.cnf
@@ -0,0 +1,9 @@
+!include ../galera_2nodes.cnf
+
+[mysqld.1]
+wsrep_log_conflicts=ON
+wsrep_debug=1
+
+[mysqld.2]
+wsrep_log_conflicts=ON
+wsrep_debug=1
diff --git a/mysql-test/suite/galera/t/galera_bf_kill_debug.test b/mysql-test/suite/galera/t/galera_bf_kill_debug.test
new file mode 100644
index 00000000000..f83d4a28ce9
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_bf_kill_debug.test
@@ -0,0 +1,283 @@
+--source include/galera_cluster.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+
+--echo #
+--echo # Case 1: We execute bf kill to wsrep_innobase_kill_one_trx
+--echo # function just before wsrep_thd_LOCK(thd) call. Then we
+--echo # try to kill victim transaction by KILL QUERY
+--echo #
+
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+
+#
+# This will be victim transaction for both bf kill and
+# user KILL
+#
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+
+#
+# Take thread id for above query
+#
+--connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--let $k_thread = `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'root' AND COMMAND = 'Sleep' LIMIT 1`
+
+#
+# Set DEBUG_SYNC and send conflicting DDL that will be TOI (bf) and
+# cause bf_kill
+#
+--connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1
+SET DEBUG_SYNC='wsrep_before_BF_victim_lock SIGNAL bf_kill WAIT_FOR bf_continue';
+--send ALTER TABLE t1 ADD UNIQUE KEY b1(b);
+
+#
+# Wait until we have reached the sync point
+#
+--connection node_1
+SET DEBUG_SYNC='now WAIT_FOR bf_kill';
+
+#
+# Try to kill update query
+#
+--connection node_1b
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
+
+
+#
+# Let bf_kill continue
+#
+--connection node_1
+SET DEBUG_SYNC='now SIGNAL bf_continue';
+--connection node_1c
+--reap
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+--connection node_1b
+--reap
+--enable_query_log
+
+--connection node_1
+SET DEBUG_SYNC= 'RESET';
+DROP TABLE t1;
+
+--disconnect node_1a
+--disconnect node_1b
+--disconnect node_1c
+
+--echo #
+--echo # Case 2: We execute bf kill to wsrep_innobase_kill_one_trx
+--echo # function just after wsrep_thd_LOCK(thd) call. Then we
+--echo # try to kill victim transaction by KILL QUERY
+--echo #
+
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+
+#
+# This will be victim transaction for both bf kill and
+# user KILL
+#
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+
+#
+# Take thread id for above query
+#
+--connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--let $k_thread = `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'root' AND COMMAND = 'Sleep' LIMIT 1`
+
+#
+# Set DEBUG_SYNC and send conflicting DDL that will be TOI (bf) and
+# cause bf_kill
+#
+--connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1
+SET DEBUG_SYNC='wsrep_after_BF_victim_lock SIGNAL bf_kill WAIT_FOR bf_continue';
+--send ALTER TABLE t1 ADD UNIQUE KEY b1(b);
+
+#
+# Wait until we have reached the sync point
+#
+--connection node_1
+SET DEBUG_SYNC='now WAIT_FOR bf_kill';
+
+#
+# Try to kill update query
+#
+--connection node_1b
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
+
+#
+# Let bf_kill continue
+#
+--connection node_1
+SET DEBUG_SYNC='now SIGNAL bf_continue';
+--connection node_1c
+--reap
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+--connection node_1b
+--reap
+--enable_query_log
+
+--connection node_1
+SET DEBUG_SYNC= 'RESET';
+DROP TABLE t1;
+
+--disconnect node_1a
+--disconnect node_1b
+--disconnect node_1c
+
+--echo #
+--echo # Case 3: Create victim transaction and try to send user KILL
+--echo # from several threads
+--echo #
+
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+
+#
+# This will be victim transaction for user KILL
+#
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+
+#
+# Take thread id for above query
+#
+--connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--connect node_1d, 127.0.0.1, root, , test, $NODE_MYPORT_1
+
+--connection node_1b
+--let $k_thread = `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'root' AND COMMAND = 'Sleep' LIMIT 1`
+
+#
+# Try to kill update query from several connections concurrently
+#
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
+
+--connection node_1c
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
+
+--connection node_1d
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
+
+#
+# We do not know execution order so any of these could fail as KILL
+# has been already done
+#
+--connection node_1b
+--enable_query_log
+--error 0,ER_KILL_DENIED_ERROR
+--reap
+--connection node_1c
+--enable_query_log
+--error 0,ER_KILL_DENIED_ERROR
+--reap
+--connection node_1d
+--enable_query_log
+--error 0,ER_KILL_DENIED_ERROR
+--reap
+
+--connection node_1
+--disconnect node_1a
+--disconnect node_1b
+--disconnect node_1c
+--disconnect node_1d
+DROP TABLE t1;
+
+--echo #
+--echo # Case 4: MDL-conflict, we execute ALTER until we hit gap in
+--echo # wsrep_abort_transaction, while we are there we try to
+--echo # manually KILL conflicting transaction (UPDATE) and
+--echo # send conflicting transaction from other node to be executed
+--echo # in this node by applier. As ALTER and KILL are TOI they
+--echo # are not executed concurrently. Similarly UPDATE from other
+--echo # node will wait for certification.
+--echo #
+
+CREATE TABLE t1(id int not null primary key, b int) engine=innodb;
+INSERT INTO t1 values (1,1),(2,2),(3,3),(4,4),(5,5);
+
+#
+# This will be victim transaction for both bf kill and
+# user KILL, and should not have any effect on result
+#
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+begin;
+update t1 set b = b * 10 where id between 2 and 4;
+
+#
+# Take thread id for above query
+#
+--connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--let $k_thread = `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'root' AND COMMAND = 'Sleep' LIMIT 1`
+
+#
+# Set DEBUG_SYNC and send conflicting DDL that will be TOI (bf) and
+# cause bf_kill but let's execute it only to gap in wsrep_abort_transaction
+#
+--connect node_1c, 127.0.0.1, root, , test, $NODE_MYPORT_1
+SET DEBUG_SYNC='wsrep_abort_victim_unlocked SIGNAL bf_kill_unlocked WAIT_FOR bf_continue';
+--send ALTER TABLE t1 ADD UNIQUE KEY b1(b);
+
+#
+# Wait until we have reached the sync point
+#
+--connection node_1
+SET DEBUG_SYNC='now WAIT_FOR bf_kill_unlocked';
+
+#
+# Try to kill update query
+#
+--connection node_1b
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
+
+#
+# Send conflicting update from other node, this should be applied on both nodes
+# but should not kill ALTER
+#
+--enable_query_log
+--connection node_2
+--send update t1 set b = b + 1000 where id between 2 and 4;
+
+#
+# Let bf_kill continue
+#
+--connection node_1
+SET DEBUG_SYNC='now SIGNAL bf_continue';
+--connection node_1c
+--reap
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+--connection node_1b
+--reap
+--enable_query_log
+
+--connection node_1
+SET DEBUG_SYNC= 'RESET';
+SELECT * FROM t1;
+
+--connection node_2
+--reap
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+DROP TABLE t1;
+
+--disconnect node_1a
+--disconnect node_1c
+
diff --git a/mysql-test/suite/galera/t/galera_toi_ddl_fk_insert.test b/mysql-test/suite/galera/t/galera_toi_ddl_fk_insert.test
index fadc94d78ff..3b4b427f551 100644
--- a/mysql-test/suite/galera/t/galera_toi_ddl_fk_insert.test
+++ b/mysql-test/suite/galera/t/galera_toi_ddl_fk_insert.test
@@ -54,15 +54,11 @@ INSERT INTO parent VALUES (1, 0);
 --connection node_2a
 --reap
 
---connection node_1
-SET SESSION wsrep_sync_wait=15;
-SELECT COUNT(*) FROM parent;
-SELECT COUNT(*) FROM child;
-
---connection node_2
-SET SESSION wsrep_sync_wait=15;
-SELECT COUNT(*) FROM parent;
-SELECT COUNT(*) FROM child;
+#
+# ALTER TABLE could bf kill one or more of INSERTs to parent, so
+# the actual number of rows in PARENT depends on whether
+# the INSERT is committed before ALTER TABLE is executed
+#
 
 DROP TABLE child;
 DROP TABLE parent;
diff --git a/mysql-test/suite/wsrep/t/variables.test b/mysql-test/suite/wsrep/t/variables.test
index 0cf13650ce0..875315c0e7c 100644
--- a/mysql-test/suite/wsrep/t/variables.test
+++ b/mysql-test/suite/wsrep/t/variables.test
@@ -66,7 +66,7 @@ call mtr.add_suppression("WSREP: Failed to get provider options");
 
 #evalp SET GLOBAL wsrep_provider= '$WSREP_PROVIDER';
 
---replace_regex /.*libgalera_smm.*/libgalera_smm.so/
+--replace_regex /.*libgalera.*/libgalera_smm.so/
 SELECT @@global.wsrep_provider;
 SELECT @@global.wsrep_slave_threads;
 SELECT @@global.wsrep_cluster_address;
@@ -77,7 +77,7 @@ SHOW STATUS LIKE 'wsrep_thread_count';
 
 #evalp SET GLOBAL wsrep_provider= '$WSREP_PROVIDER';
 
---replace_regex /.*libgalera_smm.*/libgalera_smm.so/
+--replace_regex /.*libgalera.*/libgalera_smm.so/
 SELECT @@global.wsrep_provider;
 SELECT @@global.wsrep_cluster_address;
 SELECT @@global.wsrep_on;
@@ -101,7 +101,7 @@ SELECT VARIABLE_VALUE AS EXPECT_1 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VA
 SELECT VARIABLE_VALUE AS EXPECT_1 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_rollbacker_thread_count';
 SELECT VARIABLE_VALUE AS EXPECT_2 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_thread_count';
 
---replace_regex /.*libgalera_smm.*/libgalera_smm.so/
+--replace_regex /.*libgalera.*/libgalera_smm.so/
 SELECT @@global.wsrep_provider;
 SELECT @@global.wsrep_cluster_address;
 SELECT @@global.wsrep_on;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 9afc701c6ba..269ad67165a 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -2391,7 +2391,7 @@ static void clean_up_mutexes()
 static void set_ports()
 {
 }
-void close_connection(THD *thd, uint sql_errno)
+void close_connection(THD *thd, uint sql_errno, my_bool locked)
 {
 }
 #else
@@ -2867,7 +2867,7 @@ static void network_init(void)
     For the connection that is doing shutdown, this is called twice
 */
 
-void close_connection(THD *thd, uint sql_errno)
+void close_connection(THD *thd, uint sql_errno, my_bool locked)
 {
   DBUG_ENTER("close_connection");
 
@@ -2877,7 +2877,10 @@ void close_connection(THD *thd, uint sql_errno)
   thd->print_aborted_warning(3, sql_errno ? ER_DEFAULT(sql_errno)
                                           : "CLOSE_CONNECTION");
 
-  thd->disconnect();
+  if (locked)
+    thd->disconnect_mutexed();
+  else
+    thd->disconnect();
 
   MYSQL_CONNECTION_DONE((int) sql_errno, thd->thread_id);
 
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 44b0491f138..1084173ae5f 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -83,7 +83,7 @@ enum enum_slave_parallel_mode {
 
 /* Function prototypes */
 void kill_mysql(THD *thd= 0);
-void close_connection(THD *thd, uint sql_errno= 0);
+void close_connection(THD *thd, uint sql_errno= 0, my_bool locked=false);
 void handle_connection_in_main_thread(CONNECT *thd);
 void create_thread_to_handle_connection(CONNECT *connect);
 void signal_thd_deleted();
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5ada018e540..05ec9a1c369 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1804,11 +1804,11 @@ void THD::awake(killed_state state_to_set)
           the Vio might be disassociated concurrently.
 */
 
-void THD::disconnect()
+void THD::disconnect_mutexed()
 {
   Vio *vio= NULL;
 
-  mysql_mutex_lock(&LOCK_thd_data);
+  mysql_mutex_assert_owner(&LOCK_thd_data);
 
   set_killed(KILL_CONNECTION);
 
@@ -1826,8 +1826,6 @@ void THD::disconnect()
   if (net.vio != vio)
     vio_close(net.vio);
   net.thd= 0;                                   // Don't collect statistics
-
-  mysql_mutex_unlock(&LOCK_thd_data);
 }
 
 
@@ -1879,16 +1877,18 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
           thread can see those instances (e.g. see partitioning code).
         */
         if (!thd_table->needs_reopen())
-        {
           signalled|= mysql_lock_abort_for_thread(this, thd_table);
-          if (WSREP(this) && wsrep_thd_is_BF(this, FALSE))
-          {
-            WSREP_DEBUG("remove_table_from_cache: %llu",
-                        (unsigned long long) this->real_id);
-            wsrep_abort_thd((void *)this, (void *)in_use, FALSE);
-          }
-        }
       }
+#ifdef WITH_WSREP
+      if (WSREP(this) && wsrep_thd_is_BF(this, false))
+      {
+        WSREP_DEBUG("notify_shared_lock: BF thread %llu query %s"
+                    " victim %llu query %s",
+                    this->real_id, wsrep_thd_query(this),
+                    in_use->real_id, wsrep_thd_query(in_use));
+        wsrep_abort_thd((void *)this, (void *)in_use, false);
+      }
+#endif /* WITH_WSREP */
     }
     mysql_mutex_unlock(&in_use->LOCK_thd_data);
   }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 5f871f9caf6..7dbc81be770 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3231,8 +3231,13 @@ public:
   void awake(killed_state state_to_set);
  
   /** Disconnect the associated communication endpoint. */
-  void disconnect();
-
+  inline void disconnect()
+  {
+    mysql_mutex_lock(&LOCK_thd_data);
+    disconnect_mutexed();
+    mysql_mutex_unlock(&LOCK_thd_data);
+  }
+  void disconnect_mutexed();
 
   /*
     Allows this thread to serve as a target for others to schedule Async 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 3e1f248b082..2bec9c6b6cd 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2020, MariaDB
+   Copyright (c) 2008, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -9069,6 +9069,18 @@ static
 void sql_kill(THD *thd, longlong id, killed_state state, killed_type type)
 {
   uint error;
+#ifdef WITH_WSREP
+  if (WSREP(thd))
+  {
+    WSREP_DEBUG("sql_kill called");
+    if (thd->wsrep_applier)
+    {
+      WSREP_DEBUG("KILL in applying, bailing out here");
+      return;
+    }
+    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL)
+  }
+#endif /* WITH_WSREP */
   if (!(error= kill_one_thread(thd, id, state, type)))
   {
     if (!thd->killed)
@@ -9078,6 +9090,11 @@ void sql_kill(THD *thd, longlong id, killed_state state, killed_type type)
   }
   else
     my_error(error, MYF(0), id);
+#ifdef WITH_WSREP
+  return;
+ wsrep_error_label:
+  my_error(ER_CANNOT_USER, MYF(0), wsrep_thd_query(thd));
+#endif /* WITH_WSREP */
 }
 
 
@@ -9086,6 +9103,18 @@ void sql_kill_user(THD *thd, LEX_USER *user, killed_state state)
 {
   uint error;
   ha_rows rows;
+#ifdef WITH_WSREP
+  if (WSREP(thd))
+  {
+    WSREP_DEBUG("sql_kill_user called");
+    if (thd->wsrep_applier)
+    {
+      WSREP_DEBUG("KILL in applying, bailing out here");
+      return;
+    }
+    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL)
+  }
+#endif /* WITH_WSREP */
   if (!(error= kill_threads_for_user(thd, user, state, &rows)))
     my_ok(thd, rows);
   else
@@ -9096,6 +9125,11 @@ void sql_kill_user(THD *thd, LEX_USER *user, killed_state state)
     */
     my_error(error, MYF(0), user->host.str, user->user.str);
   }
+#ifdef WITH_WSREP
+  return;
+ wsrep_error_label:
+  my_error(ER_CANNOT_USER, MYF(0), user->user.str);
+#endif /* WITH_WSREP */
 }
 
 
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index e60100e2e90..00a6dfe2f8a 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -1,4 +1,4 @@
-/* Copyright 2008-2015 Codership Oy <http://www.codership.com>
+/* Copyright 2008-2021 Codership Oy <http://www.codership.com>
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -835,13 +835,25 @@ void wsrep_thr_init()
   DBUG_VOID_RETURN;
 }
 
+/* This is wrapper for wsrep_break_lock in thr_lock.c */
+static int wsrep_thr_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
+{
+  THD* victim_thd= (THD *) victim_thd_ptr;
+  /* We need to lock THD::LOCK_thd_data to protect victim
+  from concurrent usage or disconnect or delete. */
+  mysql_mutex_lock(&victim_thd->LOCK_thd_data);
+  int res= wsrep_abort_thd(bf_thd_ptr, victim_thd_ptr, signal);
+  mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+  return res;
+}
+
 void wsrep_init_startup (bool first)
 {
   if (wsrep_init()) unireg_abort(1);
 
   wsrep_thr_lock_init(
      (wsrep_thd_is_brute_force_fun)wsrep_thd_is_BF,
-     (wsrep_abort_thd_fun)wsrep_abort_thd,
+     (wsrep_abort_thd_fun)wsrep_thr_abort_thd,
      wsrep_debug, wsrep_convert_LOCK_to_trx,
      (wsrep_on_fun)wsrep_on);
 
@@ -1694,6 +1706,11 @@ static int wsrep_TOI_begin(THD *thd, char *db_, char *table_,
   case SQLCOM_DROP_TABLE:
     buf_err= wsrep_drop_table_query(thd, &buf, &buf_len);
     break;
+  case SQLCOM_KILL:
+    WSREP_DEBUG("KILL as TOI: %s", thd->query());
+    buf_err= wsrep_to_buf_helper(thd, thd->query(), thd->query_length(),
+                                 &buf, &buf_len);
+    break;
   case SQLCOM_CREATE_ROLE:
     if (sp_process_definer(thd))
     {
@@ -2058,8 +2075,11 @@ bool wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
         ticket->wsrep_report(true);
       }
 
-      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+      /* This will call wsrep_abort_transaction so we should hold
+      THD::LOCK_thd_data to protect victim from concurrent usage
+      or disconnect or delete. */
       wsrep_abort_thd((void *) request_thd, (void *) granted_thd, 1);
+      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       ret= false;
     }
   }
@@ -2241,6 +2261,7 @@ error:
 static bool abort_replicated(THD *thd)
 {
   bool ret_code= false;
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   if (thd->wsrep_query_state== QUERY_COMMITTING)
   {
     WSREP_DEBUG("aborting replicated trx: %llu", (ulonglong)(thd->real_id));
@@ -2255,6 +2276,7 @@ static bool abort_replicated(THD *thd)
 /**/
 static inline bool is_client_connection(THD *thd)
 {
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   return (thd->wsrep_client_thread && thd->variables.wsrep_on);
 }
 
@@ -2263,9 +2285,8 @@ static inline bool is_replaying_connection(THD *thd)
 {
   bool ret;
 
-  mysql_mutex_lock(&thd->LOCK_thd_data);
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   ret=  (thd->wsrep_conflict_state == REPLAYING) ? true : false;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
 
   return ret;
 }
@@ -2275,9 +2296,8 @@ static inline bool is_committing_connection(THD *thd)
 {
   bool ret;
 
-  mysql_mutex_lock(&thd->LOCK_thd_data);
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   ret=  (thd->wsrep_query_state == QUERY_COMMITTING) ? true : false;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
 
   return ret;
 }
@@ -2290,13 +2310,17 @@ static bool have_client_connections()
   I_List_iterator<THD> it(threads);
   while ((tmp=it++))
   {
+    /* Protect thread from concurrent usage or disconnect or delete. */
+    mysql_mutex_lock(&tmp->LOCK_thd_data);
     DBUG_PRINT("quit",("Informing thread %lld that it's time to die",
                        (longlong) tmp->thread_id));
     if (is_client_connection(tmp) && tmp->killed == KILL_CONNECTION)
     {
       (void)abort_replicated(tmp);
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       return true;
     }
+    mysql_mutex_unlock(&tmp->LOCK_thd_data);
   }
   return false;
 }
@@ -2328,14 +2352,21 @@ static my_bool have_committing_connections()
   I_List_iterator<THD> it(threads);
   while ((tmp=it++))
   {
+    /* Protect from concurrent usage or disconnect or delete */
+    mysql_mutex_lock(&tmp->LOCK_thd_data);
     if (!is_client_connection(tmp))
+    {
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       continue;
+    }
 
     if (is_committing_connection(tmp))
     {
       mysql_mutex_unlock(&LOCK_thread_count);
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       return TRUE;
     }
+    mysql_mutex_unlock(&tmp->LOCK_thd_data);
   }
   mysql_mutex_unlock(&LOCK_thread_count);
   return FALSE;
@@ -2378,33 +2409,44 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
   {
     DBUG_PRINT("quit",("Informing thread %lld that it's time to die",
                        (longlong) tmp->thread_id));
+    /* Protect from concurrent usage or disconnect or delete */
+    mysql_mutex_lock(&tmp->LOCK_thd_data);
     /* We skip slave threads & scheduler on this first loop through. */
     if (!is_client_connection(tmp))
+    {
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       continue;
+    }
 
     if (tmp == except_caller_thd)
     {
       DBUG_ASSERT(is_client_connection(tmp));
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       continue;
     }
 
     if (is_replaying_connection(tmp))
     {
       tmp->set_killed(KILL_CONNECTION);
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       continue;
     }
 
-    /* replicated transactions must be skipped */
+    /* replicated transactions must be skipped and aborted
+    with wsrep_abort_thd. */
     if (abort_replicated(tmp))
+    {
+      mysql_mutex_unlock(&tmp->LOCK_thd_data);
       continue;
+    }
 
     WSREP_DEBUG("closing connection %lld", (longlong) tmp->thread_id);
 
     /*
-      instead of wsrep_close_thread() we do now  soft kill by THD::awake
-     */
-    mysql_mutex_lock(&tmp->LOCK_thd_data);
-
+      instead of wsrep_close_thread() we do now  soft kill by
+      THD::awake(). Here also victim needs to be protected from
+      concurrent usage or disconnect or delete.
+    */
     tmp->awake(KILL_CONNECTION);
 
     mysql_mutex_unlock(&tmp->LOCK_thd_data);
@@ -2423,16 +2465,19 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
   I_List_iterator<THD> it2(threads);
   while ((tmp=it2++))
   {
-#ifndef __bsdi__				// Bug in BSDI kernel
-    if (is_client_connection(tmp) &&
-        !abort_replicated(tmp)    &&
-        !is_replaying_connection(tmp) &&
-        tmp != except_caller_thd)
+    /* Protect from concurrent usage or disconnect or delete */
+    mysql_mutex_lock(&tmp->LOCK_thd_data);
+    if (is_client_connection(tmp))
     {
-      WSREP_INFO("killing local connection: %lld", (longlong) tmp->thread_id);
-      close_connection(tmp,0);
+      if (!abort_replicated(tmp) &&
+          !is_replaying_connection(tmp) &&
+          tmp != except_caller_thd)
+      {
+        WSREP_INFO("killing local connection: %lld", (longlong) tmp->thread_id);
+        close_connection(tmp,0, true);
+      }
     }
-#endif
+    mysql_mutex_unlock(&tmp->LOCK_thd_data);
   }
 
   DBUG_PRINT("quit",("Waiting for threads to die (count=%u)",thread_count));
@@ -2621,7 +2666,9 @@ extern "C" void wsrep_thd_set_query_state(
 
 void wsrep_thd_set_conflict_state(THD *thd, enum wsrep_conflict_state state)
 {
-  if (WSREP(thd)) thd->wsrep_conflict_state= state;
+  DBUG_ASSERT(thd);
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
+  thd->wsrep_conflict_state= state;
 }
 
 
@@ -2762,9 +2809,10 @@ extern "C" void wsrep_thd_awake(THD *thd, my_bool signal)
 {
   if (signal)
   {
-    mysql_mutex_lock(&thd->LOCK_thd_data);
+    /* Here we should hold THD::LOCK_thd_data to
+    protect from concurrent usage. */
+    mysql_mutex_assert_owner(&thd->LOCK_thd_data);
     thd->awake(KILL_QUERY);
-    mysql_mutex_unlock(&thd->LOCK_thd_data);
   }
   else
   {
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index ef8c0e132f7..d849cd2cd78 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2013 Codership Oy <info@codership.com>
+/* Copyright (C) 2013-2021 Codership Oy <info@codership.com>
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -804,10 +804,12 @@ my_bool wsrep_thd_is_local(void *thd_ptr, my_bool sync)
 
 int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
 {
-  THD *victim_thd = (THD *) victim_thd_ptr;
-  THD *bf_thd     = (THD *) bf_thd_ptr;
+  THD *victim_thd= (THD *) victim_thd_ptr;
+  THD *bf_thd= (THD *) bf_thd_ptr;
   DBUG_ENTER("wsrep_abort_thd");
 
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+
   if ( (WSREP(bf_thd) ||
          ( (WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU) &&
            bf_thd->wsrep_exec_mode == TOTAL_ORDER) )                         &&
diff --git a/sql/wsrep_thd.h b/sql/wsrep_thd.h
index 46bc08a466a..1284cad1acf 100644
--- a/sql/wsrep_thd.h
+++ b/sql/wsrep_thd.h
@@ -32,7 +32,6 @@ void wsrep_create_rollbacker();
 
 int  wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr,
                                 my_bool signal);
-
 /*
   PA = Parallel Applying (on the slave side)
 */
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index d4142481a62..efcdd201fa6 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -5233,17 +5233,21 @@ UNIV_INTERN void lock_cancel_waiting_and_release(lock_t* lock);
 @sa THD::awake() @sa ha_kill_query() */
 static void innobase_kill_query(handlerton*, THD* thd, enum thd_kill_levels)
 {
-	DBUG_ENTER("innobase_kill_query");
+  DBUG_ENTER("innobase_kill_query");
 #ifdef WITH_WSREP
-	if (wsrep_thd_get_conflict_state(thd) != NO_CONFLICT) {
-		/* if victim has been signaled by BF thread and/or aborting
-		   is already progressing, following query aborting is not necessary
-		   any more.
-		   Also, BF thread should own trx mutex for the victim, which would
-		   conflict with trx_mutex_enter() below
-		*/
-		DBUG_VOID_RETURN;
-	}
+  if (wsrep_thd_get_conflict_state(thd) != NO_CONFLICT)
+  {
+    /* if victim has been signaled by BF thread and/or aborting
+    is already progressing, following query aborting is not necessary
+    any more. E.g. wsrep_innobase_kill_one_trx().
+    Also, BF thread should own trx mutex for the victim, which would
+    conflict with trx_mutex_enter() below
+    */
+    WSREP_DEBUG("Victim thread %ld bail out conflict_state %s query %s",
+		thd_get_thread_id(thd),
+		wsrep_thd_conflict_state_str(thd), wsrep_thd_query(thd));
+    DBUG_VOID_RETURN;
+  }
 #endif /* WITH_WSREP */
 
   if (trx_t* trx= thd_to_trx(thd))
@@ -19520,9 +19524,9 @@ static struct st_mysql_storage_engine innobase_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };
 
 #ifdef WITH_WSREP
+static
 void
 wsrep_abort_slave_trx(
-/*==================*/
 	wsrep_seqno_t bf_seqno,
 	wsrep_seqno_t victim_seqno)
 {
@@ -19532,19 +19536,17 @@ wsrep_abort_slave_trx(
 		"2) a bug in the code.\n\t"
 		"3) a database corruption.\n Node consistency compromized, "
 		"need to abort. Restart the node to resync with cluster.",
-		(long long)bf_seqno, (long long)victim_seqno);
+		bf_seqno, victim_seqno);
 	abort();
 }
 /*******************************************************************//**
 This function is used to kill one transaction in BF. */
-UNIV_INTERN
 void
 wsrep_innobase_kill_one_trx(
-/*========================*/
 	MYSQL_THD const bf_thd,
 	const trx_t * const bf_trx,
 	trx_t *victim_trx,
-	ibool signal)
+	my_bool signal)
 {
         ut_ad(bf_thd);
         ut_ad(victim_trx);
@@ -19552,38 +19554,41 @@ wsrep_innobase_kill_one_trx(
         ut_ad(trx_mutex_own(victim_trx));
 
 	DBUG_ENTER("wsrep_innobase_kill_one_trx");
-	THD *thd          = (THD *) victim_trx->mysql_thd;
-	int64_t bf_seqno  = wsrep_thd_trx_seqno(bf_thd);
+	THD *thd= (THD *) victim_trx->mysql_thd;
+	int64_t bf_seqno= wsrep_thd_trx_seqno(bf_thd);
 
 	if (!thd) {
-		DBUG_PRINT("wsrep", ("no thd for conflicting lock"));
 		WSREP_WARN("no THD for trx: " TRX_ID_FMT, victim_trx->id);
 		DBUG_VOID_RETURN;
 	}
 
-	WSREP_LOG_CONFLICT(bf_thd, thd, TRUE);
-
-	WSREP_DEBUG("BF kill (" ULINTPF ", seqno: " INT64PF
-		    "), victim: (%lu) trx: " TRX_ID_FMT,
-		    signal, bf_seqno,
-		    thd_get_thread_id(thd),
-		    victim_trx->id);
-
-	WSREP_DEBUG("Aborting query: %s conf %d trx: %" PRId64,
-		    (thd && wsrep_thd_query(thd)) ? wsrep_thd_query(thd) : "void",
-		    wsrep_thd_conflict_state(thd, FALSE),
-		    wsrep_thd_ws_handle(thd)->trx_id);
-
+	/* Here we need to lock THD::LOCK_thd_data to protect from
+	concurrent usage or disconnect or delete. */
+	DEBUG_SYNC(bf_thd, "wsrep_before_BF_victim_lock");
 	wsrep_thd_LOCK(thd);
-        DBUG_EXECUTE_IF("sync.wsrep_after_BF_victim_lock",
-                 {
-                   const char act[]=
-                     "now "
-                     "wait_for signal.wsrep_after_BF_victim_lock";
-                   DBUG_ASSERT(!debug_sync_set_action(bf_thd,
-                                                      STRING_WITH_LEN(act)));
-                 };);
+	DEBUG_SYNC(bf_thd, "wsrep_after_BF_victim_lock");
+
+	WSREP_DEBUG("Aborter %s trx_id: " TRX_ID_FMT " thread: %ld "
+		    "seqno: %lld query_state: %s conflict_state: %s query: %s",
+		    wsrep_thd_is_BF(bf_thd, false) ? "BF" : "normal",
+		    bf_trx ? bf_trx->id : TRX_ID_MAX,
+		    thd_get_thread_id(bf_thd),
+		    bf_seqno,
+		    wsrep_thd_query_state_str(bf_thd),
+		    wsrep_thd_conflict_state_str(bf_thd),
+		    wsrep_thd_query(bf_thd));
+
+	WSREP_DEBUG("Victim %s trx_id: " TRX_ID_FMT " thread: %ld "
+		    "seqno: %lld query_state: %s  conflict_state: %s query: %s",
+		    wsrep_thd_is_BF(thd, false) ? "BF" : "normal",
+		    victim_trx->id,
+		    thd_get_thread_id(thd),
+		    wsrep_thd_trx_seqno(thd),
+		    wsrep_thd_query_state_str(thd),
+		    wsrep_thd_conflict_state_str(thd),
+		    wsrep_thd_query(thd));
 
+	WSREP_LOG_CONFLICT(bf_thd, thd, TRUE);
 
 	if (wsrep_thd_query_state(thd) == QUERY_EXITING) {
 		WSREP_DEBUG("kill trx EXITING for " TRX_ID_FMT,
@@ -19593,27 +19598,32 @@ wsrep_innobase_kill_one_trx(
 	}
 
 	if (wsrep_thd_exec_mode(thd) != LOCAL_STATE) {
-		WSREP_DEBUG("withdraw for BF trx: " TRX_ID_FMT ", state: %d",
+		WSREP_DEBUG("withdraw for BF trx: " TRX_ID_FMT
+			    ", state: %s exec %s",
 			    victim_trx->id,
-		wsrep_thd_get_conflict_state(thd));
+			    wsrep_thd_conflict_state_str(thd),
+			    wsrep_thd_exec_mode_str(thd));
 	}
 
 	switch (wsrep_thd_get_conflict_state(thd)) {
 	case NO_CONFLICT:
+		/* This will cause any call to innobase_kill_query()
+		for this thd to bail out. */
 		wsrep_thd_set_conflict_state(thd, MUST_ABORT);
 		break;
         case MUST_ABORT:
 		WSREP_DEBUG("victim " TRX_ID_FMT " in MUST ABORT state",
 			    victim_trx->id);
-		wsrep_thd_UNLOCK(thd);
 		wsrep_thd_awake(thd, signal);
+		wsrep_thd_UNLOCK(thd);
 		DBUG_VOID_RETURN;
 		break;
 	case ABORTED:
 	case ABORTING: // fall through
 	default:
-		WSREP_DEBUG("victim " TRX_ID_FMT " in state %d",
-			    victim_trx->id, wsrep_thd_get_conflict_state(thd));
+		WSREP_DEBUG("victim " TRX_ID_FMT " in state %s",
+			    victim_trx->id,
+			    wsrep_thd_conflict_state_str(thd));
 		wsrep_thd_UNLOCK(thd);
 		DBUG_VOID_RETURN;
 		break;
@@ -19643,8 +19653,8 @@ wsrep_innobase_kill_one_trx(
 				WSREP_DEBUG("cancel commit warning: "
 					    TRX_ID_FMT,
 					    victim_trx->id);
-				wsrep_thd_UNLOCK(thd);
 				wsrep_thd_awake(thd, signal);
+				wsrep_thd_UNLOCK(thd);
 				DBUG_VOID_RETURN;
 				break;
 			case WSREP_OK:
@@ -19662,8 +19672,8 @@ wsrep_innobase_kill_one_trx(
 				break;
 			}
 		}
-		wsrep_thd_UNLOCK(thd);
 		wsrep_thd_awake(thd, signal);
+		wsrep_thd_UNLOCK(thd);
 		break;
 	case QUERY_EXEC:
 		/* it is possible that victim trx is itself waiting for some
@@ -19685,23 +19695,19 @@ wsrep_innobase_kill_one_trx(
 				lock_cancel_waiting_and_release(wait_lock);
 			}
 
-			wsrep_thd_UNLOCK(thd);
 			wsrep_thd_awake(thd, signal);
+			wsrep_thd_UNLOCK(thd);
 		} else {
 			/* abort currently executing query */
-			DBUG_PRINT("wsrep",("sending KILL_QUERY to: %lu",
-                                            thd_get_thread_id(thd)));
 			WSREP_DEBUG("kill query for: %ld",
 				thd_get_thread_id(thd));
-			/* Note that innobase_kill_query will take lock_mutex
-			and trx_mutex */
-			wsrep_thd_UNLOCK(thd);
 			wsrep_thd_awake(thd, signal);
+			wsrep_thd_UNLOCK(thd);
 
 			/* for BF thd, we need to prevent him from committing */
 			if (wsrep_thd_exec_mode(thd) == REPL_RECV) {
 				wsrep_abort_slave_trx(bf_seqno,
-						    wsrep_thd_trx_seqno(thd));
+						      wsrep_thd_trx_seqno(thd));
 			}
 		}
 		break;
@@ -19711,29 +19717,27 @@ wsrep_innobase_kill_one_trx(
 
 		if (wsrep_thd_exec_mode(thd) == REPL_RECV) {
 			WSREP_DEBUG("kill BF IDLE, seqno: %lld",
-				    (long long)wsrep_thd_trx_seqno(thd));
+				    wsrep_thd_trx_seqno(thd));
 			wsrep_thd_UNLOCK(thd);
 			wsrep_abort_slave_trx(bf_seqno,
 					      wsrep_thd_trx_seqno(thd));
 			DBUG_VOID_RETURN;
 		}
-                /* This will lock thd from proceeding after net_read() */
+		/* This will lock thd from proceeding after net_read()
+		and innobase_kill_query to bail out for this thd. */
 		wsrep_thd_set_conflict_state(thd, ABORTING);
 
 		wsrep_lock_rollback();
 
 		if (wsrep_aborting_thd_contains(thd)) {
 			WSREP_WARN("duplicate thd aborter %lu",
-			           (ulong) thd_get_thread_id(thd));
+			           thd_get_thread_id(thd));
 		} else {
 			wsrep_aborting_thd_enqueue(thd);
-			DBUG_PRINT("wsrep",("enqueuing trx abort for %lu",
-			                    thd_get_thread_id(thd)));
 			WSREP_DEBUG("enqueuing trx abort for (%lu)",
-			            thd_get_thread_id(thd));
+				    thd_get_thread_id(thd));
 		}
 
-		DBUG_PRINT("wsrep",("signalling wsrep rollbacker"));
 		WSREP_DEBUG("signaling aborter");
 		wsrep_unlock_rollback();
 		wsrep_thd_UNLOCK(thd);
@@ -19741,10 +19745,7 @@ wsrep_innobase_kill_one_trx(
 		break;
 	}
 	default:
-		WSREP_WARN("bad wsrep query state: %d",
-			  wsrep_thd_query_state(thd));
-		wsrep_thd_UNLOCK(thd);
-		break;
+		ut_error;
 	}
 
 	DBUG_VOID_RETURN;
@@ -19753,7 +19754,6 @@ wsrep_innobase_kill_one_trx(
 static
 void
 wsrep_abort_transaction(
-/*====================*/
 	handlerton* hton,
 	THD *bf_thd,
 	THD *victim_thd,
@@ -19761,27 +19761,65 @@ wsrep_abort_transaction(
 {
 	DBUG_ENTER("wsrep_abort_transaction");
 
-	trx_t* victim_trx	= thd_to_trx(victim_thd);
-	trx_t* bf_trx		= (bf_thd) ? thd_to_trx(bf_thd) : NULL;
-
-	WSREP_DEBUG("abort transaction: BF: %s victim: %s victim conf: %d",
-			wsrep_thd_query(bf_thd),
-			wsrep_thd_query(victim_thd),
-			wsrep_thd_conflict_state(victim_thd, FALSE));
+	ut_ad(bf_thd);
+	ut_ad(victim_thd);
+	trx_t* victim_trx= thd_to_trx(victim_thd);
+	trx_t* bf_trx= thd_to_trx(bf_thd);
+
+	/* Here we should hold THD::LOCK_thd_data to protect
+	victim from concurrent usage or disconnect or delete. */
+	WSREP_DEBUG("wsrep_abort_transaction: BF:"
+		    " thread %ld query_state %s conflict_state %s"
+		    " exec %s query %s trx " TRX_ID_FMT,
+		    thd_get_thread_id(bf_thd),
+		    wsrep_thd_query_state_str(bf_thd),
+		    wsrep_thd_conflict_state_str(bf_thd),
+		    wsrep_thd_exec_mode_str(bf_thd),
+		    wsrep_thd_query(bf_thd),
+		    bf_trx ? bf_trx->id : 0);
+
+	WSREP_DEBUG("wsrep_abort_transaction: victim:"
+		    " thread %ld query_state %s conflict_state %s"
+		    " exec %s query %s trx " TRX_ID_FMT,
+		    thd_get_thread_id(victim_thd),
+		    wsrep_thd_query_state_str(victim_thd),
+		    wsrep_thd_conflict_state_str(victim_thd),
+		    wsrep_thd_exec_mode_str(victim_thd),
+		    wsrep_thd_query(victim_thd),
+		    victim_trx ? victim_trx->id : 0);
 
 	if (victim_trx) {
+		WSREP_DEBUG("wsrep_abort_transaction: Victim thread %ld "
+			    "transaction " TRX_ID_FMT " trx_state %d",
+			    thd_get_thread_id(victim_thd),
+			    victim_trx->id,
+			    victim_trx->state);
+		/* This is necessary as correct mutexing order is
+		lock_sys -> trx -> THD::LOCK_thd_data and below
+		function assumes we have lock_sys and trx locked
+		and takes THD::LOCK_thd_data for THD state check. */
+		wsrep_thd_UNLOCK(victim_thd);
+		DEBUG_SYNC(bf_thd, "wsrep_abort_victim_unlocked");
+		DBUG_EXECUTE_IF("wsrep_abort_replicated_sleep",
+				WSREP_DEBUG("wsrep_abort_transaction: sleeping "
+					    "for thread %ld ",
+					    thd_get_thread_id(victim_thd));
+				my_sleep(100000););
 		lock_mutex_enter();
 		trx_mutex_enter(victim_trx);
 		wsrep_innobase_kill_one_trx(bf_thd, bf_trx, victim_trx, signal);
 		lock_mutex_exit();
 		trx_mutex_exit(victim_trx);
 		wsrep_srv_conc_cancel_wait(victim_trx);
+		wsrep_thd_LOCK(victim_thd);
 		DBUG_VOID_RETURN;
 	} else {
-		WSREP_DEBUG("victim does not have transaction");
-		wsrep_thd_LOCK(victim_thd);
+		WSREP_DEBUG("wsrep_abort_transaction: Victim thread %ld "
+			    "no transaction",
+			    thd_get_thread_id(victim_thd));
+		/* This will cause any call to innobase_kill_query()
+		for this thd to bail out. */
 		wsrep_thd_set_conflict_state(victim_thd, MUST_ABORT);
-		wsrep_thd_UNLOCK(victim_thd);
 		wsrep_thd_awake(victim_thd, signal);
 	}
 
diff --git a/storage/innobase/include/ha_prototypes.h b/storage/innobase/include/ha_prototypes.h
index 3eab2135969..427e57f09d2 100644
--- a/storage/innobase/include/ha_prototypes.h
+++ b/storage/innobase/include/ha_prototypes.h
@@ -233,12 +233,11 @@ innobase_casedn_str(
 	char*	a);	/*!< in/out: string to put in lower case */
 
 #ifdef WITH_WSREP
-UNIV_INTERN
 void
 wsrep_innobase_kill_one_trx(MYSQL_THD const thd_ptr,
                             const trx_t * const bf_trx,
                             trx_t *victim_trx,
-                            ibool signal);
+                            my_bool signal);
 int wsrep_innobase_mysql_sort(int mysql_type, uint charset_number,
                              unsigned char* str, unsigned int str_length,
                              unsigned int buf_length);