diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 8c2b4ae363d..f5d0ef87e07 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -42,7 +42,7 @@ variables:
   CMAKE_FLAGS: "-DWITH_SSL=system -DPLUGIN_COLUMNSTORE=NO -DPLUGIN_ROCKSDB=NO -DPLUGIN_S3=NO -DPLUGIN_MROONGA=NO -DPLUGIN_CONNECT=NO -DPLUGIN_MROONGA=NO -DPLUGIN_TOKUDB=NO -DPLUGIN_PERFSCHEMA=NO -DWITH_WSREP=OFF"
   # Major version dictates which branches share the same ccache. E.g. 10.6-abc
   # and 10.6-xyz will have the same cache.
-  MARIADB_MAJOR_VERSION: "10.8"
+  MARIADB_MAJOR_VERSION: "10.9"
   # NOTE! Currently ccache is only used on the Centos8 build. As each job has
   # sufficiently different environments they are unable to benefit from each
   # other's ccaches. As each build generates about 1 GB of ccache, having
diff --git a/include/mysql/service_wsrep.h b/include/mysql/service_wsrep.h
index 8541b348ae4..8add709362e 100644
--- a/include/mysql/service_wsrep.h
+++ b/include/mysql/service_wsrep.h
@@ -57,6 +57,7 @@ extern struct wsrep_service_st {
   my_bool                     (*wsrep_on_func)(const MYSQL_THD thd);
   bool                        (*wsrep_prepare_key_for_innodb_func)(MYSQL_THD thd, const unsigned char*, size_t, const unsigned char*, size_t, struct wsrep_buf*, size_t*);
   void                        (*wsrep_thd_LOCK_func)(const MYSQL_THD thd);
+  int                         (*wsrep_thd_TRYLOCK_func)(const MYSQL_THD thd);
   void                        (*wsrep_thd_UNLOCK_func)(const MYSQL_THD thd);
   const char *                (*wsrep_thd_query_func)(const MYSQL_THD thd);
   int                         (*wsrep_thd_retry_counter_func)(const MYSQL_THD thd);
@@ -89,7 +90,6 @@ extern struct wsrep_service_st {
   ulong                       (*wsrep_OSU_method_get_func)(const MYSQL_THD thd);
   my_bool                     (*wsrep_thd_has_ignored_error_func)(const MYSQL_THD thd);
   void                        (*wsrep_thd_set_ignored_error_func)(MYSQL_THD thd, my_bool val);
-  bool                        (*wsrep_thd_set_wsrep_aborter_func)(MYSQL_THD bf_thd, MYSQL_THD thd);
   void                        (*wsrep_report_bf_lock_wait_func)(const MYSQL_THD thd,
                                                                 unsigned long long trx_id);
   void                        (*wsrep_thd_kill_LOCK_func)(const MYSQL_THD thd);
@@ -111,6 +111,7 @@ extern struct wsrep_service_st {
 #define wsrep_on(thd) (thd) && WSREP_ON && wsrep_service->wsrep_on_func(thd)
 #define wsrep_prepare_key_for_innodb(A,B,C,D,E,F,G) wsrep_service->wsrep_prepare_key_for_innodb_func(A,B,C,D,E,F,G)
 #define wsrep_thd_LOCK(T) wsrep_service->wsrep_thd_LOCK_func(T)
+#define wsrep_thd_TRYLOCK(T) wsrep_service->wsrep_thd_TRYLOCK_func(T)
 #define wsrep_thd_UNLOCK(T) wsrep_service->wsrep_thd_UNLOCK_func(T)
 #define wsrep_thd_kill_LOCK(T) wsrep_service->wsrep_thd_kill_LOCK_func(T)
 #define wsrep_thd_kill_UNLOCK(T) wsrep_service->wsrep_thd_kill_UNLOCK_func(T)
@@ -141,7 +142,6 @@ extern struct wsrep_service_st {
 #define wsrep_OSU_method_get(T) wsrep_service->wsrep_OSU_method_get_func(T)
 #define wsrep_thd_has_ignored_error(T) wsrep_service->wsrep_thd_has_ignored_error_func(T)
 #define wsrep_thd_set_ignored_error(T,V) wsrep_service->wsrep_thd_set_ignored_error_func(T,V)
-#define wsrep_thd_set_wsrep_aborter(T) wsrep_service->wsrep_thd_set_wsrep_aborter_func(T1, T2)
 #define wsrep_report_bf_lock_wait(T,I) wsrep_service->wsrep_report_bf_lock_wait(T,I)
 #define wsrep_thd_set_PA_unsafe(T) wsrep_service->wsrep_thd_set_PA_unsafe_func(T)
 #else
@@ -175,6 +175,8 @@ void wsrep_set_data_home_dir(const char *data_dir);
 extern "C" my_bool wsrep_on(const MYSQL_THD thd);
 /* Lock thd wsrep lock */
 extern "C" void wsrep_thd_LOCK(const MYSQL_THD thd);
+/* Try thd wsrep lock. Return non-zero if lock could not be taken. */
+extern "C" int wsrep_thd_TRYLOCK(const MYSQL_THD thd);
 /* Unlock thd wsrep lock */
 extern "C" void wsrep_thd_UNLOCK(const MYSQL_THD thd);
 
@@ -197,8 +199,6 @@ extern "C" my_bool wsrep_thd_is_local(const MYSQL_THD thd);
 /* Return true if thd is in high priority mode */
 /* todo: rename to is_high_priority() */
 extern "C" my_bool wsrep_thd_is_applying(const MYSQL_THD thd);
-/* set wsrep_aborter for the target THD */
-extern "C" bool wsrep_thd_set_wsrep_aborter(MYSQL_THD bf_thd, MYSQL_THD victim_thd);
 /* Return true if thd is in TOI mode */
 extern "C" my_bool wsrep_thd_is_toi(const MYSQL_THD thd);
 /* Return true if thd is in replicating TOI mode */
@@ -249,7 +249,6 @@ extern "C" my_bool wsrep_thd_is_applying(const MYSQL_THD thd);
 extern "C" ulong wsrep_OSU_method_get(const MYSQL_THD thd);
 extern "C" my_bool wsrep_thd_has_ignored_error(const MYSQL_THD thd);
 extern "C" void wsrep_thd_set_ignored_error(MYSQL_THD thd, my_bool val);
-extern "C" bool wsrep_thd_set_wsrep_aborter(MYSQL_THD bf_thd, MYSQL_THD victim_thd);
 extern "C" void wsrep_report_bf_lock_wait(const THD *thd,
                                           unsigned long long trx_id);
 /* declare parallel applying unsafety for the THD */
diff --git a/mysql-test/suite/galera/disabled.def b/mysql-test/suite/galera/disabled.def
index bcf6078f624..84ecd3eb7fb 100644
--- a/mysql-test/suite/galera/disabled.def
+++ b/mysql-test/suite/galera/disabled.def
@@ -27,3 +27,5 @@ galera_bf_kill_debug : timeout after 900 seconds
 galera_ssl_upgrade : [Warning] Failed to load slave replication state from table mysql.gtid_slave_pos: 130: Incorrect file format 'gtid_slave_pos'
 galera_parallel_simple : timeout related to wsrep_sync_wait
 galera_insert_bulk : MDEV-30536 no expected deadlock in galera_insert_bulk test
+MDEV-27713 : test is using get_lock(), which is now rejected in cluster
+galera_bf_abort_group_commit : MDEV-30855 PR to remove the test exists
diff --git a/mysql-test/suite/galera/r/MDEV-29293.result b/mysql-test/suite/galera/r/MDEV-29293.result
new file mode 100644
index 00000000000..70c0cc84a31
--- /dev/null
+++ b/mysql-test/suite/galera/r/MDEV-29293.result
@@ -0,0 +1,21 @@
+connection node_2;
+connection node_1;
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+set wsrep_sync_wait = 0;
+CREATE TABLE t1(a int not null primary key auto_increment, b int) engine=InnoDB;
+INSERT INTO t1 VALUES (1,2);
+connection node_1a;
+BEGIN;
+UPDATE t1 SET b=3 WHERE a=1;
+connection node_1;
+set debug_sync='wsrep_kill_before_awake_no_mutex SIGNAL before_kill WAIT_FOR continue';
+connection node_1b;
+set debug_sync= 'now WAIT_FOR before_kill';
+connection node_2;
+UPDATE t1 SET b=7 WHERE a=1;
+connection node_1b;
+set debug_sync= 'now SIGNAL continue';
+connection node_1;
+DROP TABLE t1;
+SET DEBUG_SYNC= 'RESET';
diff --git a/mysql-test/suite/galera/r/galera_create_table_as_select.result b/mysql-test/suite/galera/r/galera_create_table_as_select.result
index 6f65ee99f0a..beda5f30fe2 100644
--- a/mysql-test/suite/galera/r/galera_create_table_as_select.result
+++ b/mysql-test/suite/galera/r/galera_create_table_as_select.result
@@ -82,6 +82,7 @@ connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
 LOCK TABLE t2 WRITE;
 connection node_1;
 CREATE TABLE t1 AS SELECT * FROM t2;;
+connection node_1a;
 connection node_2;
 SELECT COUNT(*) = 5 FROM t2;
 COUNT(*) = 5
diff --git a/mysql-test/suite/galera/r/galera_gcache_recover_manytrx.result b/mysql-test/suite/galera/r/galera_gcache_recover_manytrx.result
index 5caf22b39ca..5718807b5c4 100644
--- a/mysql-test/suite/galera/r/galera_gcache_recover_manytrx.result
+++ b/mysql-test/suite/galera/r/galera_gcache_recover_manytrx.result
@@ -134,6 +134,3 @@ connection node_1;
 call mtr.add_suppression("Error in Log_event::read_log_event():.*");
 CALL mtr.add_suppression("conflict state 7 after post commit");
 CALL mtr.add_suppression("Skipped GCache ring buffer recovery");
-connection node_2;
-call mtr.add_suppression("Error in Log_event::read_log_event():.*");
-CALL mtr.add_suppression("Skipped GCache ring buffer recovery");
diff --git a/mysql-test/suite/galera/r/galera_kill_group_commit.result b/mysql-test/suite/galera/r/galera_kill_group_commit.result
new file mode 100644
index 00000000000..bb59ce1486f
--- /dev/null
+++ b/mysql-test/suite/galera/r/galera_kill_group_commit.result
@@ -0,0 +1,27 @@
+connection node_2;
+connection node_1;
+connect node_1_kill, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connect node_1_ctrl, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+SET SESSION wsrep_sync_wait = 0;
+connect node_1_follower, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+SET SESSION wsrep_sync_wait = 0;
+connection node_1;
+CREATE TABLE t1 (f1 INT PRIMARY KEY) ENGINE=InnoDB;
+SET SESSION DEBUG_SYNC = "commit_before_enqueue SIGNAL leader_before_enqueue_reached WAIT_FOR leader_before_enqueue_continue";
+INSERT INTO t1 VALUES (1);
+connection node_1_ctrl;
+SET DEBUG_SYNC = "now WAIT_FOR leader_before_enqueue_reached";
+connection node_1_follower;
+INSERT INTO t1 VALUES (2);;
+connection node_1_ctrl;
+connection node_1_kill;
+# Execute KILL QUERY for group commit follower
+SET DEBUG_SYNC = "now SIGNAL leader_before_enqueue_continue";
+connection node_1_follower;
+connection node_1;
+SELECT * FROM t1;
+f1
+1
+2
+SET DEBUG_SYNC = "RESET";
+DROP TABLE t1;
diff --git a/mysql-test/suite/galera/r/galera_var_retry_autocommit.result b/mysql-test/suite/galera/r/galera_var_retry_autocommit.result
index 50667b0a4fa..eee740b6036 100644
--- a/mysql-test/suite/galera/r/galera_var_retry_autocommit.result
+++ b/mysql-test/suite/galera/r/galera_var_retry_autocommit.result
@@ -36,7 +36,10 @@ SET DEBUG_SYNC = 'now SIGNAL wsrep_retry_autocommit_continue';
 connection node_1;
 SELECT COUNT(*) FROM t1;
 COUNT(*)
-1
+connection node_1;
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+0
 SET DEBUG_SYNC = 'RESET';
 SET GLOBAL debug_dbug = NULL;
 DROP TABLE t1;
diff --git a/mysql-test/suite/galera/t/MDEV-29293.test b/mysql-test/suite/galera/t/MDEV-29293.test
new file mode 100644
index 00000000000..dacbf714c06
--- /dev/null
+++ b/mysql-test/suite/galera/t/MDEV-29293.test
@@ -0,0 +1,41 @@
+--source include/galera_cluster.inc
+--source include/have_innodb.inc
+--source include/have_debug_sync.inc
+--source include/galera_have_debug_sync.inc
+
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--connect node_1b, 127.0.0.1, root, , test, $NODE_MYPORT_1
+set wsrep_sync_wait = 0;
+
+CREATE TABLE t1(a int not null primary key auto_increment, b int) engine=InnoDB;
+INSERT INTO t1 VALUES (1,2); 
+
+--connection node_1a
+--let $victim_id = `SELECT CONNECTION_ID()`
+BEGIN;
+UPDATE t1 SET b=3 WHERE a=1;
+
+--connection node_1
+set debug_sync='wsrep_kill_before_awake_no_mutex SIGNAL before_kill WAIT_FOR continue';
+--disable_query_log
+--disable_result_log
+--send_eval KILL CONNECTION $victim_id 
+--enable_result_log
+--enable_query_log
+
+--connection node_1b
+set debug_sync= 'now WAIT_FOR before_kill';
+
+--connection node_2
+UPDATE t1 SET b=7 WHERE a=1;
+
+--connection node_1b 
+--let $wait_condition = SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST WHERE User = 'system user' AND  State LIKE 'Update_rows_log_event%'; 
+--source include/wait_condition.inc
+set debug_sync= 'now SIGNAL continue'; 
+
+--connection node_1
+--reap 
+DROP TABLE t1;
+SET DEBUG_SYNC= 'RESET';
+
diff --git a/mysql-test/suite/galera/t/galera_create_table_as_select.test b/mysql-test/suite/galera/t/galera_create_table_as_select.test
index a6c1f657280..cfee63e5e27 100644
--- a/mysql-test/suite/galera/t/galera_create_table_as_select.test
+++ b/mysql-test/suite/galera/t/galera_create_table_as_select.test
@@ -113,6 +113,10 @@ LOCK TABLE t2 WRITE;
 --connection node_1
 --send CREATE TABLE t1 AS SELECT * FROM t2;
 
+--connection node_1a
+--let $wait_condition = SELECT COUNT(*) = 1 FROM information_schema.processlist WHERE STATE LIKE 'Waiting for table metadata lock%'
+--source include/wait_condition.inc
+
 --connection node_2
 SELECT COUNT(*) = 5 FROM t2;
 CREATE TABLE t1 AS SELECT * FROM t2;
@@ -121,7 +125,7 @@ CREATE TABLE t1 AS SELECT * FROM t2;
 UNLOCK TABLES;
 
 --connection node_1
---error ER_TABLE_EXISTS_ERROR,ER_LOCK_DEADLOCK
+--error ER_TABLE_EXISTS_ERROR,ER_QUERY_INTERRUPTED
 --reap
 
 DROP TABLE t1, t2;
diff --git a/mysql-test/suite/galera/t/galera_kill_group_commit.cnf b/mysql-test/suite/galera/t/galera_kill_group_commit.cnf
new file mode 100644
index 00000000000..60f4f776409
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_kill_group_commit.cnf
@@ -0,0 +1,5 @@
+!include ../galera_2nodes.cnf
+
+[mysqld]
+log-bin
+log-slave-updates
diff --git a/mysql-test/suite/galera/t/galera_kill_group_commit.test b/mysql-test/suite/galera/t/galera_kill_group_commit.test
new file mode 100644
index 00000000000..4b84f2d90ef
--- /dev/null
+++ b/mysql-test/suite/galera/t/galera_kill_group_commit.test
@@ -0,0 +1,69 @@
+#
+# Verify that transaction which has reached group commit queue
+# cannot be killed. If the kill succeeds, assertion for
+# wsrep transaction state will fail.
+#
+# If the bug is present, i.e. wsrep transaction gets killed during
+# group commit wait, this test is enough to reproduce the crash
+# most of the time.
+#
+
+--source include/have_innodb.inc
+--source include/have_debug_sync.inc
+--source include/galera_cluster.inc
+
+# Connection for KILL commands
+--connect node_1_kill, 127.0.0.1, root, , test, $NODE_MYPORT_1
+# Connection for sync point control
+--connect node_1_ctrl, 127.0.0.1, root, , test, $NODE_MYPORT_1
+SET SESSION wsrep_sync_wait = 0;
+# Connection for group commit follower
+--connect node_1_follower, 127.0.0.1, root, , test, $NODE_MYPORT_1
+# Need to disable sync wait to reach commit queue when leader
+# is blocked.
+SET SESSION wsrep_sync_wait = 0;
+--let $follower_id = `SELECT CONNECTION_ID()`
+
+--connection node_1
+CREATE TABLE t1 (f1 INT PRIMARY KEY) ENGINE=InnoDB;
+
+SET SESSION DEBUG_SYNC = "commit_before_enqueue SIGNAL leader_before_enqueue_reached WAIT_FOR leader_before_enqueue_continue";
+--send INSERT INTO t1 VALUES (1)
+
+--connection node_1_ctrl
+SET DEBUG_SYNC = "now WAIT_FOR leader_before_enqueue_reached";
+
+--connection node_1_follower
+# SET SESSION DEBUG_SYNC = "group_commit_waiting_for_prior SIGNAL follower_waiting_for_prior_reached WAIT_FOR follower_waiting_for_prior_continue";
+--send INSERT INTO t1 VALUES (2);
+
+--connection node_1_ctrl
+# TODO: Is it possible to use sync points to enforce group commit to happen?
+# The leader will hold commit monitor in commit_before_enqueue sync point,
+# which prevents the follower to reach the group commit wait state.
+# We now sleep and expect the follower to reach group commit, but this
+# may cause false negatives.
+--sleep 1
+
+--connection node_1_kill
+--echo # Execute KILL QUERY for group commit follower
+--disable_query_log
+--disable_result_log
+# Because it is currently impossible to verify that the
+# follower has reached group commit queue, the KILL may
+# sometimes return success.
+--error 0,ER_KILL_DENIED_ERROR
+--eval KILL QUERY $follower_id
+--enable_result_log
+--enable_query_log
+
+SET DEBUG_SYNC = "now SIGNAL leader_before_enqueue_continue";
+--connection node_1_follower
+--reap
+
+--connection node_1
+--reap
+SELECT * FROM t1;
+
+SET DEBUG_SYNC = "RESET";
+DROP TABLE t1;
diff --git a/mysql-test/suite/galera/t/galera_var_retry_autocommit.test b/mysql-test/suite/galera/t/galera_var_retry_autocommit.test
index c58eba1410e..8009fe88c65 100644
--- a/mysql-test/suite/galera/t/galera_var_retry_autocommit.test
+++ b/mysql-test/suite/galera/t/galera_var_retry_autocommit.test
@@ -64,6 +64,7 @@ SELECT COUNT(*) FROM t1;
 SET DEBUG_SYNC = 'now SIGNAL wsrep_retry_autocommit_continue';
 
 --connection node_1
+--error 0,ER_LOCK_DEADLOCK
 --reap
 SELECT COUNT(*) FROM t1;
 
diff --git a/mysql-test/suite/innodb/r/monitor.result b/mysql-test/suite/innodb/r/monitor.result
index 221c92faaea..174fd530c72 100644
--- a/mysql-test/suite/innodb/r/monitor.result
+++ b/mysql-test/suite/innodb/r/monitor.result
@@ -1,10 +1,9 @@
-set global innodb_monitor_disable = All;
 select name, if(enabled,'enabled','disabled') status
 from information_schema.innodb_metrics;
 name	status
 metadata_table_handles_opened	disabled
-lock_deadlocks	disabled
-lock_timeouts	disabled
+lock_deadlocks	enabled
+lock_timeouts	enabled
 lock_rec_lock_waits	disabled
 lock_table_lock_waits	disabled
 lock_rec_lock_requests	disabled
@@ -14,32 +13,32 @@ lock_rec_locks	disabled
 lock_table_lock_created	disabled
 lock_table_lock_removed	disabled
 lock_table_locks	disabled
-lock_row_lock_current_waits	disabled
-lock_row_lock_time	disabled
-lock_row_lock_time_max	disabled
-lock_row_lock_waits	disabled
-lock_row_lock_time_avg	disabled
-buffer_pool_size	disabled
-buffer_pool_reads	disabled
-buffer_pool_read_requests	disabled
-buffer_pool_write_requests	disabled
-buffer_pool_wait_free	disabled
-buffer_pool_read_ahead	disabled
-buffer_pool_read_ahead_evicted	disabled
-buffer_pool_pages_total	disabled
-buffer_pool_pages_misc	disabled
-buffer_pool_pages_data	disabled
-buffer_pool_bytes_data	disabled
-buffer_pool_pages_dirty	disabled
-buffer_pool_bytes_dirty	disabled
-buffer_pool_pages_free	disabled
-buffer_pages_created	disabled
-buffer_pages_written	disabled
-buffer_pages_read	disabled
-buffer_index_sec_rec_cluster_reads	disabled
-buffer_index_sec_rec_cluster_reads_avoided	disabled
-buffer_data_reads	disabled
-buffer_data_written	disabled
+lock_row_lock_current_waits	enabled
+lock_row_lock_time	enabled
+lock_row_lock_time_max	enabled
+lock_row_lock_waits	enabled
+lock_row_lock_time_avg	enabled
+buffer_pool_size	enabled
+buffer_pool_reads	enabled
+buffer_pool_read_requests	enabled
+buffer_pool_write_requests	enabled
+buffer_pool_wait_free	enabled
+buffer_pool_read_ahead	enabled
+buffer_pool_read_ahead_evicted	enabled
+buffer_pool_pages_total	enabled
+buffer_pool_pages_misc	enabled
+buffer_pool_pages_data	enabled
+buffer_pool_bytes_data	enabled
+buffer_pool_pages_dirty	enabled
+buffer_pool_bytes_dirty	enabled
+buffer_pool_pages_free	enabled
+buffer_pages_created	enabled
+buffer_pages_written	enabled
+buffer_pages_read	enabled
+buffer_index_sec_rec_cluster_reads	enabled
+buffer_index_sec_rec_cluster_reads_avoided	enabled
+buffer_data_reads	enabled
+buffer_data_written	enabled
 buffer_flush_batch_scanned	disabled
 buffer_flush_batch_num_scan	disabled
 buffer_flush_batch_scanned_per_call	disabled
@@ -72,8 +71,8 @@ buffer_flush_background_pages	disabled
 buffer_LRU_batch_scanned	disabled
 buffer_LRU_batch_num_scan	disabled
 buffer_LRU_batch_scanned_per_call	disabled
-buffer_LRU_batch_flush_total_pages	disabled
-buffer_LRU_batch_evict_total_pages	disabled
+buffer_LRU_batch_flush_total_pages	enabled
+buffer_LRU_batch_evict_total_pages	enabled
 buffer_LRU_single_flush_failure_count	disabled
 buffer_LRU_get_free_search	disabled
 buffer_LRU_search_scanned	disabled
@@ -114,21 +113,21 @@ buffer_page_written_blob	disabled
 buffer_page_written_zblob	disabled
 buffer_page_written_zblob2	disabled
 buffer_page_written_other	disabled
-os_data_reads	disabled
-os_data_writes	disabled
-os_data_fsyncs	disabled
-os_pending_reads	disabled
-os_pending_writes	disabled
-os_log_bytes_written	disabled
+os_data_reads	enabled
+os_data_writes	enabled
+os_data_fsyncs	enabled
+os_pending_reads	enabled
+os_pending_writes	enabled
+os_log_bytes_written	enabled
 trx_rw_commits	disabled
 trx_ro_commits	disabled
 trx_nl_ro_commits	disabled
 trx_commits_insert_update	disabled
 trx_rollbacks	disabled
 trx_rollbacks_savepoint	disabled
-trx_rseg_history_len	disabled
+trx_rseg_history_len	enabled
 trx_undo_slots_used	disabled
-trx_undo_slots_cached	disabled
+trx_undo_slots_cached	enabled
 trx_rseg_current_size	disabled
 purge_del_mark_records	disabled
 purge_upd_exist_or_extern_records	disabled
@@ -144,9 +143,9 @@ log_lsn_current	disabled
 log_lsn_checkpoint_age	disabled
 log_lsn_buf_pool_oldest	disabled
 log_max_modified_age_async	disabled
-log_waits	disabled
-log_write_requests	disabled
-log_writes	disabled
+log_waits	enabled
+log_write_requests	enabled
+log_writes	enabled
 compress_pages_compressed	disabled
 compress_pages_decompressed	disabled
 compression_pad_increments	disabled
@@ -164,42 +163,42 @@ index_page_merge_successful	disabled
 index_page_reorg_attempts	disabled
 index_page_reorg_successful	disabled
 index_page_discards	disabled
-adaptive_hash_searches	disabled
-adaptive_hash_searches_btree	disabled
+adaptive_hash_searches	enabled
+adaptive_hash_searches_btree	enabled
 adaptive_hash_pages_added	disabled
 adaptive_hash_pages_removed	disabled
 adaptive_hash_rows_added	disabled
 adaptive_hash_rows_removed	disabled
 adaptive_hash_rows_deleted_no_hash_entry	disabled
 adaptive_hash_rows_updated	disabled
-file_num_open_files	disabled
-ibuf_merges_insert	disabled
-ibuf_merges_delete_mark	disabled
-ibuf_merges_delete	disabled
-ibuf_merges_discard_insert	disabled
-ibuf_merges_discard_delete_mark	disabled
-ibuf_merges_discard_delete	disabled
-ibuf_merges	disabled
-ibuf_size	disabled
+file_num_open_files	enabled
+ibuf_merges_insert	enabled
+ibuf_merges_delete_mark	enabled
+ibuf_merges_delete	enabled
+ibuf_merges_discard_insert	enabled
+ibuf_merges_discard_delete_mark	enabled
+ibuf_merges_discard_delete	enabled
+ibuf_merges	enabled
+ibuf_size	enabled
 innodb_master_thread_sleeps	disabled
-innodb_activity_count	disabled
+innodb_activity_count	enabled
 innodb_master_active_loops	disabled
 innodb_master_idle_loops	disabled
 innodb_log_flush_usec	disabled
 innodb_dict_lru_usec	disabled
 innodb_dict_lru_count_active	disabled
 innodb_dict_lru_count_idle	disabled
-innodb_dblwr_writes	disabled
-innodb_dblwr_pages_written	disabled
-innodb_page_size	disabled
+innodb_dblwr_writes	enabled
+innodb_dblwr_pages_written	enabled
+innodb_page_size	enabled
 dml_reads	disabled
-dml_inserts	disabled
-dml_deletes	disabled
-dml_updates	disabled
-dml_system_reads	disabled
-dml_system_inserts	disabled
-dml_system_deletes	disabled
-dml_system_updates	disabled
+dml_inserts	enabled
+dml_deletes	enabled
+dml_updates	enabled
+dml_system_reads	enabled
+dml_system_inserts	enabled
+dml_system_deletes	enabled
+dml_system_updates	enabled
 ddl_background_drop_indexes	disabled
 ddl_online_create_index	disabled
 ddl_pending_alter_table	disabled
@@ -209,6 +208,9 @@ icp_attempts	disabled
 icp_no_match	disabled
 icp_out_of_range	disabled
 icp_match	disabled
+set global innodb_monitor_disable = All;
+select name from information_schema.innodb_metrics where enabled;
+name
 set global innodb_monitor_enable = all;
 select name from information_schema.innodb_metrics where not enabled;
 name
diff --git a/mysql-test/suite/innodb/t/monitor.test b/mysql-test/suite/innodb/t/monitor.test
index d6fa3f2fbc9..65a93e5a97a 100644
--- a/mysql-test/suite/innodb/t/monitor.test
+++ b/mysql-test/suite/innodb/t/monitor.test
@@ -5,12 +5,14 @@
 # sys_vars.innodb_monitor_enable_basic
 
 --source include/have_innodb.inc
-set global innodb_monitor_disable = All;
 # Test turn on/off the monitor counter  with "all" option
 # By default, they will be off.
 select name, if(enabled,'enabled','disabled') status
 from information_schema.innodb_metrics;
 
+set global innodb_monitor_disable = All;
+select name from information_schema.innodb_metrics where enabled;
+
 # Turn on all monitor counters
 set global innodb_monitor_enable = all;
 
diff --git a/sql/handler.cc b/sql/handler.cc
index a12e9ea18f5..ae87c3a0e00 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -7972,6 +7972,9 @@ Compare_keys handler::compare_key_parts(const Field &old_field,
   concurrent accesses. And it's an overkill to take LOCK_plugin and
   iterate the whole installed_htons[] array every time.
 
+  @note Object victim_thd is not guaranteed to exist after this
+        function returns.
+
   @param bf_thd       brute force THD asking for the abort
   @param victim_thd   victim THD to be aborted
 
@@ -7985,6 +7988,8 @@ int ha_abort_transaction(THD *bf_thd, THD *victim_thd, my_bool signal)
   if (!WSREP(bf_thd) &&
       !(bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU &&
         wsrep_thd_is_toi(bf_thd))) {
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
     DBUG_RETURN(0);
   }
 
@@ -7996,6 +8001,8 @@ int ha_abort_transaction(THD *bf_thd, THD *victim_thd, my_bool signal)
   else
   {
     WSREP_WARN("Cannot abort InnoDB transaction");
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
   }
 
   DBUG_RETURN(0);
diff --git a/sql/handler.h b/sql/handler.h
index 523c8472784..9d03baf1c0d 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -45,6 +45,7 @@
 #include "sql_sequence.h"
 #include "mem_root_array.h"
 #include <utility>     // pair
+#include <my_attribute.h> /* __attribute__ */
 
 class Alter_info;
 class Virtual_column_info;
@@ -1460,9 +1461,9 @@ struct handlerton
                             const char *query, uint query_length,
                             const char *db, const char *table_name);
 
-   void (*abort_transaction)(handlerton *hton, THD *bf_thd,
-			    THD *victim_thd, my_bool signal);
-   int (*set_checkpoint)(handlerton *hton, const XID* xid);
+   void (*abort_transaction)(handlerton *hton, THD *bf_thd, THD *victim_thd,
+                             my_bool signal) __attribute__((nonnull));
+   int (*set_checkpoint)(handlerton *hton, const XID *xid);
    int (*get_checkpoint)(handlerton *hton, XID* xid);
   /**
      Check if the version of the table matches the version in the .frm
diff --git a/sql/service_wsrep.cc b/sql/service_wsrep.cc
index dd12149ff48..e1a4a25b27a 100644
--- a/sql/service_wsrep.cc
+++ b/sql/service_wsrep.cc
@@ -32,6 +32,11 @@ extern "C" void wsrep_thd_LOCK(const THD *thd)
   mysql_mutex_lock(&thd->LOCK_thd_data);
 }
 
+extern "C" int wsrep_thd_TRYLOCK(const THD *thd)
+{
+  return mysql_mutex_trylock(&thd->LOCK_thd_data);
+}
+
 extern "C" void wsrep_thd_UNLOCK(const THD *thd)
 {
   mysql_mutex_unlock(&thd->LOCK_thd_data);
@@ -196,6 +201,7 @@ extern "C" void wsrep_handle_SR_rollback(THD *bf_thd,
 
   /* Note: do not store/reset globals before wsrep_bf_abort() call
      to avoid losing BF thd context. */
+  mysql_mutex_lock(&victim_thd->LOCK_thd_data);
   if (!(bf_thd && bf_thd != victim_thd))
   {
     DEBUG_SYNC(victim_thd, "wsrep_before_SR_rollback");
@@ -208,6 +214,7 @@ extern "C" void wsrep_handle_SR_rollback(THD *bf_thd,
   {
     wsrep_thd_self_abort(victim_thd);
   }
+  mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
   if (bf_thd)
   {
     wsrep_store_threadvars(bf_thd);
@@ -218,7 +225,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,19 +234,8 @@ 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_HARD);
-    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
   } else {
     WSREP_DEBUG("wsrep_thd_bf_abort skipped awake, signal %d", signal);
   }
@@ -368,25 +364,6 @@ extern "C" ulong wsrep_OSU_method_get(const MYSQL_THD thd)
     return(global_system_variables.wsrep_OSU_method);
 }
 
-extern "C" bool wsrep_thd_set_wsrep_aborter(THD *bf_thd, THD *victim_thd)
-{
-  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
-  if (!bf_thd)
-  {
-    victim_thd->wsrep_aborter= 0;
-    WSREP_DEBUG("wsrep_thd_set_wsrep_aborter resetting wsrep_aborter");
-    return false;
-  }
-  if (victim_thd->wsrep_aborter && victim_thd->wsrep_aborter != bf_thd->thread_id)
-  {
-    return true;
-  }
-  victim_thd->wsrep_aborter= bf_thd->thread_id;
-  WSREP_DEBUG("wsrep_thd_set_wsrep_aborter setting wsrep_aborter %u",
-              victim_thd->wsrep_aborter);
-  return false;
-}
-
 extern "C" void wsrep_report_bf_lock_wait(const THD *thd,
                                           unsigned long long trx_id)
 {
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 532696345a7..6619a867318 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1298,6 +1298,11 @@ void THD::init()
   wsrep_affected_rows     = 0;
   m_wsrep_next_trx_id     = WSREP_UNDEFINED_TRX_ID;
   wsrep_aborter           = 0;
+  wsrep_abort_by_kill     = NOT_KILLED;
+  wsrep_abort_by_kill_err = 0;
+#ifndef DBUG_OFF
+  wsrep_killed_state      = 0;
+#endif /* DBUG_OFF */
   wsrep_desynced_backup_stage= false;
 #endif /* WITH_WSREP */
 
@@ -1646,6 +1651,13 @@ void THD::reset_for_reuse()
 #endif
 #ifdef WITH_WSREP
   wsrep_free_status(this);
+  wsrep_cs().reset_error();
+  wsrep_aborter= 0;
+  wsrep_abort_by_kill= NOT_KILLED;
+  wsrep_abort_by_kill_err= 0;
+#ifndef DBUG_OFF
+  wsrep_killed_state= 0;
+#endif /* DBUG_OFF */
 #endif /* WITH_WSREP */
 }
 
@@ -1902,7 +1914,9 @@ void THD::awake_no_mutex(killed_state state_to_set)
   }
 
   /* Interrupt target waiting inside a storage engine. */
-  if (state_to_set != NOT_KILLED  && !wsrep_is_bf_aborted(this))
+  if (state_to_set != NOT_KILLED &&
+      IF_WSREP(!wsrep_is_bf_aborted(this) && wsrep_abort_by_kill == NOT_KILLED,
+               true))
     ha_kill_query(this, thd_kill_level(this));
 
   abort_current_cond_wait(false);
@@ -2129,6 +2143,17 @@ void THD::reset_killed()
     mysql_mutex_unlock(&LOCK_thd_kill);
   }
 #ifdef WITH_WSREP
+  if (WSREP_NNULL(this))
+  {
+    if (wsrep_abort_by_kill != NOT_KILLED)
+    {
+      mysql_mutex_assert_not_owner(&LOCK_thd_kill);
+      mysql_mutex_lock(&LOCK_thd_kill);
+      wsrep_abort_by_kill= NOT_KILLED;
+      wsrep_abort_by_kill_err= 0;
+      mysql_mutex_unlock(&LOCK_thd_kill);
+    }
+  }
   mysql_mutex_assert_not_owner(&LOCK_thd_data);
   mysql_mutex_lock(&LOCK_thd_data);
   wsrep_aborter= 0;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 5e5c0128a05..11f18823cea 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -5405,7 +5405,14 @@ public:
   bool                      wsrep_ignore_table;
   /* thread who has started kill for this THD protected by LOCK_thd_data*/
   my_thread_id              wsrep_aborter;
-
+  /* Kill signal used, if thread was killed by manual KILL. Protected by
+     LOCK_thd_kill. */
+  std::atomic<killed_state> wsrep_abort_by_kill;
+  /* */
+  struct err_info*          wsrep_abort_by_kill_err;
+#ifndef DBUG_OFF
+  int                       wsrep_killed_state;
+#endif /* DBUG_OFF */
   /* true if BF abort is observed in do_command() right after reading
   client's packet, and if the client has sent PS execute command. */
   bool                      wsrep_delayed_BF_abort;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 7589dc51a12..7adafab445f 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -7892,7 +7892,7 @@ static bool wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
           thd->wsrep_retry_counter < thd->variables.wsrep_retry_autocommit)
       {
 #ifdef ENABLED_DEBUG_SYNC
-	DBUG_EXECUTE_IF("sync.wsrep_retry_autocommit",
+        DBUG_EXECUTE_IF("sync.wsrep_retry_autocommit",
                     {
                       const char act[]=
                         "now "
@@ -9255,23 +9255,20 @@ kill_one_thread(THD *thd, my_thread_id id, killed_state kill_signal, killed_type
         thd->security_ctx->user_matches(tmp->security_ctx))
 #endif /* WITH_WSREP */
     {
+      {
 #ifdef WITH_WSREP
-      DEBUG_SYNC(thd, "before_awake_no_mutex");
-      if (tmp->wsrep_aborter && tmp->wsrep_aborter != thd->thread_id)
-      {
-        /* victim is in hit list already, bail out */
-	WSREP_DEBUG("victim %lld has wsrep aborter: %lu, skipping awake()",
-		    id, tmp->wsrep_aborter);
-        error= 0;
-      }
-      else
+        if (WSREP(tmp))
+        {
+          error = wsrep_kill_thd(thd, tmp, kill_signal);
+        }
+        else
+        {
 #endif /* WITH_WSREP */
-      {
-        WSREP_DEBUG("kill_one_thread victim: %lld wsrep_aborter %lu"
-                    " by signal %d",
-                    id, tmp->wsrep_aborter, kill_signal);
         tmp->awake_no_mutex(kill_signal);
         error= 0;
+#ifdef WITH_WSREP
+        }
+#endif /* WITH_WSREP */
       }
     }
     else
@@ -9394,18 +9391,6 @@ static
 void sql_kill(THD *thd, my_thread_id 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)
@@ -9415,11 +9400,6 @@ void sql_kill(THD *thd, my_thread_id id, killed_state state, killed_type type)
   }
   else
     my_error(error, MYF(0), id);
-#ifdef WITH_WSREP
-  return;
- wsrep_error_label:
-  my_error(ER_KILL_DENIED_ERROR, MYF(0), (long long) thd->thread_id);
-#endif /* WITH_WSREP */
 }
 
 
@@ -9428,18 +9408,6 @@ 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 */
   switch (error= kill_threads_for_user(thd, user, state, &rows))
   {
   case 0:
@@ -9455,11 +9423,6 @@ sql_kill_user(THD *thd, LEX_USER *user, killed_state state)
   default:
     my_error(error, MYF(0));
   }
-#ifdef WITH_WSREP
-  return;
- wsrep_error_label:
-  my_error(ER_KILL_DENIED_ERROR, MYF(0), (long long) thd->thread_id);
-#endif /* WITH_WSREP */
 }
 
 
diff --git a/sql/sql_plugin_services.inl b/sql/sql_plugin_services.inl
index 3a66e982e7b..f2b2d08de1d 100644
--- a/sql/sql_plugin_services.inl
+++ b/sql/sql_plugin_services.inl
@@ -151,6 +151,7 @@ static struct wsrep_service_st wsrep_handler = {
   wsrep_on,
   wsrep_prepare_key_for_innodb,
   wsrep_thd_LOCK,
+  wsrep_thd_TRYLOCK,
   wsrep_thd_UNLOCK,
   wsrep_thd_query,
   wsrep_thd_retry_counter,
@@ -179,7 +180,6 @@ static struct wsrep_service_st wsrep_handler = {
   wsrep_OSU_method_get,
   wsrep_thd_has_ignored_error,
   wsrep_thd_set_ignored_error,
-  wsrep_thd_set_wsrep_aborter,
   wsrep_report_bf_lock_wait,
   wsrep_thd_kill_LOCK,
   wsrep_thd_kill_UNLOCK,
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index 9bfaf9285f3..e1508884075 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -56,6 +56,11 @@ my_bool wsrep_on(const THD *)
 void wsrep_thd_LOCK(const THD *)
 { }
 
+int wsrep_thd_TRYLOCK(const THD *)
+{
+  return 0;
+}
+
 void wsrep_thd_UNLOCK(const THD *)
 { }
 
@@ -154,8 +159,6 @@ void wsrep_thd_set_ignored_error(THD*, my_bool)
 { }
 ulong wsrep_OSU_method_get(const THD*)
 { return 0;}
-bool wsrep_thd_set_wsrep_aborter(THD*, THD*)
-{ return 0;}
 
 void wsrep_report_bf_lock_wait(const THD*,
                                unsigned long long)
diff --git a/sql/wsrep_high_priority_service.cc b/sql/wsrep_high_priority_service.cc
index 53ef20f3e78..0a2fa273723 100644
--- a/sql/wsrep_high_priority_service.cc
+++ b/sql/wsrep_high_priority_service.cc
@@ -510,6 +510,7 @@ int Wsrep_high_priority_service::log_dummy_write_set(const wsrep::ws_handle& ws_
       m_thd->wait_for_prior_commit();
     }
 
+    WSREP_DEBUG("checkpointing dummy write set %lld", ws_meta.seqno().get());
     wsrep_set_SE_checkpoint(ws_meta.gtid(), wsrep_gtid_server.gtid());
 
     if (!WSREP_EMULATE_BINLOG(m_thd))
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index 00e3413ecd3..1f446868653 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -52,6 +52,7 @@
 #include "log_event.h"
 #include "sql_connect.h"
 #include "thread_cache.h"
+#include "debug_sync.h"
 
 #include <sstream>
 
@@ -3059,6 +3060,20 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                   request_thd, granted_thd);
     ticket->wsrep_report(wsrep_debug);
 
+    DEBUG_SYNC(request_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(request_thd, STRING_WITH_LEN(act)));
+    };);
+
+    /* Here we will call wsrep_abort_transaction so we should hold
+    THD::LOCK_thd_data to protect victim from concurrent usage
+    and THD::LOCK_thd_kill to protect from disconnect or delete.
+
+    */
+    mysql_mutex_lock(&granted_thd->LOCK_thd_kill);
     mysql_mutex_lock(&granted_thd->LOCK_thd_data);
 
     if (wsrep_thd_is_toi(granted_thd) ||
@@ -3070,13 +3085,11 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                     wsrep_thd_query(request_thd));
         THD_STAGE_INFO(request_thd, stage_waiting_isolation);
         ticket->wsrep_report(wsrep_debug);
-        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       }
       else if (wsrep_thd_is_SR(granted_thd) && !wsrep_thd_is_SR(request_thd))
       {
         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_DEBUG("wsrep_handle_mdl_conflict DDL vs SR for %s",
                     wsrep_thd_query(request_thd));
         THD_STAGE_INFO(request_thd, stage_waiting_isolation);
@@ -3088,6 +3101,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);
       }
     }
