MDEV-33509 Failed to apply write set with flags=(rollback|pa_unsafe)

Fix function `remove_fragment()` in wsrep_schema so that no error is
raised if the fragment to be removed is not found in the
wsrep_streaming_log table. This is necessary to handle the case where
streaming transaction in idle state is BF aborted. This may result in
the case where the rollbacker thread successfully removes the
transaction's fragments, followed by the applier's attempt to remove
the same fragments. Causing the node to leave the cluster after
reporting a "Failed to apply write set" error.

Signed-off-by: Julius Goryavsky <julius.goryavsky@mariadb.com>

mysql-test/suite/galera_sr/r/galera_sr_bf_abort_idle.result

@@ -0,0 +1,33 @@
+connection node_2;
+connection node_1;
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connection node_1;
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 INTEGER);
+INSERT INTO t1 VALUES (1,1),(2,1),(3,1),(4,1),(5,1),(6,1),(7,1),(8,1);
+SET SESSION wsrep_trx_fragment_size=10;
+SET SESSION wsrep_trx_fragment_unit='rows';
+START TRANSACTION;
+UPDATE t1 SET f2 = f2 + 10;
+connection node_2;
+INSERT INTO t1 VALUES (10,2);
+connection node_1a;
+connection node_1;
+INSERT INTO t1 VALUES (9,1);
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+ROLLBACK;
+DROP TABLE t1;
+connection node_1;
+CREATE TABLE t1(f1 INTEGER PRIMARY KEY, f2 INTEGER);
+INSERT INTO t1 VALUES (1,1),(2,1),(3,1),(4,1),(5,1),(6,1),(7,1),(8,1);
+SET SESSION wsrep_trx_fragment_size=5;
+SET SESSION wsrep_trx_fragment_unit='rows';
+START TRANSACTION;
+UPDATE t1 SET f2 = f2 + 10;
+connection node_2;
+INSERT INTO t1 VALUES (10,2);
+connection node_1a;
+connection node_1;
+INSERT INTO t1 VALUES (9,1);
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+ROLLBACK;
+DROP TABLE t1;

mysql-test/suite/galera_sr/t/galera_sr_bf_abort_idle.test

@@ -0,0 +1,68 @@
+#
+# Test BF abort for idle SR transactions
+#
+
+--source include/galera_cluster.inc
+
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+
+#
+# Case 1: BF abort idle SR transaction that has not yet replicated any fragments
+#
+--connection node_1
+CREATE TABLE t1 (f1 INTEGER PRIMARY KEY, f2 INTEGER);
+INSERT INTO t1 VALUES (1,1),(2,1),(3,1),(4,1),(5,1),(6,1),(7,1),(8,1);
+
+--let $bf_count = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.global_status WHERE VARIABLE_NAME = 'wsrep_local_bf_aborts'`
+
+SET SESSION wsrep_trx_fragment_size=10;
+SET SESSION wsrep_trx_fragment_unit='rows';
+START TRANSACTION;
+UPDATE t1 SET f2 = f2 + 10;
+
+--connection node_2
+INSERT INTO t1 VALUES (10,2);
+
+# Wait for SR transaction to be BF aborted
+--connection node_1a
+--let $wait_condition = SELECT VARIABLE_VALUE = $bf_count + 1 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_local_bf_aborts'
+--source include/wait_condition.inc
+
+
+--connection node_1
+--error ER_LOCK_DEADLOCK
+INSERT INTO t1 VALUES (9,1);
+ROLLBACK;
+
+DROP TABLE t1;
+
+
+#
+# Case 2: BF abort idle SR transaction that has already replicated a fragment
+#
+--connection node_1
+CREATE TABLE t1(f1 INTEGER PRIMARY KEY, f2 INTEGER);
+INSERT INTO t1 VALUES (1,1),(2,1),(3,1),(4,1),(5,1),(6,1),(7,1),(8,1);
+
+--let $bf_count = `SELECT VARIABLE_VALUE FROM INFORMATION_SCHEMA.global_status WHERE VARIABLE_NAME = 'wsrep_local_bf_aborts'`
+
+
+SET SESSION wsrep_trx_fragment_size=5;
+SET SESSION wsrep_trx_fragment_unit='rows';
+START TRANSACTION;
+UPDATE t1 SET f2 = f2 + 10;
+
+--connection node_2
+INSERT INTO t1 VALUES (10,2);
+
+# Wait for SR transaction to be BF aborted
+--connection node_1a
+--let $wait_condition = SELECT VARIABLE_VALUE = $bf_count + 1 FROM INFORMATION_SCHEMA.GLOBAL_STATUS WHERE VARIABLE_NAME = 'wsrep_local_bf_aborts'
+--source include/wait_condition.inc
+
+--connection node_1
+--error ER_LOCK_DEADLOCK
+INSERT INTO t1 VALUES (9,1);
+ROLLBACK;
+
+DROP TABLE t1;

sql/wsrep_schema.cc

@@ -1117,7 +1117,10 @@ static int remove_fragment(THD*                  thd,
                  seqno.get(),
                  error);
     }
-    ret= error;
+    else
+    {
+      ret= error;
+    }
   }
   else if (Wsrep_schema_impl::delete_row(frag_table))
   {