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

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>

21 files changed, 640 insertions, 332 deletions

diff --git a/mysql-test/suite/galera/r/galera_UK_conflict.result b/mysql-test/suite/galera/r/galera_UK_conflict.result
index 57b9c1279ea..48a7e372b25 100644
--- a/mysql-test/suite/galera/r/galera_UK_conflict.result
+++ b/mysql-test/suite/galera/r/galera_UK_conflict.result
@@ -68,9 +68,9 @@ f1	f2	f3
 10	10	0
 INSERT INTO t1 VALUES (7,7,7);
 INSERT INTO t1 VALUES (8,8,8);
-DROP TABLE t1;
-test scenario 2
-connection node_1;
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+7
 CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 int, f3 int,  unique key keyj (f2));
 INSERT INTO t1 VALUES (1, 1, 0);
 INSERT INTO t1 VALUES (3, 3, 0);
@@ -92,9 +92,9 @@ SET SESSION wsrep_on = 1;
 SET GLOBAL wsrep_provider_options = 'dbug=';
 SET GLOBAL wsrep_provider_options = 'dbug=d,commit_monitor_master_enter_sync';
 connection node_1;
-COMMIT;
-connection node_1a;
-SET SESSION wsrep_on = 0;
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+7
 SET SESSION wsrep_on = 1;
 SET GLOBAL wsrep_provider_options = 'dbug=';
 SET GLOBAL DEBUG_DBUG = "d,sync.wsrep_replay_cb";
@@ -126,6 +126,7 @@ f1	f2	f3
 3	3	1
 4	4	2
 5	5	2
+8	8	8
 10	10	0
 INSERT INTO t1 VALUES (7,7,7);
 INSERT INTO t1 VALUES (8,8,8);
diff --git a/mysql-test/suite/galera/r/galera_bf_kill_debug.result b/mysql-test/suite/galera/r/galera_bf_kill_debug.result
index c3eae243f47..2c7227c25c7 100644
--- a/mysql-test/suite/galera/r/galera_bf_kill_debug.result
+++ b/mysql-test/suite/galera/r/galera_bf_kill_debug.result
@@ -1,54 +1,165 @@
 connection node_2;
 connection node_1;
-connection node_2;
-CREATE TABLE t1(a int not null primary key auto_increment,b int) engine=InnoDB;
-insert into t1 values (NULL,1);
-connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2;
-connection node_2a;
-truncate t1;
-insert into t1 values (1,0);
+#
+# 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=2 where a=1;
-connection node_2;
-set session wsrep_sync_wait=0;
-connect node_2b, 127.0.0.1, root, , test, $NODE_MYPORT_2;
-connection node_2b;
-SET GLOBAL debug_dbug = "d,sync.before_wsrep_thd_abort";
-connection node_1;
-select * from t1;
-a	b
-1	0
-update t1 set b= 1 where a=1;
-connection node_2b;
-SET SESSION DEBUG_SYNC = "now WAIT_FOR sync.before_wsrep_thd_abort_reached";
-connection node_2;
-SET DEBUG_SYNC= 'before_awake_no_mutex SIGNAL awake_reached WAIT_FOR continue_kill';
-connection node_2b;
-SET DEBUG_SYNC='now WAIT_FOR awake_reached';
-SET GLOBAL debug_dbug = "";
-SET DEBUG_SYNC = "now SIGNAL signal.before_wsrep_thd_abort";
-SET DEBUG_SYNC = "now SIGNAL continue_kill";
-connection node_2;
-connection node_2a;
-select * from t1;
-connection node_2;
-SET DEBUG_SYNC = "RESET";
-drop table t1;
-disconnect node_2a;
-connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2;
-connection node_2a;
-CREATE TABLE t1 (i int primary key);
-SET DEBUG_SYNC = "before_wsrep_ordered_commit SIGNAL bwoc_reached WAIT_FOR bwoc_continue";
-INSERT INTO t1 VALUES (1);
-connection node_2;
-SET DEBUG_SYNC = "now WAIT_FOR bwoc_reached";
-SET DEBUG_SYNC = "now SIGNAL bwoc_continue";
-SET DEBUG_SYNC='RESET';
-connection node_2a;
+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;
-select * from t1;
-i
-1
-disconnect node_2a;
+update t1 set b = b + 1000 where id between 2 and 4;;
 connection node_1;