@@ -3098,7 +3112,6 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                   wsrep_thd_query(request_thd));
       THD_STAGE_INFO(request_thd, stage_waiting_ddl);
       ticket->wsrep_report(wsrep_debug);
-      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       if (granted_thd->current_backup_stage != BACKUP_FINISHED &&
 	  wsrep_check_mode(WSREP_MODE_BF_MARIABACKUP))
       {
@@ -3112,7 +3125,6 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                   wsrep_thd_query(request_thd));
       THD_STAGE_INFO(request_thd, stage_waiting_isolation);
       ticket->wsrep_report(wsrep_debug);
-      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
       wsrep_abort_thd(request_thd, granted_thd, 1);
     }
     else
@@ -3126,7 +3138,6 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
 
       if (granted_thd->wsrep_trx().active())
       {
-        mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
         wsrep_abort_thd(request_thd, granted_thd, 1);
       }
       else
@@ -3135,10 +3146,9 @@ 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))
         {
+          granted_thd->awake_no_mutex(KILL_QUERY_HARD);
           ha_abort_transaction(request_thd, granted_thd, TRUE);
         }
         else
@@ -3147,10 +3157,14 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
                         schema, schema_len,
                         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);
         }
       }
     }
+    mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
+    mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
   }
   else
   {
@@ -3162,13 +3176,17 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
 static bool abort_replicated(THD *thd)
 {
   bool ret_code= false;
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  mysql_mutex_lock(&thd->LOCK_thd_data);
   if (thd->wsrep_trx().state() == wsrep::transaction::s_committing)
   {
     WSREP_DEBUG("aborting replicated trx: %llu", (ulonglong)(thd->real_id));
 
-    (void)wsrep_abort_thd(thd, thd, TRUE);
+    wsrep_abort_thd(thd, thd, TRUE);
     ret_code= true;
   }
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
   return ret_code;
 }
 
diff --git a/sql/wsrep_server_service.cc b/sql/wsrep_server_service.cc
index 9be6af71c56..52a0a9753c1 100644
--- a/sql/wsrep_server_service.cc
+++ b/sql/wsrep_server_service.cc
@@ -148,9 +148,13 @@ void Wsrep_server_service::release_high_priority_service(wsrep::high_priority_se
   wsrep_delete_threadvars();
 }
 
-void Wsrep_server_service::background_rollback(wsrep::client_state& client_state)
+void Wsrep_server_service::background_rollback(
+    wsrep::unique_lock<wsrep::mutex> &lock WSREP_UNUSED,
+    wsrep::client_state &client_state)
 {
-  Wsrep_client_state& cs= static_cast<Wsrep_client_state&>(client_state);
+  DBUG_ASSERT(lock.owns_lock());
+  Wsrep_client_state &cs= static_cast<Wsrep_client_state &>(client_state);
+  mysql_mutex_assert_owner(&cs.thd()->LOCK_thd_data);
   wsrep_fire_rollbacker(cs.thd());
 }
 
diff --git a/sql/wsrep_server_service.h b/sql/wsrep_server_service.h
index 168e98206e3..0fc48402024 100644
--- a/sql/wsrep_server_service.h
+++ b/sql/wsrep_server_service.h
@@ -46,7 +46,8 @@ public:
 
   void release_high_priority_service(wsrep::high_priority_service*);
 
-  void background_rollback(wsrep::client_state&);
+  void background_rollback(wsrep::unique_lock<wsrep::mutex> &,
+                           wsrep::client_state &);
 
   void bootstrap();
   void log_message(enum wsrep::log::level, const char*);
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index 420a25dd2ae..682e64859b4 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -307,48 +307,9 @@ void wsrep_fire_rollbacker(THD *thd)
   }
 }
 
