Backport Bug#37148 to 5.1

1 files changed, 24 insertions, 17 deletions

diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index 684655d1c3b..666622dbac4 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -58,10 +58,14 @@ injector::transaction::~transaction()
   my_free(the_memory, MYF(0));
 }
 
+/**
+   @retval 0 transaction committed
+   @retval 1 transaction rolled back
+ */
 int injector::transaction::commit()
 {
    DBUG_ENTER("injector::transaction::commit()");
-   m_thd->binlog_flush_pending_rows_event(true);
+   int error= m_thd->binlog_flush_pending_rows_event(true);
    /*
      Cluster replication does not preserve statement or
      transaction boundaries of the master.  Instead, a new
@@ -81,9 +85,9 @@ int injector::transaction::commit()
      is committed by committing the statement transaction
      explicitly.
    */
-   ha_autocommit_or_rollback(m_thd, 0);
-   end_trans(m_thd, COMMIT);
-   DBUG_RETURN(0);
+   error |= ha_autocommit_or_rollback(m_thd, error);
+   end_trans(m_thd, error ? ROLLBACK : COMMIT);
+   DBUG_RETURN(error);
 }
 
 int injector::transaction::use_table(server_id_type sid, table tbl)
@@ -109,16 +113,17 @@ int injector::transaction::write_row (server_id_type sid, table tbl,
 				      record_type record)
 {
    DBUG_ENTER("injector::transaction::write_row(...)");
-
-   if (int error= check_state(ROW_STATE))
+ 
+   int error= 0;
+   if (error= check_state(ROW_STATE))
      DBUG_RETURN(error);
 
    server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
-   m_thd->binlog_write_row(tbl.get_table(), tbl.is_transactional(), 
-                           cols, colcnt, record);
+   error= m_thd->binlog_write_row(tbl.get_table(), tbl.is_transactional(), 
+                                  cols, colcnt, record);
    m_thd->set_server_id(save_id);
-   DBUG_RETURN(0);
+   DBUG_RETURN(error);
 }
 
 
@@ -128,15 +133,16 @@ int injector::transaction::delete_row(server_id_type sid, table tbl,
 {
    DBUG_ENTER("injector::transaction::delete_row(...)");
 
-   if (int error= check_state(ROW_STATE))
+   int error= 0;
+   if (error= check_state(ROW_STATE))
      DBUG_RETURN(error);
 
    server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
-   m_thd->binlog_delete_row(tbl.get_table(), tbl.is_transactional(), 
-                            cols, colcnt, record);
+   error= m_thd->binlog_delete_row(tbl.get_table(), tbl.is_transactional(), 
+                                   cols, colcnt, record);
    m_thd->set_server_id(save_id);
-   DBUG_RETURN(0);
+   DBUG_RETURN(error);
 }
 
 
@@ -146,15 +152,16 @@ int injector::transaction::update_row(server_id_type sid, table tbl,
 {
    DBUG_ENTER("injector::transaction::update_row(...)");
 
-   if (int error= check_state(ROW_STATE))
+   int error= 0;
+   if (error= check_state(ROW_STATE))
      DBUG_RETURN(error);
 
    server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
-   m_thd->binlog_update_row(tbl.get_table(), tbl.is_transactional(),
-		            cols, colcnt, before, after);
+   error= m_thd->binlog_update_row(tbl.get_table(), tbl.is_transactional(),
+                                   cols, colcnt, before, after);
    m_thd->set_server_id(save_id);
-   DBUG_RETURN(0);
+   DBUG_RETURN(error);
 }