-drop table t1;
+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_inject_bf_long_wait.result b/mysql-test/suite/galera/r/galera_inject_bf_long_wait.result
deleted file mode 100644
index eeacc9ab212..00000000000
--- a/mysql-test/suite/galera/r/galera_inject_bf_long_wait.result
+++ /dev/null
@@ -1,24 +0,0 @@
-connection node_2;
-connection node_1;
-CREATE TABLE t1(id int not null primary key, b int) engine=InnoDB;
-INSERT INTO t1 VALUES (0,0),(1,1),(2,2),(3,3);
-BEGIN;
-UPDATE t1 set b = 100 where id between 1 and 2;;
-connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1;
-connection node_1b;
-SET @save_dbug = @@SESSION.debug_dbug;
-SET @@SESSION.innodb_lock_wait_timeout=2;
-SET @@SESSION.debug_dbug = '+d,wsrep_instrument_BF_lock_wait';
-UPDATE t1 set b = 200 WHERE id = 1;
-ERROR HY000: Lock wait timeout exceeded; try restarting transaction
-SET @@SESSION.debug_dbug = @save_dbug;
-connection node_1;
-COMMIT;
-SELECT * FROM t1;
-id	b
-0	0
-1	100
-2	100
-3	3
-disconnect node_1b;
-DROP TABLE t1;
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 6e55c59ad15..2493075b635 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
@@ -23,22 +23,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 fa200c58ff7..c5255855aa4 100644
--- a/mysql-test/suite/galera/t/galera_UK_conflict.test
+++ b/mysql-test/suite/galera/t/galera_UK_conflict.test
@@ -140,7 +140,8 @@ 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;
 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
index e68f891792c..77bb6af9f35 100644
--- a/mysql-test/suite/galera/t/galera_bf_kill_debug.cnf
+++ b/mysql-test/suite/galera/t/galera_bf_kill_debug.cnf
@@ -1,7 +1,9 @@
 !include ../galera_2nodes.cnf
 
 [mysqld.1]
-wsrep-debug=SERVER
+wsrep_log_conflicts=ON
+wsrep_debug=1
 
 [mysqld.2]
-wsrep-debug=SERVER
+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
index c322f283757..f83d4a28ce9 100644
--- a/mysql-test/suite/galera/t/galera_bf_kill_debug.test
+++ b/mysql-test/suite/galera/t/galera_bf_kill_debug.test
@@ -1,140 +1,283 @@
 --source include/galera_cluster.inc
---source include/have_innodb.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);
+
 #
-# Test case 7:
-# 1. Start a transaction on node_2,
-#    and leave it pending while holding a row locked
-# 2. set sync point pause applier
-# 3. send a conflicting write on node_1, it will pause
-#    at the sync point
-# 4. though another connection to node_2, kill the local
-#    transaction
+# 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;
 
---connection node_2
-CREATE TABLE t1(a int not null primary key auto_increment,b int) engine=InnoDB;
-insert into t1 values (NULL,1);
+#
+# 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`
 
 #
-# connection node_2a runs a local transaction, that is victim of BF abort
-# and victim of KILL command by connection node_2
+# Set DEBUG_SYNC and send conflicting DDL that will be TOI (bf) and
+# cause bf_kill
 #
---connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2
---connection node_2a
-truncate t1;
-insert into t1 values (1,0);
+--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);
 
-# start a transaction that will conflict with later applier
-begin;
-update t1 set b=2 where a=1;
+#
+# Wait until we have reached the sync point
+#
+--connection node_1
+SET DEBUG_SYNC='now WAIT_FOR bf_kill';
 
---connection node_2
-set session wsrep_sync_wait=0;
---let $wait_condition = SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'root' AND COMMAND = 'Sleep' LIMIT 1
---source include/wait_condition.inc
+#
+# 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`
 
-# connection node_2b is for controlling debug syn points
-# first set a sync point for applier, to pause during BF aborting
-# and before THD::awake would be called
 #
---connect node_2b, 127.0.0.1, root, , test, $NODE_MYPORT_2
---connection node_2b
-SET GLOBAL debug_dbug = "d,sync.before_wsrep_thd_abort";
+# 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);
 
 #
-# replicate an update, which will BF abort the victim node_2a
-# however, while applier in node 2 is handling the abort,
-# it will pause in sync point set by node_2b
+# Wait until we have reached the sync point
 #
 --connection node_1
-select * from t1;
-update t1 set b= 1 where a=1;
+SET DEBUG_SYNC='now WAIT_FOR bf_kill';
 
 #
-# wait until the applying of above update has reached the sync point
-# in node 2
+# Try to kill update query
 #
---connection node_2b
-SET SESSION DEBUG_SYNC = "now WAIT_FOR sync.before_wsrep_thd_abort_reached";
+--connection node_1b
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
 
---connection node_2
 #
-# pause KILL execution before awake
+# Let bf_kill continue
 #
-SET DEBUG_SYNC= 'before_awake_no_mutex SIGNAL awake_reached WAIT_FOR continue_kill';
---disable_query_log
---send_eval KILL $k_thread
+--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;
 
---connection node_2b
-SET DEBUG_SYNC='now WAIT_FOR awake_reached';
+--disconnect node_1a
+--disconnect node_1b
+--disconnect node_1c
 
-# release applier and KILL operator
-SET GLOBAL debug_dbug = "";
-SET DEBUG_SYNC = "now SIGNAL signal.before_wsrep_thd_abort";
-SET DEBUG_SYNC = "now SIGNAL continue_kill";
+--echo #
+--echo # Case 3: Create victim transaction and try to send user KILL
+--echo # from several threads
+--echo #
 
---connection node_2
---reap
+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);
 
---connection node_2a
---error 0,1213,2013
-select * from t1;
+#
+# 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;
 
---connection node_2
-SET DEBUG_SYNC = "RESET";
+#
+# 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
 
