MDEV-25616 XA PREPARE event group is not binlogged when..

the only query of the XA transaction is on a non-transactional table
errors out:

XA BEGIN 'x';
--error ER_DUP_ENTRY
INSERT INTO t1 VALUES (1),(1);
XA END 'x';
XA PREPARE 'x';

The binlogging pattern is correctly started as expected with
the errored-out Query or its ROW format events, but there is
no empty XA_prepare_log_event group.
The following

XA COMMIT 'x';

therefore should not be logged either, but it does.

The bug is fixed with proper maintaining of a read-write binlog hton
property and use it to enforce correct binlogging decisions.
Specifically in the bug description case XA COMMIT won't be binlogged
in both when given in the same connection and externally after disconnect.
The same continue to apply to an empty XA that do not change any data in all
transactional engines involved.

mysql-test/suite/binlog/r/binlog_empty_xa_prepared.result

@@ -108,3 +108,111 @@ master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; DROP SEQUENCE `s` /* generated by server */
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+connect con1,localhost,root,,;
+XA START '1';
+INSERT INTO t1 VALUES (2),(1);
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+SELECT * FROM t1 WHERE a = 2;
+a
+XA END '1';
+XA PREPARE '1';
+disconnect con1;
+connection default;
+XA RECOVER;
+formatID	gtrid_length	bqual_length	data
+1	1	0	1
+XA COMMIT '1';
+ERROR XA100: XA_RBROLLBACK: Transaction branch was rolled back
+Must be no XA PREPARE group nor XA completion one:
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Annotate_rows	#	#	INSERT INTO t1 VALUES (1)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE SEQUENCE s ENGINE=InnoDB
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Annotate_rows	#	#	SELECT NEXT VALUE FOR s
+master-bin.000001	#	Table_map	#	#	table_id: # (test.s)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	XA START X'32',X'',1 GTID #-#-#
+master-bin.000001	#	Query	#	#	XA END X'32',X'',1
+master-bin.000001	#	XA_prepare	#	#	XA PREPARE X'32',X'',1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	XA ROLLBACK X'32',X'',1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP SEQUENCE `s` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Annotate_rows	#	#	INSERT INTO t1 VALUES (1)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+DROP TABLE t1;
+connect con2,localhost,root,,;
+CREATE TABLE tm (a INT PRIMARY KEY) ENGINE=MyISAM;
+XA START '1';
+INSERT INTO tm VALUES (1),(1);
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+SELECT * FROM tm WHERE a = 2;
+a
+XA END '1';
+XA PREPARE '1';
+disconnect con2;
+connection default;
+XA RECOVER;
+formatID	gtrid_length	bqual_length	data
+1	1	0	1
+XA ROLLBACK '1';
+ERROR XA100: XA_RBROLLBACK: Transaction branch was rolled back
+Must be no XA PREPARE group nor XA completion one:
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Annotate_rows	#	#	INSERT INTO t1 VALUES (1)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE SEQUENCE s ENGINE=InnoDB
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Annotate_rows	#	#	SELECT NEXT VALUE FOR s
+master-bin.000001	#	Table_map	#	#	table_id: # (test.s)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	XA START X'32',X'',1 GTID #-#-#
+master-bin.000001	#	Query	#	#	XA END X'32',X'',1
+master-bin.000001	#	XA_prepare	#	#	XA PREPARE X'32',X'',1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	XA ROLLBACK X'32',X'',1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP SEQUENCE `s` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Annotate_rows	#	#	INSERT INTO t1 VALUES (1)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE tm (a INT PRIMARY KEY) ENGINE=MyISAM
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO tm VALUES (1),(1)
+master-bin.000001	#	Query	#	#	COMMIT
+DROP TABLE tm;

mysql-test/suite/binlog/t/binlog_empty_xa_prepared.test

@@ -80,3 +80,55 @@ DROP TABLE t1;
 
 --echo # Proof of  correct logging incl empty XA-PREPARE
 --source include/show_binlog_events.inc