-
-int wsrep_abort_thd(THD *bf_thd,
-                    THD *victim_thd,
-                    my_bool signal)
+static bool wsrep_bf_abort_low(THD *bf_thd, THD *victim_thd)
 {
-  DBUG_ENTER("wsrep_abort_thd");
-
-  mysql_mutex_lock(&victim_thd->LOCK_thd_data);
-
-  /* Note that when you use RSU node is desynced from cluster, thus WSREP(thd)
-  might not be true.
-  */
-  if ((WSREP_NNULL(bf_thd) ||
-       ((WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU) &&
-	 wsrep_thd_is_toi(bf_thd))) &&
-       !wsrep_thd_is_aborting(victim_thd))
-  {
-      WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu",
-                  (long long)bf_thd->real_id, (long long)victim_thd->real_id);
-      mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
-      ha_abort_transaction(bf_thd, victim_thd, signal);
-      DBUG_RETURN(1);
-  }
-  else
-  {
-    WSREP_DEBUG("wsrep_abort_thd not effective: bf %llu victim %llu "
-                "wsrep %d wsrep_on %d RSU %d TOI %d aborting %d",
-                (long long)bf_thd->real_id, (long long)victim_thd->real_id,
-                WSREP_NNULL(bf_thd), WSREP_ON,
-                bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU,
-                wsrep_thd_is_toi(bf_thd),
-                wsrep_thd_is_aborting(victim_thd));
-  }
-
-  mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
-  DBUG_RETURN(1);
-}
-
-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);
 
 #ifdef ENABLED_DEBUG_SYNC
   DBUG_EXECUTE_IF("sync.wsrep_bf_abort",
@@ -362,6 +323,85 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
                   };);
 #endif
 