-drop table t1;
+--connection node_1b
+--let $k_thread = `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE USER = 'root' AND COMMAND = 'Sleep' LIMIT 1`
 
---disconnect node_2a
 #
-# Test case 7:
-# run a transaction in node 2, and set a sync point to pause the transaction
-# in commit phase.
-# Through another connection to node 2, kill the committing transaction by
-# KILL QUERY command
+# Try to kill update query from several connections concurrently
 #
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
 
---connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2
---connection node_2a
---let $connection_id = `SELECT CONNECTION_ID()`
+--connection node_1c
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
 
-CREATE TABLE t1 (i int primary key);
+--connection node_1d
+--disable_query_log
+--send_eval KILL QUERY $k_thread;
 
-# Set up sync point
-SET DEBUG_SYNC = "before_wsrep_ordered_commit SIGNAL bwoc_reached WAIT_FOR bwoc_continue";
+#
+# 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
 
-# Send insert which will block in the sync point above
---send INSERT INTO t1 VALUES (1)
+--connection node_1
+--disconnect node_1a
+--disconnect node_1b
+--disconnect node_1c
+--disconnect node_1d
+DROP TABLE t1;
 
---connection node_2
-SET DEBUG_SYNC = "now WAIT_FOR bwoc_reached";
+--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
---disable_result_log
-# victim has passed the point of no return, kill is not possible anymore
---eval KILL QUERY $connection_id
---enable_result_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;
 
-SET DEBUG_SYNC = "now SIGNAL bwoc_continue";
-SET DEBUG_SYNC='RESET';
---connection node_2a
---error 0,1213
+#
+# 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_2
-# victim was able to complete the INSERT
-select * from t1;
-
---disconnect node_2a
+--connection node_1b
+--reap
+--enable_query_log
 
 --connection node_1
-drop table t1;
+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_inject_bf_long_wait.test b/mysql-test/suite/galera/t/galera_inject_bf_long_wait.test
deleted file mode 100644
index f4aac7fd795..00000000000
--- a/mysql-test/suite/galera/t/galera_inject_bf_long_wait.test
+++ /dev/null
@@ -1,25 +0,0 @@
---source include/galera_cluster.inc
---source include/have_debug.inc
---source include/have_debug_sync.inc
-
-CREATE TABLE t1(id int not null primary key, b int) engine=InnoDB;
-INSERT INTO t1 VALUES (0,0),(1,1),(2,2),(3,3);
-
-BEGIN;
---send UPDATE t1 set b = 100 where id between 1 and 2;
-
---connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1
---connection node_1b
-SET @save_dbug = @@SESSION.debug_dbug;
-SET @@SESSION.innodb_lock_wait_timeout=2;
-SET @@SESSION.debug_dbug = '+d,wsrep_instrument_BF_lock_wait';
---error ER_LOCK_WAIT_TIMEOUT
-UPDATE t1 set b = 200 WHERE id = 1;
-SET @@SESSION.debug_dbug = @save_dbug;
-
---connection node_1
---reap
-COMMIT;
-SELECT * FROM t1;
---disconnect node_1b
-DROP TABLE t1;
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/sql/handler.cc b/sql/handler.cc
index cce50dc9289..22db83fffc1 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -928,7 +928,6 @@ static my_bool kill_handlerton(THD *thd, plugin_ref plugin,
 {
   handlerton *hton= plugin_hton(plugin);
 
-  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
   if (hton->kill_query && thd_get_ha_data(thd, hton))
     hton->kill_query(hton, thd, *(enum thd_kill_levels *) level);
   return FALSE;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 820ff51515a..27881452372 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -2076,7 +2076,7 @@ static void clean_up_mutexes()
 ****************************************************************************/
 
 #ifdef EMBEDDED_LIBRARY
-void close_connection(THD *thd, uint sql_errno)
+void close_connection(THD *thd, uint sql_errno, my_bool locked)
 {
 }
 #else
@@ -2625,7 +2625,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)
 {
   int lvl= (thd->main_security_ctx.user ? 3 : 1);
   DBUG_ENTER("close_connection");
@@ -2641,7 +2641,10 @@ void close_connection(THD *thd, uint sql_errno)
                                      "This connection closed normally without"
                                       " authentication"));
 