+
+
+# MDEV-25616 Binlog event for XA COMMIT is generated without matching XA START
+
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+
+--source include/count_sessions.inc
+--connect(con1,localhost,root,,)
+
+XA START '1';
+  --error ER_DUP_ENTRY
+  INSERT INTO t1 VALUES (2),(1);
+  SELECT * FROM t1 WHERE a = 2;
+XA END '1';
+XA PREPARE '1';
+
+--disconnect con1
+
+--connection default
+--source include/wait_until_count_sessions.inc
+XA RECOVER;
+
+--error ER_XA_RBROLLBACK
+XA COMMIT '1';
+--echo Must be no XA PREPARE group nor XA completion one:
+--source include/show_binlog_events.inc
+DROP TABLE t1;
+
+--source include/count_sessions.inc
+
+--connect(con2,localhost,root,,)
+CREATE TABLE tm (a INT PRIMARY KEY) ENGINE=MyISAM;
+XA START '1';
+  --error ER_DUP_ENTRY
+  INSERT INTO tm VALUES (1),(1);
+  SELECT * FROM tm WHERE a = 2;
+XA END '1';
+XA PREPARE '1';
+
+--disconnect con2
+
+--connection default
+--source include/wait_until_count_sessions.inc
+XA RECOVER;
+
+--error ER_XA_RBROLLBACK
+XA ROLLBACK '1';
+--echo Must be no XA PREPARE group nor XA completion one:
+--source include/show_binlog_events.inc
+DROP TABLE tm;
+

mysql-test/suite/rpl/include/rpl_xa_empty_transaction.inc

+#
+#   Helper file to run each empty-due-to-err XA transaction test case both with
+# and without detaching from the connection when the transaction is prepared.
+#
+
+--let $use_disconnect=0
+--source rpl_xa_empty_transaction_test_case.inc
+
+--let $use_disconnect=1
+--source rpl_xa_empty_transaction_test_case.inc

mysql-test/suite/rpl/include/rpl_xa_empty_transaction_test_case.inc