+  wsrep::seqno bf_seqno(bf_thd->wsrep_trx().ws_meta().seqno());
+  bool ret;
+
+  {
+    /* Adopt the lock, it is being held by the caller. */
+    Wsrep_mutex wsm{&victim_thd->LOCK_thd_data};
+    wsrep::unique_lock<wsrep::mutex> lock{wsm, std::adopt_lock};
+
+    if (wsrep_thd_is_toi(bf_thd))
+    {
+      ret= victim_thd->wsrep_cs().total_order_bf_abort(lock, bf_seqno);
+    }
+    else
+    {
+      DBUG_ASSERT(WSREP(victim_thd) ? victim_thd->wsrep_trx().active() : 1);
+      ret= victim_thd->wsrep_cs().bf_abort(lock, bf_seqno);
+    }
+    if (ret)
+    {
+      /* BF abort should be allowed only once by wsrep-lib.*/
+      DBUG_ASSERT(victim_thd->wsrep_aborter == 0);
+      victim_thd->wsrep_aborter= bf_thd->thread_id;
+      wsrep_bf_aborts_counter++;
+    }
+    lock.release(); /* No unlock at the end of the scope. */
+  }
+
+  /* Sanity check for wsrep-lib calls to return with LOCK_thd_data held. */
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+
+  return ret;
+}
+
+void wsrep_abort_thd(THD *bf_thd,
+                    THD *victim_thd,
+                    my_bool signal)
+{
+  DBUG_ENTER("wsrep_abort_thd");
+
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+
+  /* Note that when you use RSU node is desynced from cluster, thus WSREP(thd)
+  might not be true.
+  */
+  if ((WSREP(bf_thd)
+       || ((WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU)
+           &&  wsrep_thd_is_toi(bf_thd))
+       || bf_thd->lex->sql_command == SQLCOM_KILL)
+      && !wsrep_thd_is_aborting(victim_thd) &&
+      wsrep_bf_abort_low(bf_thd, victim_thd) &&
+      !victim_thd->wsrep_cs().is_rollbacker_active())
+  {
+    WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu",
+                (long long)bf_thd->real_id, (long long)victim_thd->real_id);
+    victim_thd->awake_no_mutex(KILL_QUERY_HARD);
+    ha_abort_transaction(bf_thd, victim_thd, signal);
+  }
+  else
+  {
+    WSREP_DEBUG("wsrep_abort_thd not effective: bf %llu victim %llu "
+                "wsrep %d wsrep_on %d RSU %d TOI %d aborting %d",
+                (long long)bf_thd->real_id, (long long)victim_thd->real_id,
+                WSREP_NNULL(bf_thd), WSREP_ON,
+                bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU,
+                wsrep_thd_is_toi(bf_thd),
+                wsrep_thd_is_aborting(victim_thd));
+  }
+
+  DBUG_VOID_RETURN;
+}
+
+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);
+
   if (WSREP(victim_thd) && !victim_thd->wsrep_trx().active())
   {
     WSREP_DEBUG("wsrep_bf_abort, BF abort for non active transaction."
@@ -384,30 +424,84 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
         wsrep_check_mode(WSREP_MODE_BF_MARIABACKUP))
     {
       WSREP_DEBUG("killing connection for non wsrep session");
-      mysql_mutex_lock(&victim_thd->LOCK_thd_data);
       victim_thd->awake_no_mutex(KILL_CONNECTION);
-      mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
     }
     return false;
   }
 
