MDEV-25114 Crash: WSREP: invalid state ROLLED_BACK (FATAL)bb-10.4-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>

20 files changed, 691 insertions, 326 deletions

diff --git a/mysql-test/suite/galera/r/galera_UK_conflict.result b/mysql-test/suite/galera/r/galera_UK_conflict.result
index 44bb64c9d63..cc7e17d7c58 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";
@@ -125,6 +125,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_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 9978ba9b8bf..7e5b5541214 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_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 757fa95a9a3..8dd3868e6fb 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -859,7 +859,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->state == SHOW_OPTION_YES && hton->kill_query &&
       thd_get_ha_data(thd, hton))
     hton->kill_query(hton, thd, *(enum thd_kill_levels *) level);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index ba1d477882f..c225e35153a 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -2122,7 +2122,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
@@ -2526,7 +2526,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");
@@ -2542,7 +2542,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 8e33d6488e4..e090042f442 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 14f136ca480..2ef1ea26e61 100644
--- a/sql/service_wsrep.cc
+++ b/sql/service_wsrep.cc
@@ -1,4 +1,4 @@
-/* Copyright 2018 Codership Oy <info@codership.com>
+/* Copyright 2018-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
@@ -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)
@@ -189,6 +191,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(),
@@ -209,6 +214,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);
@@ -219,7 +228,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
@@ -228,22 +237,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,13 +274,12 @@ 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)
 {
+  DBUG_ASSERT(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,9 +288,7 @@ extern "C" my_bool wsrep_thd_is_aborting(const MYSQL_THD thd)
       return true;
     default:
       return false;
-    }
   }
-  return false;
 }
 
 static inline enum wsrep::key::type
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 28bf77c94e8..c2dcc07997e 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1977,13 +1977,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
   /*
@@ -1999,8 +1999,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);
 }
 
 
@@ -2027,6 +2025,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)
@@ -2043,11 +2044,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 7570211f586..cf90f78ea30 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3336,8 +3336,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 8999397fee7..276608e036c 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -9332,6 +9332,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)
@@ -9341,6 +9353,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 */
 }
 
 
@@ -9349,6 +9366,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 (likely(!(error= kill_threads_for_user(thd, user, state, &rows))))
     my_ok(thd, rows);
   else
@@ -9359,6 +9388,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 55564dbe235..92f1760cbff 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -2319,7 +2319,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))
     {
@@ -2328,13 +2330,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
       {
@@ -2342,6 +2346,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);
       }
     }
@@ -2351,14 +2356,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
     {
@@ -2367,8 +2374,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
       {
@@ -2376,10 +2384,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
         {
@@ -2392,15 +2401,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));
@@ -2412,29 +2421,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_replaying_connection(THD *thd)
+static inline my_bool is_replaying_connection(THD *thd, my_bool *sync)
 {
-  bool ret;
-
-  mysql_mutex_lock(&thd->LOCK_thd_data);
-  ret=  (thd->wsrep_trx().state() == wsrep::transaction::s_replaying) ? true : false;
-  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  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);
 
   return ret;
 }
 
-static inline bool is_committing_connection(THD *thd)
+static inline my_bool is_committing_connection(THD *thd, my_bool *sync)
 {
-  bool ret;
-
-  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);
+  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;
 }
@@ -2443,12 +2458,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)
@@ -2460,59 +2480,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 2e02110d697..7eb08321905 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[]=
@@ -358,7 +358,7 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
   if (WSREP(victim_thd) && !victim_thd->wsrep_trx().active())
   {
     WSREP_DEBUG("wsrep_bf_abort, BF abort for non active transaction");
-    switch (victim_thd->wsrep_trx().state()) 
+    switch (victim_thd->wsrep_trx().state())
     {
     case wsrep::transaction::s_aborting: /* fall through */
     case wsrep::transaction::s_aborted:
@@ -367,7 +367,14 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
     default:
       break;
     }
+
+    /* Test: galera_create_table_as_select. Here we enter wsrep-lib
+    were LOCK_thd_data will be acquired, thus we need to release it.
+    However, we can still hold LOCK_thd_kill to protect from
+    disconnect or delete. */
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
     wsrep_start_transaction(victim_thd, victim_thd->wsrep_next_trx_id());
+    mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   }
 
   bool ret;