-  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 d0a33fabb51..ca34626a96b 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -75,7 +75,7 @@ enum enum_slave_parallel_mode {
 
 /* Function prototypes */
 void kill_mysql(THD *thd);
-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 unlink_thd(THD *thd);
diff --git a/sql/service_wsrep.cc b/sql/service_wsrep.cc
index 91b1a48cad7..5bcf0a50735 100644
--- a/sql/service_wsrep.cc
+++ b/sql/service_wsrep.cc
@@ -29,12 +29,14 @@ extern "C" my_bool wsrep_on(const THD *thd)
 
 extern "C" void wsrep_thd_LOCK(const THD *thd)
 {
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
   mysql_mutex_lock(&thd->LOCK_thd_data);
 }
 
 extern "C" void wsrep_thd_UNLOCK(const THD *thd)
 {
   mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
 }
 
 extern "C" void wsrep_thd_kill_LOCK(const THD *thd)
@@ -188,6 +190,9 @@ extern "C" void wsrep_handle_SR_rollback(THD *bf_thd,
   DBUG_ASSERT(wsrep_thd_is_SR(victim_thd));
   if (!victim_thd || !wsrep_on(bf_thd)) return;
 
+  mysql_mutex_lock(&victim_thd->LOCK_thd_kill);
+  mysql_mutex_lock(&victim_thd->LOCK_thd_data);
+
   WSREP_DEBUG("handle rollback, for deadlock: thd %llu trx_id %" PRIu64 " frags %zu conf %s",
               victim_thd->thread_id,
               victim_thd->wsrep_trx_id(),
@@ -208,6 +213,10 @@ extern "C" void wsrep_handle_SR_rollback(THD *bf_thd,
   {
     wsrep_thd_self_abort(victim_thd);
   }
+
+  mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
+  mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+
   if (bf_thd)
   {
     wsrep_store_threadvars(bf_thd);
@@ -218,7 +227,7 @@ extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
                                       my_bool signal)
 {
   mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
-  mysql_mutex_assert_not_owner(&victim_thd->LOCK_thd_data);
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
   my_bool ret= wsrep_bf_abort(bf_thd, victim_thd);
   /*
     Send awake signal if victim was BF aborted or does not
@@ -227,22 +236,19 @@ extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
    */
   if ((ret || !wsrep_on(victim_thd)) && signal)
   {
-    mysql_mutex_lock(&victim_thd->LOCK_thd_data);
-
     if (victim_thd->wsrep_aborter && victim_thd->wsrep_aborter != bf_thd->thread_id)
     {
       WSREP_DEBUG("victim is killed already by %llu, skipping awake",
                   victim_thd->wsrep_aborter);
-      mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
       return false;
     }
 
     victim_thd->wsrep_aborter= bf_thd->thread_id;
     victim_thd->awake_no_mutex(KILL_QUERY);
-    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
-  } else {
-    WSREP_DEBUG("wsrep_thd_bf_abort skipped awake");
   }
+  else
+    WSREP_DEBUG("wsrep_thd_bf_abort skipped awake");
+
   return ret;
 }
 
@@ -268,12 +274,11 @@ extern "C" my_bool wsrep_thd_order_before(const THD *left, const THD *right)
 extern "C" my_bool wsrep_thd_is_aborting(const MYSQL_THD thd)
 {
   mysql_mutex_assert_owner(&thd->LOCK_thd_data);
-  if (thd != 0)
+
+  const wsrep::client_state& cs(thd->wsrep_cs());
+  const enum wsrep::transaction::state tx_state(cs.transaction().state());
+  switch (tx_state)
   {
-    const wsrep::client_state& cs(thd->wsrep_cs());
-    const enum wsrep::transaction::state tx_state(cs.transaction().state());
-    switch (tx_state)
-    {
     case wsrep::transaction::s_must_abort:
       return (cs.state() == wsrep::client_state::s_exec ||
               cs.state() == wsrep::client_state::s_result);
@@ -283,8 +288,6 @@ extern "C" my_bool wsrep_thd_is_aborting(const MYSQL_THD thd)
     default:
       return false;
     }
-  }
-  return false;
 }
 
 static inline enum wsrep::key::type
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 954d53b9d72..396e4106788 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -2004,13 +2004,13 @@ void THD::abort_current_cond_wait(bool force)
           the Vio might be disassociated concurrently.
 */
 
-void THD::disconnect()
+void THD::disconnect_mutexed()
 {
   Vio *vio= NULL;
 
-  set_killed(KILL_CONNECTION);
-
-  mysql_mutex_lock(&LOCK_thd_data);
+  mysql_mutex_assert_owner(&LOCK_thd_data);
+  mysql_mutex_assert_owner(&LOCK_thd_kill);
+  set_killed_no_mutex(KILL_CONNECTION);
 
 #ifdef SIGNAL_WITH_VIO_CLOSE
   /*
@@ -2026,8 +2026,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);
 }
 
 
@@ -2054,6 +2052,9 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
 
   if (needs_thr_lock_abort)
   {
+    /* Protect thread from concurrent disconnect and delete */
+    mysql_mutex_lock(&in_use->LOCK_thd_kill);
+    /* Protect thread from concurrent usage */
     mysql_mutex_lock(&in_use->LOCK_thd_data);
     /* If not already dying */
     if (in_use->killed != KILL_CONNECTION_HARD)
@@ -2070,11 +2071,20 @@ 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);
-        }
       }
+#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(this, in_use, false);
+      }
+#endif /* WITH_WSREP */
     }
+    mysql_mutex_unlock(&in_use->LOCK_thd_kill);
     mysql_mutex_unlock(&in_use->LOCK_thd_data);
   }
   DBUG_RETURN(signalled);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index b5874d2435a..d3b4bcbd567 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3734,8 +3734,15 @@ public:
   void abort_current_cond_wait(bool force);
  
   /** Disconnect the associated communication endpoint. */