-  bool ret;
-  wsrep::seqno bf_seqno(bf_thd->wsrep_trx().ws_meta().seqno());
+  return wsrep_bf_abort_low(bf_thd, victim_thd);
+}
 
-  if (wsrep_thd_is_toi(bf_thd))
+uint wsrep_kill_thd(THD *thd, THD *victim_thd, killed_state kill_signal)
+{
+  DBUG_ENTER("wsrep_kill_thd");
+  DBUG_ASSERT(WSREP(victim_thd));
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+  using trans= wsrep::transaction;
+  auto trx_state= victim_thd->wsrep_trx().state();
+#ifndef DBUG_OFF
+  victim_thd->wsrep_killed_state= trx_state;
+#endif /* DBUG_OFF */
+  /*
+    Already killed or in commit codepath. Mark the victim as killed,
+    the killed status will be restored in wsrep_after_commit() and
+    will be processed after the commit is over. In case of multiple
+    KILLs happened on commit codepath, the last one will be effective.
+  */
+  if (victim_thd->wsrep_abort_by_kill ||
+      trx_state == trans::s_preparing ||
+      trx_state == trans::s_committing ||
+      trx_state == trans::s_ordered_commit)
   {
-    ret= victim_thd->wsrep_cs().total_order_bf_abort(bf_seqno);
+    victim_thd->wsrep_abort_by_kill= kill_signal;
+    DBUG_RETURN(0);
   }
-  else
+  /*
+    Mark killed victim_thd with kill_signal so that awake_no_mutex does
+    not dive into storage engine. We use ha_abort_transaction()
+    to do the storage engine part for wsrep THDs.
+  */
+  DEBUG_SYNC(thd, "wsrep_kill_before_awake_no_mutex");
+  victim_thd->wsrep_abort_by_kill= kill_signal;
+  victim_thd->awake_no_mutex(kill_signal);
+  /* ha_abort_transaction() releases tmp->LOCK_thd_kill, so tmp
+     is not safe to access anymore. */
+  ha_abort_transaction(thd, victim_thd, 1);
+  DBUG_RETURN(0);
+}
+
+void wsrep_backup_kill_for_commit(THD *thd)
+{
+  DBUG_ASSERT(WSREP(thd));
+  mysql_mutex_assert_owner(&thd->LOCK_thd_kill);
+  DBUG_ASSERT(thd->killed != NOT_KILLED);
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  /* If the transaction will roll back, keep the killed state.
+     For must replay, the replay will happen in different THD context
+     which is high priority and cannot be killed. The owning thread will
+     pick the killed state in after statement processing. */
+  if (thd->wsrep_trx().state() != wsrep::transaction::s_cert_failed &&
+      thd->wsrep_trx().state() != wsrep::transaction::s_must_abort &&
+      thd->wsrep_trx().state() != wsrep::transaction::s_aborting &&
+      thd->wsrep_trx().state() != wsrep::transaction::s_must_replay)
   {
-    DBUG_ASSERT(WSREP(victim_thd) ? victim_thd->wsrep_trx().active() : 1);
-    ret= victim_thd->wsrep_cs().bf_abort(bf_seqno);
+    thd->wsrep_abort_by_kill= thd->killed;
+    thd->wsrep_abort_by_kill_err= thd->killed_err;
+    thd->killed= NOT_KILLED;
+    thd->killed_err= 0;
   }
-  if (ret)
-  {
-    wsrep_bf_aborts_counter++;
-  }
-  return ret;
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+}
+
+void wsrep_restore_kill_after_commit(THD *thd)
+{
+  DBUG_ASSERT(WSREP(thd));
+  mysql_mutex_assert_owner(&thd->LOCK_thd_kill);
+  thd->killed= thd->wsrep_abort_by_kill;
+  thd->killed_err= thd->wsrep_abort_by_kill_err;
+  thd->wsrep_abort_by_kill= NOT_KILLED;
+  thd->wsrep_abort_by_kill_err= 0;
 }
 
 int wsrep_create_threadvars()