@@ -375,12 +382,25 @@ 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. However, we can still hold
+    LOCK_thd_kill to protect from disconnect or delete. */
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
     ret= victim_thd->wsrep_cs().total_order_bf_abort(bf_seqno);
+    mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   }
   else
   {
+    /* 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 c4368c80c9b..9ce3790b14a 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -5185,7 +5185,7 @@ static void innobase_kill_query(handlerton*, THD *thd, enum thd_kill_levels)
   {
     ut_ad(trx->mysql_thd == thd);
 #ifdef WITH_WSREP
-    if (trx->is_wsrep() && wsrep_thd_is_aborting(thd))
+    if (wsrep_thd_is_aborting(thd) || trx->lock.was_chosen_as_wsrep_victim)
       /* 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. */
@@ -18637,12 +18637,6 @@ static struct st_mysql_storage_engine innobase_storage_engine=
 
 #ifdef WITH_WSREP
 
-struct bg_wsrep_kill_trx_arg {
-       my_thread_id    thd_id, bf_thd_id;
-       trx_id_t        trx_id, bf_trx_id;
-       bool            signal;
-};
-
 /** This function is used to kill one transaction.
 
 This transaction was open on this node (not-yet-committed), and a
@@ -18666,13 +18660,11 @@ comparison as in the local certification failure.
 @param[in]	bf_thd		Brute force (BF) thread
 @param[in,out]	victim_trx	Vimtim trx to be killed
 @param[in]	signal		Should victim be signaled */
-UNIV_INTERN
 void
 wsrep_innobase_kill_one_trx(
-	THD* bf_thd,
+	MYSQL_THD const bf_thd,
 	trx_t *victim_trx,
-	bool signal)
-
+	my_bool signal)
 {
 	ut_ad(bf_thd);
 	ut_ad(victim_trx);
@@ -18680,9 +18672,14 @@ 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;
-	ut_ad(thd);
+
+	if (!thd)
+	{
+		WSREP_WARN("no THD for trx: " TRX_ID_FMT, victim_trx->id);
+		DBUG_VOID_RETURN;
+	}
+
 	/* Note that bf_trx might not exist here e.g. on MDL conflict
 	case (test: galera_concurrent_ctas). Similarly, BF thread
 	could be also acquiring MDL-lock causing victim to be
@@ -18691,31 +18688,37 @@ wsrep_innobase_kill_one_trx(
 	trx_t* bf_trx= thd_to_trx(bf_thd);
 	DBUG_ASSERT(wsrep_on(bf_thd));
 
-	WSREP_LOG_CONFLICT(bf_thd, thd, TRUE);
+	/* 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);
+	DEBUG_SYNC(bf_thd, "wsrep_after_BF_victim_lock");
 
 	WSREP_DEBUG("Aborter %s trx_id: " TRX_ID_FMT " thread: %ld "
-		"seqno: %lld client_state: %s client_mode: %s transaction_mode: %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),
-		wsrep_thd_trx_seqno(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));
+		    "seqno: %lld client_state: %s client_mode: %s transaction_mode: %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),
+		    wsrep_thd_trx_seqno(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));
 
 	WSREP_DEBUG("Victim %s trx_id: " TRX_ID_FMT " thread: %ld "
-		"seqno: %lld client_state: %s  client_mode: %s transaction_mode: %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_client_state_str(thd),
-		wsrep_thd_client_mode_str(thd),
-		wsrep_thd_transaction_state_str(thd),
-		wsrep_thd_query(thd));
+		    "seqno: %lld client_state: %s  client_mode: %s transaction_mode: %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_client_state_str(thd),
+		    wsrep_thd_client_mode_str(thd),
+		    wsrep_thd_transaction_state_str(thd),
+		    wsrep_thd_query(thd));
+
+	WSREP_LOG_CONFLICT(bf_thd, thd, TRUE);
 
 	/* Mark transaction as a victim for Galera abort */
 	victim_trx->lock.was_chosen_as_wsrep_victim= true;
@@ -18741,6 +18744,8 @@ wsrep_innobase_kill_one_trx(
 		}
 	}
 
+	wsrep_thd_UNLOCK(thd);
+
 	DBUG_VOID_RETURN;
 }
 
@@ -18764,33 +18769,61 @@ wsrep_abort_transaction(
 	DBUG_ENTER("wsrep_abort_transaction");
 	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);
 