-  void disconnect();
-
+  inline void disconnect()
+  {
+    mysql_mutex_lock(&LOCK_thd_kill);
+    mysql_mutex_lock(&LOCK_thd_data);
+    disconnect_mutexed();
+    mysql_mutex_unlock(&LOCK_thd_kill);
+    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 2a48c8fb1ce..10dd094e137 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -9156,7 +9156,8 @@ void add_join_natural(TABLE_LIST *a, TABLE_LIST *b, List<String> *using_fields,
   @param query_id If true, search by query_id instead of thread_id
 
   @return NULL    - not found
-          pointer - thread found, and its LOCK_thd_kill is locked.
+          pointer - thread found, and its LOCK_thd_data and
+                    LOCK_thd_kill is locked.
 */
 
 struct find_thread_callback_arg
@@ -9174,6 +9175,7 @@ static my_bool find_thread_callback(THD *thd, find_thread_callback_arg *arg)
   if (arg->id == (arg->query_id ? thd->query_id : (longlong) thd->thread_id))
   {
     mysql_mutex_lock(&thd->LOCK_thd_kill);    // Lock from delete
+    mysql_mutex_lock(&thd->LOCK_thd_data);    // Lock from concurrent usage
     arg->thd= thd;
     return 1;
   }
@@ -9231,7 +9233,6 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
       faster and do a harder kill than KILL_SYSTEM_THREAD;
     */
 
-    mysql_mutex_lock(&tmp->LOCK_thd_data); // for various wsrep* checks below
 #ifdef WITH_WSREP
     if (((thd->security_ctx->master_access & PRIV_KILL_OTHER_USER_PROCESS) ||
         thd->security_ctx->user_matches(tmp->security_ctx)) &&
@@ -9263,8 +9264,8 @@ kill_one_thread(THD *thd, longlong id, killed_state kill_signal, killed_type typ
     else
       error= (type == KILL_TYPE_QUERY ? ER_KILL_QUERY_DENIED_ERROR :
                                         ER_KILL_DENIED_ERROR);
-    mysql_mutex_unlock(&tmp->LOCK_thd_data);
   }
+  mysql_mutex_unlock(&tmp->LOCK_thd_data);
   mysql_mutex_unlock(&tmp->LOCK_thd_kill);
   DBUG_PRINT("exit", ("%d", error));
   DBUG_RETURN(error);
@@ -9377,6 +9378,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 (likely(!(error= kill_one_thread(thd, id, state, type))))
   {
     if (!thd->killed)
@@ -9386,6 +9399,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 */
 }
 
 
@@ -9394,6 +9412,18 @@ 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 (likely(!(error= kill_threads_for_user(thd, user, state, &rows))))
     my_ok(thd, rows);
   else
@@ -9404,6 +9434,11 @@ 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/sql_show.cc b/sql/sql_show.cc
index f9049f47324..87324baa8b7 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3131,6 +3131,7 @@ int fill_show_explain(THD *thd, TABLE_LIST *table, COND *cond)
   if ((tmp= find_thread_by_id(thread_id)))
   {
     Security_context *tmp_sctx= tmp->security_ctx;
+    mysql_mutex_unlock(&tmp->LOCK_thd_data);
     /*
       If calling_user==NULL, calling thread has SUPER or PROCESS
       privilege, and so can do SHOW EXPLAIN on any user.
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 89823cea3bf..ea70931209b 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -2916,7 +2916,9 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                   request_thd, granted_thd);
     ticket->wsrep_report(wsrep_debug);
 
+    mysql_mutex_lock(&granted_thd->LOCK_thd_kill);
     mysql_mutex_lock(&granted_thd->LOCK_thd_data);
+
     if (wsrep_thd_is_toi(granted_thd) ||
         wsrep_thd_is_applying(granted_thd))
     {
@@ -2925,13 +2927,15 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
         WSREP_DEBUG("BF thread waiting for SR in aborting state");
         ticket->wsrep_report(wsrep_debug);
         mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
       }
       else if (wsrep_thd_is_SR(granted_thd) && !wsrep_thd_is_SR(request_thd))
       {
-        WSREP_MDL_LOG(INFO, "MDL conflict, DDL vs SR", 
+        WSREP_MDL_LOG(INFO, "MDL conflict, DDL vs SR",
                       schema, schema_len, request_thd, granted_thd);
-        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
         wsrep_abort_thd(request_thd, granted_thd, 1);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
       }
       else
       {
@@ -2939,6 +2943,7 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                       request_thd, granted_thd);
         ticket->wsrep_report(true);
         mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
         unireg_abort(1);
       }
     }
@@ -2948,14 +2953,16 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
       WSREP_DEBUG("BF thread waiting for FLUSH");
       ticket->wsrep_report(wsrep_debug);
       mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+      mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
     }
     else if (request_thd->lex->sql_command == SQLCOM_DROP_TABLE)
     {
       WSREP_DEBUG("DROP caused BF abort, conf %s",
                   wsrep_thd_transaction_state_str(granted_thd));
       ticket->wsrep_report(wsrep_debug);
-      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       wsrep_abort_thd(request_thd, granted_thd, 1);
+      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+      mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
     }
     else
     {
@@ -2964,8 +2971,9 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
       ticket->wsrep_report(wsrep_debug);
       if (granted_thd->wsrep_trx().active())
       {
-        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
         wsrep_abort_thd(request_thd, granted_thd, 1);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+        mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
       }
       else
       {
@@ -2973,10 +2981,11 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
           Granted_thd is likely executing with wsrep_on=0. If the requesting
           thd is BF, BF abort and wait.
         */
-        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
-        if (wsrep_thd_is_BF(request_thd, FALSE))
+        if (wsrep_thd_is_BF(request_thd, false))
         {
           ha_abort_transaction(request_thd, granted_thd, TRUE);
+          mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+          mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
         }
         else
         {
@@ -2989,15 +2998,15 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
     }
   }
   else
-  {
     mysql_mutex_unlock(&request_thd->LOCK_thd_data);
-  }
 }
 
 /**/
 static bool abort_replicated(THD *thd)
 {
   bool ret_code= false;
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
+  mysql_mutex_assert_owner(&thd->LOCK_thd_kill);
   if (thd->wsrep_trx().state() == wsrep::transaction::s_committing)
   {
     WSREP_DEBUG("aborting replicated trx: %llu", (ulonglong)(thd->real_id));
@@ -3009,18 +3018,35 @@ static bool abort_replicated(THD *thd)
 }
 
 /**/
-static inline bool is_client_connection(THD *thd)
+static inline my_bool is_client_connection(THD *thd, my_bool *sync)
 {
-  return (thd->wsrep_client_thread && thd->variables.wsrep_on);
+  if (sync)
+    mysql_mutex_lock(&thd->LOCK_thd_data);
+  my_bool ret= (thd->wsrep_client_thread && thd->variables.wsrep_on);
+  if (sync)
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+
+  return ret;
 }
 
-static inline bool is_committing_connection(THD *thd)
+static inline my_bool is_replaying_connection(THD *thd, my_bool *sync)
 {
-  bool ret;
+  if (sync)
+    mysql_mutex_lock(&thd->LOCK_thd_data);
+  my_bool ret= ((thd->wsrep_trx().state() == wsrep::transaction::s_replaying) ? true : false);
+  if (sync)
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
 
-  mysql_mutex_lock(&thd->LOCK_thd_data);
-  ret=  (thd->wsrep_trx().state() == wsrep::transaction::s_committing) ? true : false;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  return ret;
+}
+
+static inline my_bool is_committing_connection(THD *thd, my_bool *sync)
+{
+  if (sync)
+    mysql_mutex_lock(&thd->LOCK_thd_data);
+  my_bool ret= ((thd->wsrep_trx().state() == wsrep::transaction::s_committing) ? true : false);
+  if (sync)
+    mysql_mutex_lock(&thd->LOCK_thd_data);
 
   return ret;
 }
@@ -3029,12 +3055,17 @@ static my_bool have_client_connections(THD *thd, void*)
 {
   DBUG_PRINT("quit",("Informing thread %lld that it's time to die",
                      (longlong) thd->thread_id));
-  if (is_client_connection(thd) && thd->killed == KILL_CONNECTION)
+  my_bool ret=false;
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  if (is_client_connection(thd, NULL) && thd->killed == KILL_CONNECTION)
   {
     (void)abort_replicated(thd);
-    return 1;
+    ret= true;
   }
-  return 0;
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
+  return ret;
 }
 
 static void wsrep_close_thread(THD *thd)
@@ -3046,59 +3077,72 @@ static void wsrep_close_thread(THD *thd)
   mysql_mutex_unlock(&thd->LOCK_thd_kill);
 }
 
-static my_bool have_committing_connections(THD *thd, void *)
+static my_bool have_committing_connections(THD *thd, my_bool *sync)
 {
-  return is_client_connection(thd) && is_committing_connection(thd) ? 1 : 0;
+  my_bool *need_sync= sync ? NULL : (my_bool *)thd;
+  if (sync)
+    mysql_mutex_lock(&thd->LOCK_thd_data);
+  my_bool ret= (is_client_connection(thd, need_sync) && is_committing_connection(thd, need_sync) ? 1 : 0);
+  if (sync)
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+  return ret;
 }
 
 int wsrep_wait_committing_connections_close(int wait_time)
 {
   int sleep_time= 100;
+  my_bool sync=true;
 
   WSREP_DEBUG("wait for committing transaction to close: %d sleep: %d", wait_time, sleep_time);
-  while (server_threads.iterate(have_committing_connections) && wait_time > 0)
+  while (server_threads.iterate(have_committing_connections, &sync) && wait_time > 0)
   {
     WSREP_DEBUG("wait for committing transaction to close: %d", wait_time);
     my_sleep(sleep_time);
     wait_time -= sleep_time;
   }
-  return server_threads.iterate(have_committing_connections);
+  return server_threads.iterate(have_committing_connections, &sync);
 }
 
 static my_bool kill_all_threads(THD *thd, THD *caller_thd)
 {
   DBUG_PRINT("quit", ("Informing thread %lld that it's time to die",
                       (longlong) thd->thread_id));
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  mysql_mutex_lock(&thd->LOCK_thd_data);
   /* We skip slave threads & scheduler on this first loop through. */
-  if (is_client_connection(thd) && thd != caller_thd)
+  if (is_client_connection(thd, NULL) && thd != caller_thd)
   {
-    if (is_replaying_connection(thd))
-      thd->set_killed(KILL_CONNECTION);
+	  if (is_replaying_connection(thd,NULL))
+      thd->set_killed_no_mutex(KILL_CONNECTION);
     else if (!abort_replicated(thd))
     {
       /* replicated transactions must be skipped */
       WSREP_DEBUG("closing connection %lld", (longlong) thd->thread_id);
       /* instead of wsrep_close_thread() we do now  soft kill by THD::awake */
-      thd->awake(KILL_CONNECTION);
+      thd->awake_no_mutex(KILL_CONNECTION);
     }
   }
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
   return 0;
 }
 
 static my_bool kill_remaining_threads(THD *thd, THD *caller_thd)
 {
-#ifndef __bsdi__				// Bug in BSDI kernel
-  if (is_client_connection(thd) &&
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  if (is_client_connection(thd,NULL) &&
       !abort_replicated(thd)    &&
-      !is_replaying_connection(thd) &&
+	  !is_replaying_connection(thd,NULL) &&
       thd_is_connection_alive(thd) &&
       thd != caller_thd)
   {
 
     WSREP_INFO("killing local connection: %lld", (longlong) thd->thread_id);
-    close_connection(thd);
+    close_connection(thd, true);
   }
-#endif
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
   return 0;
 }
 
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index bccd1a4f8b3..c87fc31e9ca 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
@@ -64,7 +64,7 @@ static void wsrep_replication_process(THD *thd,
 
   delete thd->wsrep_rgi->rli->mi;
   delete thd->wsrep_rgi->rli;
-  
+
   thd->wsrep_rgi->cleanup_after_session();
   delete thd->wsrep_rgi;
   thd->wsrep_rgi= NULL;
@@ -314,8 +314,8 @@ int wsrep_abort_thd(THD *bf_thd_ptr, THD *victim_thd_ptr, my_bool signal)
   THD *victim_thd= (THD *) victim_thd_ptr;
   THD *bf_thd= (THD *) bf_thd_ptr;
 
-  mysql_mutex_lock(&victim_thd->LOCK_thd_data);
-
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
   /* Note that when you use RSU node is desynced from cluster, thus WSREP(thd)
   might not be true.
   */
@@ -327,16 +327,13 @@ int wsrep_abort_thd(THD *bf_thd_ptr, THD *victim_thd_ptr, my_bool signal)
   {
       WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu", (bf_thd) ?
                   (long long)bf_thd->real_id : 0, (long long)victim_thd->real_id);
-      mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
       ha_abort_transaction(bf_thd, victim_thd, signal);
-      mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   }
   else
   {
     WSREP_DEBUG("wsrep_abort_thd not effective: %p %p", bf_thd, victim_thd);
   }
 
-  mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
   DBUG_RETURN(1);
 }
 
@@ -345,6 +342,9 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
   WSREP_LOG_THD(bf_thd, "BF aborter before");
   WSREP_LOG_THD(victim_thd, "victim before");
 
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
+
   DBUG_EXECUTE_IF("sync.wsrep_bf_abort",
                   {
                     const char act[]=
@@ -381,13 +381,26 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
 
   if (wsrep_thd_is_toi(bf_thd))
   {
+    /* Here we enter wsrep-lib were LOCK_thd_data will be acquired,
+    thus we need to release it. */
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
     ret= victim_thd->wsrep_cs().total_order_bf_abort(bf_seqno);
+    mysql_mutex_lock(&victim_thd->LOCK_thd_kill);
+    mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   }
   else
   {
-    DBUG_ASSERT(WSREP(victim_thd) ? victim_thd->wsrep_trx().active() : 1);
+    /* Test: mysql-wsrep-features#165. Here we enter wsrep-lib
+    were LOCK_thd_data will be acquired and later LOCK_thd_kill
+    thus we need to release them. */
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
     ret= victim_thd->wsrep_cs().bf_abort(bf_seqno);
+    mysql_mutex_lock(&victim_thd->LOCK_thd_kill);
+    mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   }
+
   if (ret)
   {
     wsrep_bf_aborts_counter++;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index e431e0499c4..ad4f29e415a 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -4936,7 +4936,8 @@ static void innobase_kill_query(handlerton*, THD *thd, enum thd_kill_levels)
     ut_ad(trx->mysql_thd == thd);
     if (!trx->lock.wait_lock);
 #ifdef WITH_WSREP
-    else if (trx->is_wsrep() && wsrep_thd_is_aborting(thd))
+    else if (trx->is_wsrep() && (wsrep_thd_is_aborting(thd)
+	     || trx->lock.was_chosen_as_deadlock_victim.fetch_and(byte(~2))))
       /* 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. */;
@@ -18486,10 +18487,12 @@ void lock_wait_wsrep_kill(trx_t *bf_trx, ulong thd_id, trx_id_t trx_id)
 {
   THD *bf_thd= bf_trx->mysql_thd;
 
+  /* Below function will lock THD::LOCK_thd_kill to protect victim
+  from concurrent delete or disconnect and THD::LOCK_thd_data
+  to protect from concurrent usage. */
   if (THD *vthd= find_thread_by_id(thd_id))
   {
     bool aborting= false;
-    wsrep_thd_LOCK(vthd);
     trx_t *vtrx= thd_to_trx(vthd);
     if (vtrx)
     {
@@ -18542,18 +18545,16 @@ void lock_wait_wsrep_kill(trx_t *bf_trx, ulong thd_id, trx_id_t trx_id)
       mysql_mutex_unlock(&lock_sys.wait_mutex);
       vtrx->mutex_unlock();
     }
-    wsrep_thd_UNLOCK(vthd);
+
     if (aborting)
     {
       /* if victim is waiting for some other lock, we have to cancel
          that waiting
       */
       lock_sys.cancel_lock_wait_for_trx(vtrx);
-
-      DEBUG_SYNC(bf_thd, "before_wsrep_thd_abort");
       wsrep_thd_bf_abort(bf_thd, vthd, true);
     }
-    wsrep_thd_kill_UNLOCK(vthd);
+    wsrep_thd_UNLOCK(vthd);
   }
 }
 
@@ -18579,47 +18580,46 @@ wsrep_abort_transaction(
 	ut_ad(victim_thd);
 
 	trx_t* victim_trx= thd_to_trx(victim_thd);
-
-	WSREP_DEBUG("abort transaction: BF: %s victim: %s victim conf: %s",
-			wsrep_thd_query(bf_thd),
-			wsrep_thd_query(victim_thd),
-			wsrep_thd_transaction_state_str(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 and THD::LOCK_thd_kill
+	to protect from disconnect or delete. */
+	WSREP_DEBUG("wsrep_abort_transaction: BF:"
+		    " thread %ld client_state %s client_mode %s"
+		    " trx_state %s query %s trx " TRX_ID_FMT,
+		    thd_get_thread_id(bf_thd),
+		    wsrep_thd_client_state_str(bf_thd),
+		    wsrep_thd_client_mode_str(bf_thd),
+		    wsrep_thd_transaction_state_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_client_state_str(victim_thd),
+		    wsrep_thd_client_mode_str(victim_thd),
+		    wsrep_thd_transaction_state_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,
+			    (int)victim_trx->state);
 		victim_trx->lock.was_chosen_as_deadlock_victim.fetch_or(2);
-
-		wsrep_thd_kill_LOCK(victim_thd);
-		wsrep_thd_LOCK(victim_thd);
-		bool aborting= !wsrep_thd_set_wsrep_aborter(bf_thd, victim_thd);
-		wsrep_thd_UNLOCK(victim_thd);
-		if (aborting) {
-			DEBUG_SYNC(bf_thd, "before_wsrep_thd_abort");
-			DBUG_EXECUTE_IF("sync.before_wsrep_thd_abort",
-					 {
-					   const char act[]=
-					     "now "
-					     "SIGNAL sync.before_wsrep_thd_abort_reached "
-					     "WAIT_FOR signal.before_wsrep_thd_abort";
-					   DBUG_ASSERT(!debug_sync_set_action(bf_thd,
-									      STRING_WITH_LEN(act)));
-					 };);
+		if (!wsrep_thd_set_wsrep_aborter(bf_thd, victim_thd))
 			wsrep_thd_bf_abort(bf_thd, victim_thd, signal);
-		}
-		wsrep_thd_kill_UNLOCK(victim_thd);
 		DBUG_VOID_RETURN;
 	} else {
-		DBUG_EXECUTE_IF("sync.before_wsrep_thd_abort",
-				 {
-				   const char act[]=
-				     "now "
-				     "SIGNAL sync.before_wsrep_thd_abort_reached "
-				     "WAIT_FOR signal.before_wsrep_thd_abort";
-				   DBUG_ASSERT(!debug_sync_set_action(bf_thd,
-								      STRING_WITH_LEN(act)));
-				 };);
-		wsrep_thd_kill_LOCK(victim_thd);
+		WSREP_DEBUG("wsrep_abort_transaction: Victim thread %ld "
+			    "no transaction",
+			    thd_get_thread_id(victim_thd));
 		wsrep_thd_bf_abort(bf_thd, victim_thd, signal);
-		wsrep_thd_kill_UNLOCK(victim_thd);
 	}
 
 	DBUG_VOID_RETURN;
diff --git a/storage/innobase/include/ha_prototypes.h b/storage/innobase/include/ha_prototypes.h
index 2dd7c571386..fdd4eb4bce2 100644
--- a/storage/innobase/include/ha_prototypes.h
+++ b/storage/innobase/include/ha_prototypes.h
@@ -192,6 +192,10 @@ innobase_casedn_str(
 	char*	a);	/*!< in/out: string to put in lower case */
 
 #ifdef WITH_WSREP
+void
+wsrep_innobase_kill_one_trx(MYSQL_THD const thd_ptr,
+			    trx_t *victim_trx,
+        		    my_bool signal);
 ulint wsrep_innobase_mysql_sort(int mysql_type, uint charset_number,
                              unsigned char* str, ulint str_length,
                              ulint buf_length);