diff --git a/sql/wsrep_thd.h b/sql/wsrep_thd.h
index 0ce612d6097..f3790887bf5 100644
--- a/sql/wsrep_thd.h
+++ b/sql/wsrep_thd.h
@@ -88,10 +88,39 @@ bool wsrep_create_appliers(long threads, bool mutex_protected=false);
 void wsrep_create_rollbacker();
 
 bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd);
-int  wsrep_abort_thd(THD *bf_thd,
+/*
+  Abort transaction for victim_thd. This function is called from
+  MDL BF abort codepath.
+*/
+void wsrep_abort_thd(THD *bf_thd,
                      THD *victim_thd,
                      my_bool signal) __attribute__((nonnull(1,2)));
 
+/**
+  Kill wsrep connection with kill_signal. Object thd is not
+  guaranteed to exist anymore when this function returns.
+
+  Asserts that the caller holds victim_thd->LOCK_thd_kill,
+  victim_thd->LOCK_thd_data.
+
+  @param thd THD object for connection that executes the KILL.
+  @param victim_thd THD object for connection to be killed.
+  @param kill_signal Kill signal.
+
+  @return Zero if the kill was successful, otherwise non-zero error code.
+ */
+uint wsrep_kill_thd(THD *thd, THD *victim_thd, killed_state kill_signal);
+
+/*
+  Backup kill status for commit.
+ */
+void wsrep_backup_kill_for_commit(THD *);
+
+/*
+  Restore KILL status after commit.
+ */
+void wsrep_restore_kill_after_commit(THD *);
+
 /*
   Helper methods to deal with thread local storage.
   The purpose of these methods is to hide the details of thread
diff --git a/sql/wsrep_trans_observer.h b/sql/wsrep_trans_observer.h
index 812452f451a..6b656f84c78 100644
--- a/sql/wsrep_trans_observer.h
+++ b/sql/wsrep_trans_observer.h
@@ -256,6 +256,11 @@ static inline int wsrep_before_prepare(THD* thd, bool all)
                    thd->wsrep_trx().ws_meta().gtid(),
                    wsrep_gtid_server.gtid());
   }
+
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  if (thd->killed) wsrep_backup_kill_for_commit(thd);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
+
   DBUG_RETURN(ret);
 }
 
@@ -325,6 +330,11 @@ static inline int wsrep_before_commit(THD* thd, bool all)
                    wsrep_gtid_server.gtid());
     wsrep_register_for_group_commit(thd);
   }
+
+  mysql_mutex_lock(&thd->LOCK_thd_kill);
+  if (thd->killed) wsrep_backup_kill_for_commit(thd);
+  mysql_mutex_unlock(&thd->LOCK_thd_kill);
+
   DBUG_RETURN(ret);
 }
 
@@ -343,7 +353,8 @@ static inline int wsrep_before_commit(THD* thd, bool all)
 static inline int wsrep_ordered_commit(THD* thd, bool all)
 {
   DBUG_ENTER("wsrep_ordered_commit");
-  WSREP_DEBUG("wsrep_ordered_commit: %d", wsrep_is_real(thd, all));
+  WSREP_DEBUG("wsrep_ordered_commit: %d %lld", wsrep_is_real(thd, all),
+              (long long) wsrep_thd_trx_seqno(thd));
   DBUG_ASSERT(wsrep_run_commit_hook(thd, all));
   DBUG_RETURN(thd->wsrep_cs().ordered_commit());
 }
@@ -451,10 +462,18 @@ int wsrep_after_statement(THD* thd)
               wsrep::to_c_string(thd->wsrep_cs().state()),
               wsrep::to_c_string(thd->wsrep_cs().mode()),
               wsrep::to_c_string(thd->wsrep_cs().transaction().state()));
-  DBUG_RETURN((thd->wsrep_cs().state() != wsrep::client_state::s_none &&
+  int ret= ((thd->wsrep_cs().state() != wsrep::client_state::s_none &&
                thd->wsrep_cs().mode() == Wsrep_client_state::m_local) &&
               !thd->internal_transaction() ?
               thd->wsrep_cs().after_statement() : 0);
+
+  if (wsrep_is_active(thd))
+  {
+    mysql_mutex_lock(&thd->LOCK_thd_kill);
+    wsrep_restore_kill_after_commit(thd);
+    mysql_mutex_unlock(&thd->LOCK_thd_kill);
+  }
+  DBUG_RETURN(ret);
 }
 
 static inline void wsrep_after_apply(THD* thd)
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index e4b352a05aa..49573a13787 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -447,7 +447,9 @@ static bool fil_node_open_file(fil_node_t *node)
     }
   }
 
-  return fil_node_open_file_low(node);
+  /* The node can be opened beween releasing and acquiring fil_system.mutex
+  in the above code */
+  return node->is_open() || fil_node_open_file_low(node);
 }
 
 /** Close the file handle. */
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 8a67192c5da..bd81e46d1e4 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -945,7 +945,8 @@ static SHOW_VAR innodb_status_variables[]= {
   {"buffer_pool_read_ahead", &buf_pool.stat.n_ra_pages_read, SHOW_SIZE_T},
   {"buffer_pool_read_ahead_evicted",
    &buf_pool.stat.n_ra_pages_evicted, SHOW_SIZE_T},
-  {"buffer_pool_read_requests", &buf_pool.stat.n_page_gets, SHOW_SIZE_T},
+  {"buffer_pool_read_requests",
+   &export_vars.innodb_buffer_pool_read_requests, SHOW_SIZE_T},
   {"buffer_pool_reads", &buf_pool.stat.n_pages_read, SHOW_SIZE_T},
   {"buffer_pool_wait_free", &buf_pool.stat.LRU_waits, SHOW_SIZE_T},
   {"buffer_pool_write_requests", &buf_pool.flush_list_requests, SHOW_SIZE_T},
@@ -1935,8 +1936,9 @@ static void innodb_disable_internal_writes(bool disable)
     sst_enable_innodb_writes();
 }
 