+#
+#   Helper script to create an XA transaction and validate it was not
+# binlogged
+#
+# Parameters
+#   $xa_completion_action     : The action to end the XA transaction, either
+#                               COMMIT or ROLLBACK
+#   $trx_statements           : A comma separated list specifying how to build
+#                               the statements of the transaction. Each item in
+#                               the list is either T (for transactional) or N
+#                               (for non-transactional). An empty list will not
+#                               add any statements to the transaction.
+#   $use_disconnect           : When TRUE, disconnect after preparing the XA
+#                               transaction to test the detach/rollback case
+#
+
+#
+# Setup
+--let $generic_assert_text= should not binlog XA transaction
+
+--connection server_1
+--let server_1_datadir=`select @@datadir`
+
+--connection server_2
+--let server_2_datadir=`select @@datadir`
+
+--connection server_3
+--let server_3_datadir=`select @@datadir`
+
+--let assert_file=$MYSQLTEST_VARDIR/tmp/binlog_decoded.out
+
+--connection server_1
+CREATE TABLE tm (a INT PRIMARY KEY) engine=myisam;
+CREATE TABLE ti (a INT PRIMARY KEY) engine=innodb;
+--source include/save_master_gtid.inc
+
+--connection server_3
+--source include/sync_with_master_gtid.inc
+
+--connection server_1
+
+if ($use_disconnect)
+{
+    --source include/count_sessions.inc
+    --connect(con1,localhost,root,,)
+}
+
+XA START 'x';
+--let $_stmt_items= $trx_statements
+--let $_ctr= 1
+while($_stmt_items)
+{
+  --let $_cur_stmt= `SELECT SUBSTRING_INDEX('$_stmt_items', ',', 1)`
+  --let $_stmt_items= `SELECT LTRIM(SUBSTRING('$_stmt_items', LENGTH('$_cur_stmt') + 2))`
+
+    if (`SELECT strcmp("$_cur_stmt","T") = 0`)
+    {
+        --let $target_table= ti
+    }
+
+    if (`SELECT strcmp("$_cur_stmt","N") = 0`)
+    {
+        --let $target_table= tm
+    }
+
+    --error ER_DUP_ENTRY
+    --eval INSERT INTO $target_table VALUES ($_ctr),($_ctr);
+    inc $_ctr;
+
+}
+XA END 'x';
+XA PREPARE 'x';
+
+if ($use_disconnect)
+{
+    --disconnect con1
+    --connection server_1
+    --source include/wait_until_count_sessions.inc
+    XA RECOVER;
+
+    --error ER_XA_RBROLLBACK
+    --eval XA $xa_completion_action 'x';
+}
+if (!$use_disconnect)
+{
+    --eval XA $xa_completion_action 'x';
+}
+
+--source include/save_master_gtid.inc
+
+--let binlog_filename= query_get_value(SHOW MASTER STATUS, File, 1)
+FLUSH LOGS;
+
+--echo # MYSQL_BINLOG server_1_datadir/binlog_filename --result-file=assert_file
+--exec $MYSQL_BINLOG $server_1_datadir/$binlog_filename --result-file=$assert_file
+
+--let assert_text= server_1 $generic_assert_text
+--let assert_count= 0
+--let assert_select= XA START|XA END|XA PREPARE|XA COMMIT|XA ROLLBACK
+--source include/assert_grep.inc
+
+--connection server_2
+--source include/sync_with_master_gtid.inc
+--let binlog_filename= query_get_value(SHOW MASTER STATUS, File, 1)
+FLUSH LOGS;
+
+--echo # MYSQL_BINLOG server_2_datadir/binlog_filename --result-file=assert_file
+--exec $MYSQL_BINLOG $server_2_datadir/$binlog_filename --result-file=$assert_file
+
+--let assert_text= server_2 $generic_assert_text
+--source include/assert_grep.inc
+
+--connection server_3
+--source include/sync_with_master_gtid.inc
+--let binlog_filename= query_get_value(SHOW MASTER STATUS, File, 1)
+FLUSH LOGS;
+
+--echo # MYSQL_BINLOG server_3_datadir/binlog_filename --result-file=assert_file
+--exec $MYSQL_BINLOG $server_3_datadir/$binlog_filename --result-file=$assert_file
+
+--let assert_text= server_3 $generic_assert_text
+--source include/assert_grep.inc
+
+#
+# Cleanup
+--connection server_1
+DROP TABLE ti,tm;
+--source include/save_master_gtid.inc
+
+--connection server_3
+--source include/sync_with_master_gtid.inc

mysql-test/suite/rpl/t/rpl_xa_empty_transaction.cnf

+!include include/default_mysqld.cnf
+
+[mysqld.1]
+log-slave-updates
+innodb
+
+[mysqld.2]
+log-slave-updates
+innodb
+
+[mysqld.3]
+log-slave-updates
+innodb
+
+[ENV]
+SERVER_MYPORT_1= @mysqld.1.port
+SERVER_MYPORT_2= @mysqld.2.port
+SERVER_MYPORT_3= @mysqld.3.port

mysql-test/suite/rpl/t/rpl_xa_empty_transaction.test