-	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));
+	/* 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,
+			    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, victim_trx, signal);
 		trx_mutex_exit(victim_trx);
 		lock_mutex_exit();
-
 		wsrep_srv_conc_cancel_wait(victim_trx);
+		wsrep_thd_LOCK(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_DEBUG("wsrep_abort_transaction: Victim thread %ld "
+			    "no transaction",
+			    thd_get_thread_id(victim_thd));
 		wsrep_thd_bf_abort(bf_thd, victim_thd, signal);
 	}
 
diff --git a/storage/innobase/include/ha_prototypes.h b/storage/innobase/include/ha_prototypes.h
index 108f6925ef7..482df3b1dd7 100644
--- a/storage/innobase/include/ha_prototypes.h
+++ b/storage/innobase/include/ha_prototypes.h
@@ -230,12 +230,10 @@ innobase_casedn_str(
 	char*	a);	/*!< in/out: string to put in lower case */
 
 #ifdef WITH_WSREP
-UNIV_INTERN
 void
-wsrep_innobase_kill_one_trx(
-	THD* bf_thd,
-	trx_t *victim_trx,
-	bool signal);
+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, unsigned int str_length,
diff --git a/storage/innobase/lock/lock0wait.cc b/storage/innobase/lock/lock0wait.cc
index 59405e4c1ad..ae909619bf9 100644
--- a/storage/innobase/lock/lock0wait.cc
+++ b/storage/innobase/lock/lock0wait.cc
@@ -186,11 +186,7 @@ as a side effect trigger lock monitor
 @param[in]    trx    transaction owning the lock
 @param[in]    locked true if trx and lock_sys.mutex is ownd
 @return	false for regular lock timeout */
-static
-bool
-wsrep_is_BF_lock_timeout(
-	const trx_t*	trx,
-	bool		locked = true)
+static bool wsrep_is_BF_lock_timeout(const trx_t* trx)
 {
 	bool long_wait= (trx->error_state != DB_DEADLOCK &&
 			 trx->is_wsrep() &&
@@ -203,19 +199,6 @@ wsrep_is_BF_lock_timeout(
 	if (long_wait) {
 		ib::info() << "WSREP: BF lock wait long for trx:" << trx->id
 			   << " query: " << wsrep_thd_query(trx->mysql_thd);
-
-		if (!locked)
-			lock_mutex_enter();
-
-		ut_ad(lock_mutex_own());
-
-		trx_print_latched(stderr, trx, 3000);
-		/* Note this will release lock_sys mutex */
-		lock_print_info_all_transactions(stderr);
-
-		if (locked)
-			lock_mutex_enter();
-
 		return was_wait;
 	} else
 		return false;
@@ -404,7 +387,7 @@ lock_wait_suspend_thread(
 	    && wait_time > (double) lock_wait_timeout
 #ifdef WITH_WSREP
 	    && (!trx->is_wsrep()
-		|| (!wsrep_is_BF_lock_timeout(trx, false)
+		|| (!wsrep_is_BF_lock_timeout(trx)
 		    && trx->error_state != DB_DEADLOCK))
 #endif /* WITH_WSREP */
 	    ) {
diff --git a/storage/innobase/srv/srv0conc.cc b/storage/innobase/srv/srv0conc.cc
index 6167c8daeba..b8a2b4e788a 100644
--- a/storage/innobase/srv/srv0conc.cc
+++ b/storage/innobase/srv/srv0conc.cc
@@ -118,11 +118,15 @@ srv_conc_enter_innodb_with_atomics(
 	for (;;) {
 		ulint	sleep_in_us;
 #ifdef WITH_WSREP
-		if (trx->is_wsrep() && wsrep_thd_is_aborting(trx->mysql_thd)) {
-			if (UNIV_UNLIKELY(wsrep_debug)) {
-				ib::info() <<
-					"srv_conc_enter due to MUST_ABORT";
+		if (trx->is_wsrep()) {
+			wsrep_thd_LOCK(trx->mysql_thd);
+			if (wsrep_thd_is_aborting(trx->mysql_thd)) {
+				if (UNIV_UNLIKELY(wsrep_debug)) {
+						ib::info() <<
+						"srv_conc_enter due to MUST_ABORT";
+				}
 			}
+			wsrep_thd_UNLOCK(trx->mysql_thd);
 			srv_conc_force_enter_innodb(trx);
 			return;
 		}