-static void wsrep_abort_transaction(handlerton*, THD *, THD *, my_bool);
-static int innobase_wsrep_set_checkpoint(handlerton* hton, const XID* xid);
+static void wsrep_abort_transaction(handlerton *, THD *, THD *, my_bool)
+    __attribute__((nonnull));
+static int innobase_wsrep_set_checkpoint(handlerton *hton, const XID *xid);
 static int innobase_wsrep_get_checkpoint(handlerton* hton, XID* xid);
 #endif /* WITH_WSREP */
 
@@ -18654,36 +18656,45 @@ void lock_wait_wsrep_kill(trx_t *bf_trx, ulong thd_id, trx_id_t trx_id)
                       wsrep_thd_client_mode_str(vthd),
                       wsrep_thd_transaction_state_str(vthd),
                       wsrep_thd_query(vthd));
-          /* Mark transaction as a victim for Galera abort */
-          vtrx->lock.set_wsrep_victim();
-          if (!wsrep_thd_set_wsrep_aborter(bf_thd, vthd))
-            aborting= true;
-          else
-            WSREP_DEBUG("kill transaction skipped due to wsrep_aborter set");
+          aborting= true;
         }
       }
       mysql_mutex_unlock(&lock_sys.wait_mutex);
       vtrx->mutex_unlock();
     }