+#
+# Purpose:
+#   This test ensures consistency in binlogging behavior for XA transactions
+# that have all statements error and rollback, effectively leaving an "empty"
+# transaction. In such cases, an empty XA transaction should be binlogged. The
+# bug reported by MDEV-25616 revealed that an "empty" XA transaction would
+# binlog an XA ROLLBACK or XA COMMIT event without a preceding setup, i.e.
+# XA START through XA PREPARE. The bug presented differently for XA
+# transactions consisting of transactional and non-transactional statements.
+# Therefore, this test validates that an entire XA transaction is binlogged
+# for different combinations of transactional or non-transactional statements.
+# Note that the behavior changes when binlogging empty XA transactions
+# depending on the binlog_row_format variables. That is, when the content of
+# the transaction consists of errored transactional statements, in row format,
+# an empty XA transaction will be binlogged; however, in mixed and statement
+# formats, nothing will be written into the binary log.
+#
+# Methodology:
+#   Create XA transactions with various combinations of erroring transactional
+# or non-transactional statements. The binary log is examined to ensure all
+# XA components are written. Chain replication is used, i.e.
+# (primary->replica->replica), to ensure replica binlogging is consistent with
+# manual execution. The transactional and non-transactional tables use InnoDB
+# and MyISAM, respectively.
+#
+# Parameters
+#   $expect_transactional_xa_binlog : Boolean indicating whether or not an
+#       errored transactional statement should result in XA statements written
+#       into the binary log.
+#
+# References:
+#   MDEV-25616: Binlog event for XA COMMIT is generated without matching
+#               XA START, replication aborts
+#
+--source include/have_log_bin.inc
+
+--let $rpl_server_count= 3
+--let $rpl_topology= 1->2->3
+--source include/rpl_init.inc
+
+--connection server_1
+-- source include/have_innodb.inc
+--connection server_2
+-- source include/have_innodb.inc
+--connection server_3
+-- source include/have_innodb.inc
+--connection server_1
+
+--echo #
+--echo # Test Case 1: An XA transaction without any statements should not be
+--echo # binlogged
+--let $trx_statements=
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+
+--echo #
+--echo # Test Case 2: An XA transaction consisting of a successfully rolled back
+--echo # statement should not be binlogged
+--let $trx_statements= T
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+
+--echo #
+--echo # Test Case 3: An XA transaction with a statement that cannot be rolled
+--echo # back should be binlogged
+
+# TODO: remove work-around MDEV-24654 when fixed.
+--connection server_1
+set @sav_binlog_format = @@binlog_format;
+set @@binlog_format = row;
+set @@global.binlog_format = row;
+--let $trx_statements= N
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+--connection server_1
+set @@binlog_format = @sav_binlog_format;
+set @@global.binlog_format = @sav_binlog_format;
+
+
+--echo #
+--echo # Test Case 4: An XA transaction with multiple statements that can all
+--echo # be rolled back should not be binlogged
+--let $trx_statements= T,T
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+
+--echo #
+--echo # Test Case 5: A mixed XA transaction consisting of one statement that
+--echo # can successfully be rolled back (first statement), and another that
+--echo # can not (second statement) should be binlogged
+
+--connection server_1
+set @sav_binlog_format = @@binlog_format;
+set @@binlog_format = row;
+set @@global.binlog_format = row;
+--let $trx_statements= T,N
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+--connection server_1
+set @@binlog_format = @sav_binlog_format;
+set @@global.binlog_format = @sav_binlog_format;
+
+
+--echo #
+--echo # Test Case 6: A mixed XA transaction consisting of one statement that
+--echo # cannot successfully be rolled back (first statement), and another that
+--echo # can (second statement) should be binlogged
+
+--connection server_1
+set @sav_binlog_format = @@binlog_format;
+set @@binlog_format = row;
+set @@global.binlog_format = row;
+--let $trx_statements= N,T
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+--connection server_1
+set @@binlog_format = @sav_binlog_format;
+set @@global.binlog_format = @sav_binlog_format;
+
+--echo #
+--echo # Test Case 7: An XA transaction consisting of two failed
+--echo # non-transactional statements should be binlogged
+
+--connection server_1
+set @sav_binlog_format = @@binlog_format;
+set @@binlog_format = row;
+set @@global.binlog_format = row;
+--let $trx_statements= N,N
+
+--let $xa_completion_action= COMMIT
+--source include/rpl_xa_empty_transaction.inc
+
+--let $xa_completion_action= ROLLBACK
+--source include/rpl_xa_empty_transaction.inc
+
+--connection server_1
+set @@binlog_format = @sav_binlog_format;
+set @@global.binlog_format = @sav_binlog_format;
+
+#
+# Cleanup
+--connection server_1
+--source include/rpl_end.inc
+
+--echo # End of rpl_xa_empty_transaction.test

sql/log.cc

