Bug#17095 circular replication for ndb

- test case
- reset server id in injector

client/mysqltest.c

@@ -2573,7 +2573,7 @@ int do_save_master_pos()
 
     if (have_ndbcluster)
     {
-      ulonglong start_epoch= 0, applied_epoch= 0,
+      ulonglong start_epoch= 0, applied_epoch= 0, handled_epoch= 0,
 	latest_epoch=0, latest_trans_epoch=0,
 	latest_handled_binlog_epoch= 0, latest_received_binlog_epoch= 0,
 	latest_applied_binlog_epoch= 0;
@@ -2676,9 +2676,9 @@ int do_save_master_pos()
 	if (!row)
 	  die("result does not contain '%s' in '%s'",
 	      binlog, query);
-	if (latest_applied_binlog_epoch > applied_epoch)
+	if (latest_handled_binlog_epoch > handled_epoch)
 	  count= 0;
-	applied_epoch= latest_applied_binlog_epoch;
+	handled_epoch= latest_handled_binlog_epoch;
 	count++;
 	if (latest_handled_binlog_epoch >= start_epoch)
           do_continue= 0;

mysql-test/r/rpl_ndb_circular.result

+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+RESET MASTER;
+CHANGE MASTER TO master_host="127.0.0.1",master_port=SLAVE_PORT,master_user="root";
+START SLAVE;
+CREATE TABLE t1 (a int key, b int) ENGINE=ndb;
+SHOW TABLES;
+Tables_in_test
+t1
+INSERT INTO t1 VALUES (1,2);
+INSERT INTO t1 VALUES (2,3);
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	2
+2	3
+show slave status;;
+Slave_IO_State	#
+Master_Host	127.0.0.1
+Master_User	root
+Master_Port	#
+Connect_Retry	1
+Master_Log_File	master-bin.000001
+Read_Master_Log_Pos	#
+Relay_Log_File	#
+Relay_Log_Pos	#
+Relay_Master_Log_File	master-bin.000001
+Slave_IO_Running	Yes
+Slave_SQL_Running	Yes
+Replicate_Do_DB	
+Replicate_Ignore_DB	
+Replicate_Do_Table	
+Replicate_Ignore_Table	
+Replicate_Wild_Do_Table	
+Replicate_Wild_Ignore_Table	
+Last_Errno	0
+Last_Error	
+Skip_Counter	0
+Exec_Master_Log_Pos	#
+Relay_Log_Space	#
+Until_Condition	None
+Until_Log_File	
+Until_Log_Pos	0
+Master_SSL_Allowed	No
+Master_SSL_CA_File	
+Master_SSL_CA_Path	
+Master_SSL_Cert	
+Master_SSL_Cipher	
+Master_SSL_Key	
+Seconds_Behind_Master	#
+Master_SSL_Verify_Server_Cert	No
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	2
+2	3
+show slave status;;
+Slave_IO_State	#
+Master_Host	127.0.0.1
+Master_User	root
+Master_Port	#
+Connect_Retry	60
+Master_Log_File	slave-bin.000001
+Read_Master_Log_Pos	#
+Relay_Log_File	#
+Relay_Log_Pos	#
+Relay_Master_Log_File	slave-bin.000001
+Slave_IO_Running	Yes
+Slave_SQL_Running	Yes
+Replicate_Do_DB	
+Replicate_Ignore_DB	
+Replicate_Do_Table	
+Replicate_Ignore_Table	
+Replicate_Wild_Do_Table	
+Replicate_Wild_Ignore_Table	
+Last_Errno	0
+Last_Error	
+Skip_Counter	0
+Exec_Master_Log_Pos	#
+Relay_Log_Space	#
+Until_Condition	None
+Until_Log_File	
+Until_Log_Pos	0
+Master_SSL_Allowed	No
+Master_SSL_CA_File	
+Master_SSL_CA_Path	
+Master_SSL_Cert	
+Master_SSL_Cipher	
+Master_SSL_Key	
+Seconds_Behind_Master	#
+Master_SSL_Verify_Server_Cert	No
+STOP SLAVE;
+DROP TABLE t1;

mysql-test/t/rpl_ndb_circular.test

+--source include/have_ndb.inc
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+# set up circular replication
+
+--connection slave
+RESET MASTER;
+--connection master
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+--eval CHANGE MASTER TO master_host="127.0.0.1",master_port=$SLAVE_MYPORT,master_user="root"
+START SLAVE;
+
+# create the table on the "slave"
+--connection slave
+CREATE TABLE t1 (a int key, b int) ENGINE=ndb;
+#CREATE TABLE t2 (a int key, b int) ENGINE=ndb;
+--save_master_pos
+--connection master
+--sync_with_master
+# now we should have a table on the master as well
+SHOW TABLES;
+
+# insert some values on the slave and master
+--connection master
+INSERT INTO t1 VALUES (1,2);
+--connection slave
+INSERT INTO t1 VALUES (2,3);
+
+# ensure data has propagated both ways
+--connection slave
+--save_master_pos
+--connection master
+--sync_with_master
+--sync_slave_with_master
+
+# connect to slave and ensure data it there.
+--connection slave
+SELECT * FROM t1 ORDER BY a;
+#SELECT * FROM t2 ORDER BY a;
+--replace_column 1 # 4 # 7 # 8 # 9 # 22 # 23 # 33 #
+--query_vertical show slave status;
+# connect to master and ensure data it there.
+--connection master
+SELECT * FROM t1 ORDER BY a;
+#SELECT * FROM t2 ORDER BY a;
+--replace_column 1 # 4 # 7 # 8 # 9 # 22 # 23 # 33 #
+--query_vertical show slave status;
+
+# stop replication on "master" as not to replicate
+# shutdown circularly, eg drop table
+--connection master
+STOP SLAVE;
+
+# cleanup
+--connection master
+DROP TABLE t1;

sql/rpl_injector.cc

@@ -75,9 +75,11 @@ int injector::transaction::use_table(server_id_type sid, table tbl)
   if ((error= check_state(TABLE_STATE)))
     DBUG_RETURN(error);
 
+  server_id_type save_id= m_thd->server_id;
   m_thd->set_server_id(sid);
   error= m_thd->binlog_write_table_map(tbl.get_table(),
                                        tbl.is_transactional());
+  m_thd->set_server_id(save_id);
   DBUG_RETURN(error);
 }
 
@@ -91,9 +93,11 @@ int injector::transaction::write_row (server_id_type sid, table tbl,
    if (int 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);
+   m_thd->set_server_id(save_id);
    DBUG_RETURN(0);
 }
 
@@ -107,9 +111,11 @@ int injector::transaction::delete_row(server_id_type sid, table tbl,
    if (int 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);
+   m_thd->set_server_id(save_id);
    DBUG_RETURN(0);
 }
 
@@ -123,9 +129,11 @@ int injector::transaction::update_row(server_id_type sid, table tbl,
    if (int 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);
+   m_thd->set_server_id(save_id);
    DBUG_RETURN(0);
 }