-    wsrep_thd_UNLOCK(vthd);
-    if (aborting)
+
+    DEBUG_SYNC(bf_thd, "before_wsrep_thd_abort");
+    if (aborting && wsrep_thd_bf_abort(bf_thd, vthd, true))
     {
+      /* Need to grab mutexes again to ensure that the trx is still in
+         right state. */
+      lock_sys.wr_lock(SRW_LOCK_CALL);
+      mysql_mutex_lock(&lock_sys.wait_mutex);
+      vtrx->mutex_lock();
+
       /* 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");
-      if (!wsrep_thd_bf_abort(bf_thd, vthd, true))
+      if (vtrx->id == trx_id)
       {
-        wsrep_thd_LOCK(vthd);
-        wsrep_thd_set_wsrep_aborter(NULL, vthd);
-        wsrep_thd_UNLOCK(vthd);
-
-        WSREP_DEBUG("wsrep_thd_bf_abort has failed, victim %lu will survive",
-                     thd_get_thread_id(vthd));
+        switch (vtrx->state) {
+        default:
+          break;
+        case TRX_STATE_ACTIVE:
+        case TRX_STATE_PREPARED:
+          lock_sys.cancel_lock_wait_for_wsrep_bf_abort(vtrx);
+        }
       }
+      lock_sys.wr_unlock();
+      mysql_mutex_unlock(&lock_sys.wait_mutex);
+      vtrx->mutex_unlock();
     }
+    else
+    {
+      WSREP_DEBUG("wsrep_thd_bf_abort has failed, victim %lu will survive",
+                  thd_get_thread_id(vthd));
+    }
+    wsrep_thd_UNLOCK(vthd);
     wsrep_thd_kill_UNLOCK(vthd);
   }
 }
@@ -18691,68 +18702,50 @@ void lock_wait_wsrep_kill(trx_t *bf_trx, ulong thd_id, trx_id_t trx_id)
 /** This function forces the victim transaction to abort. Aborting the
   transaction does NOT end it, it still has to be rolled back.
 
+  The caller must lock LOCK_thd_kill and LOCK_thd_data.
+
   @param bf_thd       brute force THD asking for the abort
   @param victim_thd   victim THD to be aborted
-
-  @return 0 victim was aborted
-  @return -1 victim thread was aborted (no transaction)
 */
-static
-void
-wsrep_abort_transaction(
-	handlerton*,
-	THD *bf_thd,
-	THD *victim_thd,
-	my_bool signal)
+static void wsrep_abort_transaction(handlerton *, THD *bf_thd, THD *victim_thd,
+                                    my_bool signal)
 {
-	DBUG_ENTER("wsrep_abort_transaction");
-	ut_ad(bf_thd);
-	ut_ad(victim_thd);
+  DBUG_ENTER("wsrep_abort_transaction");
+  ut_ad(bf_thd);
+  ut_ad(victim_thd);
 
-	wsrep_thd_kill_LOCK(victim_thd);
-	wsrep_thd_LOCK(victim_thd);
-	trx_t* victim_trx= thd_to_trx(victim_thd);
-	wsrep_thd_UNLOCK(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));
+  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));
 
-	if (victim_trx) {
-		victim_trx->lock.set_wsrep_victim();
+  if (!victim_trx)
+  {
+    WSREP_DEBUG("abort transaction: victim did not exist");
+    DBUG_VOID_RETURN;
+  }
 
-		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)));
-					 };);
-			wsrep_thd_bf_abort(bf_thd, victim_thd, signal);
-		}
-	} 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_bf_abort(bf_thd, victim_thd, signal);
-	}
+  lock_sys.wr_lock(SRW_LOCK_CALL);
+  mysql_mutex_lock(&lock_sys.wait_mutex);
+  victim_trx->mutex_lock();
 
-	wsrep_thd_kill_UNLOCK(victim_thd);
-	DBUG_VOID_RETURN;
+  switch (victim_trx->state) {
+  default:
+    break;
+  case TRX_STATE_ACTIVE:
+  case TRX_STATE_PREPARED:
+    /* Cancel lock wait if the victim is waiting for a lock in InnoDB.
+       The transaction which is blocked somewhere else (e.g. waiting
+       for next command or MDL) has been interrupted by THD::awake_no_mutex()
+       on server level before calling this function. */
+    lock_sys.cancel_lock_wait_for_wsrep_bf_abort(victim_trx);
+  }
+  lock_sys.wr_unlock();
+  mysql_mutex_unlock(&lock_sys.wait_mutex);
+  victim_trx->mutex_unlock();
+
+  DBUG_VOID_RETURN;
 }
 
 static
diff --git a/storage/innobase/include/lock0lock.h b/storage/innobase/include/lock0lock.h
index 16acd031177..e8299bb1189 100644
--- a/storage/innobase/include/lock0lock.h
+++ b/storage/innobase/include/lock0lock.h
@@ -955,6 +955,10 @@ public:
 
   /** Cancel possible lock waiting for a transaction */
   static void cancel_lock_wait_for_trx(trx_t *trx);
+#ifdef WITH_WSREP
+  /** Cancel lock waiting for a wsrep BF abort. */
+  static void cancel_lock_wait_for_wsrep_bf_abort(trx_t *trx);
+#endif /* WITH_WSREP */
 };
 
 /** The lock system */
diff --git a/storage/innobase/include/srv0srv.h b/storage/innobase/include/srv0srv.h
index f02c10b603b..7520dae0dab 100644
--- a/storage/innobase/include/srv0srv.h
+++ b/storage/innobase/include/srv0srv.h
@@ -652,6 +652,8 @@ struct export_var_t{
 #ifdef UNIV_DEBUG
 	ulint innodb_buffer_pool_pages_latched;	/*!< Latched pages */
 #endif /* UNIV_DEBUG */
+	/** buf_pool.stat.n_page_gets (a sharded counter) */
+	ulint innodb_buffer_pool_read_requests;
 	ulint innodb_checkpoint_age;
 	ulint innodb_checkpoint_max_age;
 	ulint innodb_data_pending_reads;	/*!< Pending reads */
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 3c7c3d348af..08547f169f3 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -5732,13 +5732,14 @@ static void lock_release_autoinc_locks(trx_t *trx)
 }
 
 /** Cancel a waiting lock request and release possibly waiting transactions */
-template <bool from_deadlock= false>
+template <bool from_deadlock= false, bool inner_trx_lock= true>
 void lock_cancel_waiting_and_release(lock_t *lock)
 {
   lock_sys.assert_locked(*lock);
   mysql_mutex_assert_owner(&lock_sys.wait_mutex);
   trx_t *trx= lock->trx;
-  trx->mutex_lock();
+  if (inner_trx_lock)
+    trx->mutex_lock();
   ut_d(const auto trx_state= trx->state);
   ut_ad(trx_state == TRX_STATE_COMMITTED_IN_MEMORY ||
         trx_state == TRX_STATE_ACTIVE);
@@ -5762,7 +5763,8 @@ void lock_cancel_waiting_and_release(lock_t *lock)
 
   lock_wait_end<from_deadlock>(trx);
 
-  trx->mutex_unlock();
+  if (inner_trx_lock)
+    trx->mutex_unlock();
 }
 
 void lock_sys_t::cancel_lock_wait_for_trx(trx_t *trx)
@@ -5779,6 +5781,19 @@ void lock_sys_t::cancel_lock_wait_for_trx(trx_t *trx)
   mysql_mutex_unlock(&lock_sys.wait_mutex);
 }
 
+#ifdef WITH_WSREP
+void lock_sys_t::cancel_lock_wait_for_wsrep_bf_abort(trx_t *trx)
+{
+  lock_sys.assert_locked();
+  mysql_mutex_assert_owner(&lock_sys.wait_mutex);
+  ut_ad(trx->mutex_is_owner());
+  ut_ad(trx->state == TRX_STATE_ACTIVE || trx->state == TRX_STATE_PREPARED);
+  trx->lock.set_wsrep_victim();
+  if (lock_t *lock= trx->lock.wait_lock)
+    lock_cancel_waiting_and_release<false, false>(lock);
+}
+#endif /* WITH_WSREP */
+
 /** Cancel a waiting lock request.
 @tparam check_victim  whether to check for DB_DEADLOCK
 @param trx            active transaction
diff --git a/storage/innobase/srv/srv0mon.cc b/storage/innobase/srv/srv0mon.cc
index 30592d1db53..4837cccaf5f 100644
--- a/storage/innobase/srv/srv0mon.cc
+++ b/storage/innobase/srv/srv0mon.cc
@@ -686,7 +686,7 @@ static monitor_info_t	innodb_counter_info[] =
 	{"trx_rseg_history_len", "transaction",
 	 "Length of the TRX_RSEG_HISTORY list",
 	 static_cast<monitor_type_t>(
-	 MONITOR_EXISTING | MONITOR_DISPLAY_CURRENT),
+	 MONITOR_EXISTING | MONITOR_DISPLAY_CURRENT | MONITOR_DEFAULT_ON),
 	 MONITOR_DEFAULT_START, MONITOR_RSEG_HISTORY_LEN},
 
 	{"trx_undo_slots_used", "transaction", "Number of undo slots used",
diff --git a/wsrep-lib b/wsrep-lib
index 4951c383577..e238c0d240c 160000
--- a/wsrep-lib
+++ b/wsrep-lib
@@ -1 +1 @@
-Subproject commit 4951c38357737d568b554402bc5b6abe88a38fe1
+Subproject commit e238c0d240c2557229b0523a4a032f3cf8b41639