@@ -2003,25 +2003,56 @@ static int binlog_xa_recover_dummy(handlerton *hton __attribute__((unused)),
 
 static int binlog_commit_by_xid(handlerton *hton, XID *xid)
 {
+  int rc= 0;
   THD *thd= current_thd;
 
+  /* the asserted state can't be reachable with xa commit */
+  DBUG_ASSERT(!thd->get_stmt_da()->is_error() ||
+              thd->get_stmt_da()->sql_errno() != ER_XA_RBROLLBACK);
+  /*
+    This is a recovered user xa transaction commit.
+    Create a "temporary" binlog transaction to write the commit record
+    into binlog.
+  */
+  THD_TRANS trans;
+  trans.ha_list= NULL;
+
+  thd->ha_data[hton->slot].ha_info[1].register_ha(&trans, hton);
+  thd->ha_data[binlog_hton->slot].ha_info[1].set_trx_read_write();
   (void) thd->binlog_setup_trx_data();
 
   DBUG_ASSERT(thd->lex->sql_command == SQLCOM_XA_COMMIT);
 
-  return binlog_commit(hton, thd, TRUE);
+  rc= binlog_commit(hton, thd, TRUE);
+  thd->ha_data[binlog_hton->slot].ha_info[1].reset();
+
+  return rc;
 }
 
 
 static int binlog_rollback_by_xid(handlerton *hton, XID *xid)
 {
+  int rc= 0;
   THD *thd= current_thd;
 
+  if (thd->get_stmt_da()->is_error() &&
+      thd->get_stmt_da()->sql_errno() == ER_XA_RBROLLBACK)
+    return rc;
+
+  THD_TRANS trans;
+  trans.ha_list= NULL;
+
+  thd->ha_data[hton->slot].ha_info[1].register_ha(&trans, hton);
+  thd->ha_data[hton->slot].ha_info[1].set_trx_read_write();
   (void) thd->binlog_setup_trx_data();
 
   DBUG_ASSERT(thd->lex->sql_command == SQLCOM_XA_ROLLBACK ||
               (thd->transaction->xid_state.get_state_code() == XA_ROLLBACK_ONLY));
-  return binlog_rollback(hton, thd, TRUE);
+
+  rc= binlog_rollback(hton, thd, TRUE);
+  thd->ha_data[hton->slot].ha_info[1].reset();
+
+  return rc;
 }
 
 
@@ -2159,10 +2190,16 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
   }
 
   if (cache_mngr->trx_cache.empty() &&
-      thd->transaction->xid_state.get_state_code() != XA_PREPARED)
+      (thd->transaction->xid_state.get_state_code() != XA_PREPARED ||
+       !(thd->ha_data[binlog_hton->slot].ha_info[1].is_started() &&
+         thd->ha_data[binlog_hton->slot].ha_info[1].is_trx_read_write())))
   {
     /*
-      we're here because cache_log was flushed in MYSQL_BIN_LOG::log_xid()
+      This is an empty transaction commit (both the regular and xa),
+      or such transaction xa-prepare or
+      either one's statement having no effect on the transactional cache
+      as any prior to it.
+      The empty xa-prepare sinks in only when binlog is read-only.
     */
     cache_mngr->reset(false, true);
     THD_STAGE_INFO(thd, org_stage);
@@ -2247,10 +2284,12 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   }
 
   if (!cache_mngr->trx_cache.has_incident() && cache_mngr->trx_cache.empty() &&
-      thd->transaction->xid_state.get_state_code() != XA_PREPARED)
+      (thd->transaction->xid_state.get_state_code() != XA_PREPARED ||
+       !(thd->ha_data[binlog_hton->slot].ha_info[1].is_started() &&
+         thd->ha_data[binlog_hton->slot].ha_info[1].is_trx_read_write())))
   {
     /*
-      we're here because cache_log was flushed in MYSQL_BIN_LOG::log_xid()
+      The same comments apply as in the binlog commit method's branch.
     */
     cache_mngr->reset(false, true);
     thd->reset_binlog_for_next_statement();
@@ -10290,6 +10329,7 @@ int TC_LOG_BINLOG::unlog_xa_prepare(THD *thd, bool all)
       /* an empty XA-prepare event group is logged */
       rc= write_empty_xa_prepare(thd, cache_mngr); // normally gains need_unlog
       trans_register_ha(thd, true, binlog_hton, 0); // do it for future commmit
+      thd->ha_data[binlog_hton->slot].ha_info[1].set_trx_read_write();
     }
     if (rw_count == 0 || !cache_mngr->need_unlog)
       return rc;