diff --git a/sql/xa.cc b/sql/xa.cc
index 199d4c5..931f553 100644
--- a/sql/xa.cc
+++ b/sql/xa.cc
@@ -709,24 +709,44 @@ bool Sql_cmd_xa_prepare::trans_xa_prepare(THD *thd)
     my_error(ER_XAER_RMFAIL, MYF(0), xid_state->state_name());
   else if (!xid_state->has_same_xid(m_xid))
     my_error(ER_XAER_NOTA, MYF(0));
-  else if (ha_prepare(thd))
+  else
   {
+    MDL_request mdl_request; bool mdl_lock_timeout = false;
+
+    /*
+      We acquire the COMMIT lock in 'xa prepare' so that xtrabackup can work
+      correctly --- otherwise redo logs of prepared txns might by partially
+      copied by xtrabackup, and becomes inconsistent with binlog position it
+      notes down right after the engine log copy operation, and some
+      transactions would get lost.
+      The duriation of the lock is statement so that FTWRL can proceed when XA
+      PREPARE returns, otherwise FTWRL might have to wait a long time, and it's
+      OK for xtrabackup to backup the prepared innodb txn before it is
+      committed, as long as the innodb redo log is always consistent with binlog.
+    */
+    MDL_REQUEST_INIT(&mdl_request,
+                     MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE,
+                     MDL_STATEMENT);
+    if ((mdl_lock_timeout= thd->mdl_context.acquire_lock(&mdl_request, thd->variables.lock_wait_timeout)) ||
+        ha_prepare(thd))
+    {
 #ifdef HAVE_PSI_TRANSACTION_INTERFACE
-    DBUG_ASSERT(thd->m_transaction_psi == NULL);
+      DBUG_ASSERT(thd->m_transaction_psi == NULL);
 #endif
-
-    cleanup_trans_state(thd);
-    xid_state->set_state(XID_STATE::XA_NOTR);
-    thd->get_transaction()->cleanup();
-    my_error(ER_XA_RBROLLBACK, MYF(0));
-  }
-  else
-  {
-    xid_state->set_state(XID_STATE::XA_PREPARED);
-    MYSQL_SET_TRANSACTION_XA_STATE(thd->m_transaction_psi,
-                                   (int)xid_state->get_state());
-    if (thd->rpl_thd_ctx.session_gtids_ctx().notify_after_xa_prepare(thd))
-      sql_print_warning("Failed to collect GTID to send in the response packet!");
+      if (mdl_lock_timeout) ha_rollback_trans(thd, true);
+      cleanup_trans_state(thd);
+      xid_state->set_state(XID_STATE::XA_NOTR);
+      thd->get_transaction()->cleanup();
+      my_error(ER_XA_RBROLLBACK, MYF(0));
+    }
+    else
+    {
+      xid_state->set_state(XID_STATE::XA_PREPARED);
+      MYSQL_SET_TRANSACTION_XA_STATE(thd->m_transaction_psi,
+                                     (int)xid_state->get_state());
+      if (thd->rpl_thd_ctx.session_gtids_ctx().notify_after_xa_prepare(thd))
+        sql_print_warning("Failed to collect GTID to send in the response packet!");
+    }
   }
 
   DBUG_RETURN(thd->is_error() ||