BUG#46166

Merging to latest mysql-5.1-bugteam.

17 files changed, 962 insertions, 73 deletions

diff --git a/mysql-test/include/io_thd_fault_injection.inc b/mysql-test/include/io_thd_fault_injection.inc
new file mode 100644
index 00000000000..a76e46d772a
--- /dev/null
+++ b/mysql-test/include/io_thd_fault_injection.inc
@@ -0,0 +1,21 @@
+#
+#  Takes the flag as an argument:
+#  -- let $io_thd_injection_fault_flag=+d,fault_injection_new_file_rotate_event
+#  -- source include/io_thd_fault_injection.inc
+#
+
+SET @old_debug=@@global.debug;
+-- disable_warnings
+-- source include/stop_slave.inc 
+-- enable_warnings
+-- eval SET GLOBAL debug="+d,$io_thd_injection_fault_flag"
+
+START SLAVE io_thread;
+-- source include/wait_for_slave_io_to_stop.inc
+-- source include/wait_for_slave_io_error.inc
+
+-- eval SET GLOBAL debug="-d,$io_thd_injection_fault_flag"
+SET GLOBAL debug=@old_debug;
+
+# restart because slave is in bad shape
+-- source include/restart_mysqld.inc
diff --git a/mysql-test/include/restart_mysqld.inc b/mysql-test/include/restart_mysqld.inc
index d92115f0a61..f750385e300 100644
--- a/mysql-test/include/restart_mysqld.inc
+++ b/mysql-test/include/restart_mysqld.inc
@@ -1,14 +1,16 @@
 
 # Write file to make mysql-test-run.pl expect the "crash", but don't start
 # it until it's told to
---exec echo "wait" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--let $_server_id= `SELECT @@server_id`
+--let $_expect_file_name= $MYSQLTEST_VARDIR/tmp/mysqld.$_server_id.expect
+--exec echo "wait" > $_expect_file_name
 
 # Send shutdown to the connected server and give
 # it 10 seconds to die before zapping it
 shutdown_server 10;
 
 # Write file to make mysql-test-run.pl start up the server again
---exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--exec echo "restart" > $_expect_file_name
 
 # Turn on reconnect
 --enable_reconnect
diff --git a/mysql-test/suite/binlog/r/binlog_index.result b/mysql-test/suite/binlog/r/binlog_index.result
index 52d698e9f96..6b422b82191 100644
--- a/mysql-test/suite/binlog/r/binlog_index.result
+++ b/mysql-test/suite/binlog/r/binlog_index.result
@@ -2,7 +2,9 @@ call mtr.add_suppression('Attempting backtrace');
 call mtr.add_suppression('MSYQL_BIN_LOG::purge_logs failed to process registered files that would be purged.');
 call mtr.add_suppression('MSYQL_BIN_LOG::open failed to sync the index file');
 call mtr.add_suppression('Turning logging off for the whole duration of the MySQL server process.');
+call mtr.add_suppression('Could not open .*');
 call mtr.add_suppression('MSYQL_BIN_LOG::purge_logs failed to clean registers before purging logs.');
+RESET MASTER;
 flush logs;
 flush logs;
 flush logs;
@@ -116,11 +118,31 @@ master-bin.000011
 # This should put the server in unsafe state and stop
 # accepting any command. If we inject a fault at this
 # point and continue the execution the server crashes.
-# Besides the flush command does not report an error.
 #
+SET @index=LOAD_FILE('MYSQLTEST_VARDIR/mysqld.1/data//master-bin.index');
+SELECT @index;
+@index
+master-bin.000006
+master-bin.000007
+master-bin.000008
+master-bin.000009
+master-bin.000010
+master-bin.000011
+
 # fault_injection_registering_index
 SET SESSION debug="+d,fault_injection_registering_index";
 flush logs;
+ERROR HY000: Can't open file: './master-bin.000012' (errno: 1)
+SET @index=LOAD_FILE('MYSQLTEST_VARDIR/mysqld.1/data//master-bin.index');
+SELECT @index;
+@index
+master-bin.000006
+master-bin.000007
+master-bin.000008
+master-bin.000009
+master-bin.000010
+master-bin.000011
+
 SET @index=LOAD_FILE('MYSQLTEST_VARDIR/mysqld.1/data//master-bin.index');
 SELECT @index;
 @index
@@ -135,6 +157,18 @@ master-bin.000012
 # fault_injection_updating_index
 SET SESSION debug="+d,fault_injection_updating_index";
 flush logs;
+ERROR HY000: Can't open file: './master-bin.000013' (errno: 1)
+SET @index=LOAD_FILE('MYSQLTEST_VARDIR/mysqld.1/data//master-bin.index');
+SELECT @index;
+@index
+master-bin.000006
+master-bin.000007
+master-bin.000008
+master-bin.000009
+master-bin.000010
+master-bin.000011
+master-bin.000012
+
 SET @index=LOAD_FILE('MYSQLTEST_VARDIR/mysqld.1/data//master-bin.index');
 SELECT @index;
 @index
diff --git a/mysql-test/suite/binlog/t/binlog_index.test b/mysql-test/suite/binlog/t/binlog_index.test
index 9d4a49602a6..d2b34083182 100644
--- a/mysql-test/suite/binlog/t/binlog_index.test
+++ b/mysql-test/suite/binlog/t/binlog_index.test
@@ -10,9 +10,12 @@ call mtr.add_suppression('Attempting backtrace');
 call mtr.add_suppression('MSYQL_BIN_LOG::purge_logs failed to process registered files that would be purged.');
 call mtr.add_suppression('MSYQL_BIN_LOG::open failed to sync the index file');
 call mtr.add_suppression('Turning logging off for the whole duration of the MySQL server process.');
+call mtr.add_suppression('Could not open .*');
 call mtr.add_suppression('MSYQL_BIN_LOG::purge_logs failed to clean registers before purging logs.');
 let $old=`select @@debug`;
 
+RESET MASTER;
+
 let $MYSQLD_DATADIR= `select @@datadir`;
 let $INDEX=$MYSQLD_DATADIR/master-bin.index;
 
@@ -205,12 +208,25 @@ SELECT @index;
 --echo # This should put the server in unsafe state and stop
 --echo # accepting any command. If we inject a fault at this
 --echo # point and continue the execution the server crashes.
---echo # Besides the flush command does not report an error.
 --echo #
 
+--chmod 0644 $INDEX
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval SET @index=LOAD_FILE('$index')
+-- replace_regex /\.[\\\/]master/master/
+SELECT @index;
+
 --echo # fault_injection_registering_index
 SET SESSION debug="+d,fault_injection_registering_index";
+-- error ER_CANT_OPEN_FILE
 flush logs;
+
+--chmod 0644 $INDEX
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval SET @index=LOAD_FILE('$index')
+-- replace_regex /\.[\\\/]master/master/
+SELECT @index;
+
 --source include/restart_mysqld.inc
 
 --chmod 0644 $INDEX
@@ -221,7 +237,15 @@ SELECT @index;
 
 --echo # fault_injection_updating_index
 SET SESSION debug="+d,fault_injection_updating_index";
+-- error ER_CANT_OPEN_FILE
 flush logs;
+
+--chmod 0644 $INDEX
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval SET @index=LOAD_FILE('$index')
+-- replace_regex /\.[\\\/]master/master/
+SELECT @index;
+
 --source include/restart_mysqld.inc
 
 --chmod 0644 $INDEX
diff --git a/mysql-test/suite/rpl/r/rpl_binlog_errors.result b/mysql-test/suite/rpl/r/rpl_binlog_errors.result
new file mode 100644
index 00000000000..e67b60860ca
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_binlog_errors.result
@@ -0,0 +1,274 @@
+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;
+#######################################################################
+####################### PART 1: MASTER TESTS ##########################
+#######################################################################
+include/stop_slave.inc
+call mtr.add_suppression("Can't generate a unique log-filename");
+call mtr.add_suppression("Writing one row to the row-based binary log failed.*");
+call mtr.add_suppression("Error writing file .*");
+SET @old_debug= @@global.debug;
+SELECT repeat('x',8192) INTO OUTFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data';
+SELECT repeat('x',10) INTO OUTFILE 'MYSQLTEST_VARDIR/tmp/bug_46166-2.data';
+RESET MASTER;
+###################### TEST #1
+FLUSH LOGS;
+# assert: must show two binlogs
+show binary logs;
+Log_name	File_size
+master-bin.000001	#
+master-bin.000002	#
+###################### TEST #2
+RESET MASTER;
+SET GLOBAL debug="+d,error_unique_log_filename";
+FLUSH LOGS;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+# assert: must show one binlog
+show binary logs;
+Log_name	File_size
+master-bin.000001	#
+SET GLOBAL debug="";
+RESET MASTER;
+###################### TEST #3
+CREATE TABLE t1 (a int);
+CREATE TABLE t2 (a TEXT) Engine=InnoDB;
+CREATE TABLE t4 (a TEXT);
+INSERT INTO t1 VALUES (1);
+RESET MASTER;
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data' INTO TABLE t2;
+# assert: must show two binlog
+show binary logs;
+Log_name	File_size
+master-bin.000001	#
+master-bin.000002	#
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+###################### TEST #4
+SET GLOBAL debug="+d,error_unique_log_filename";
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data' INTO TABLE t2;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+# assert: must show one entry
+SELECT count(*) FROM t2;
+count(*)
+1
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+###################### TEST #5
+SET GLOBAL debug="+d,error_unique_log_filename";
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166-2.data' INTO TABLE t2;
+# assert: must show one entry
+SELECT count(*) FROM t2;
+count(*)
+1
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+###################### TEST #6
+SET GLOBAL debug="+d,error_unique_log_filename";
+SET AUTOCOMMIT=0;
+INSERT INTO t2 VALUES ('muse');
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data' INTO TABLE t2;
+INSERT INTO t2 VALUES ('muse');
+COMMIT;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+# assert: must show three entries
+SELECT count(*) FROM t2;
+count(*)
+3
+SET AUTOCOMMIT= 1;
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+###################### TEST #7
+SET GLOBAL debug="+d,error_unique_log_filename";
+SELECT count(*) FROM t4;
+count(*)
+0
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data' INTO TABLE t4;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+# assert: must show 1 entry
+SELECT count(*) FROM t4;
+count(*)
+1
+### check that the incident event is written to the current log
+SET GLOBAL debug="-d,error_unique_log_filename";
+FLUSH LOGS;
+SHOW BINLOG EVENTS IN 'BINLOG_FILE' FROM <binlog_start> LIMIT 1;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+BINLOG_FILE	#	Incident	#	#	#1 (LOST_EVENTS)
+DELETE FROM t4;
+RESET MASTER;
+###################### TEST #8
+SET GLOBAL debug="+d,error_unique_log_filename";
+# must show 0 entries
+SELECT count(*) FROM t4;
+count(*)
+0
+SELECT count(*) FROM t2;
+count(*)
+0
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data' INTO TABLE t4;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/bug_46166.data' INTO TABLE t2;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+INSERT INTO t2 VALUES ('aaa'), ('bbb'), ('ccc');
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+# INFO: Count(*) Before Offending DELETEs
+# assert: must show 1 entry
+SELECT count(*) FROM t4;
+count(*)
+1
+# assert: must show 4 entries
+SELECT count(*) FROM t2;
+count(*)
+4
+DELETE FROM t4;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+DELETE FROM t2;
+ERROR HY000: Can't generate a unique log-filename master-bin.(1-999)
+
+# INFO: Count(*) After Offending DELETEs
+# assert: must show zero entries
+SELECT count(*) FROM t4;
+count(*)
+0
+SELECT count(*) FROM t2;
+count(*)
+0
+SET GLOBAL debug="-d,error_unique_log_filename";
+###################### TEST #9
+SET GLOBAL debug="+d,error_unique_log_filename";
+SET SQL_LOG_BIN=0;
+INSERT INTO t2 VALUES ('aaa'), ('bbb'), ('ccc'), ('ddd');
+INSERT INTO t4 VALUES ('eee'), ('fff'), ('ggg'), ('hhh');
+# assert: must show four entries
+SELECT count(*) FROM t2;
+count(*)
+4
+SELECT count(*) FROM t4;
+count(*)
+4
+DELETE FROM t2;
+DELETE FROM t4;
+# assert: must show zero entries
+SELECT count(*) FROM t2;
+count(*)
+0
+SELECT count(*) FROM t4;
+count(*)
+0
+SET SQL_LOG_BIN=1;
+SET GLOBAL debug="-d,error_unique_log_filename";
+###################### TEST #10
+call mtr.add_suppression("MSYQL_BIN_LOG::open failed to sync the index file.");
+call mtr.add_suppression("Could not open .*");
+RESET MASTER;
+SHOW WARNINGS;
+Level	Code	Message
+SET GLOBAL debug="+d,fault_injection_registering_index";
+FLUSH LOGS;
+ERROR HY000: Can't open file: './master-bin.000002' (errno: 1)
+SET GLOBAL debug="-d,fault_injection_registering_index";
+SHOW BINARY LOGS;
+ERROR HY000: You are not using binary logging
+CREATE TABLE t5 (a INT);
+INSERT INTO t4 VALUES ('bbbbb');
+INSERT INTO t2 VALUES ('aaaaa');
+DELETE FROM t4;
+DELETE FROM t2;
+DROP TABLE t5;
+###################### TEST #11
+SET GLOBAL debug="+d,fault_injection_openning_index";
+FLUSH LOGS;
+ERROR HY000: Can't open file: './master-bin.index' (errno: 1)
+SET GLOBAL debug="-d,fault_injection_openning_index";
+RESET MASTER;
+ERROR HY000: Binlog closed, cannot RESET MASTER
+CREATE TABLE t5 (a INT);
+INSERT INTO t4 VALUES ('bbbbb');
+INSERT INTO t2 VALUES ('aaaaa');
+DELETE FROM t4;
+DELETE FROM t2;
+DROP TABLE t5;
+###################### TEST #12
+SET GLOBAL debug="+d,fault_injection_new_file_rotate_event";
+FLUSH LOGS;
+ERROR HY000: Can't open file: 'master-bin' (errno: 0)
+SET GLOBAL debug="-d,fault_injection_new_file_rotate_event";
+RESET MASTER;
+ERROR HY000: Binlog closed, cannot RESET MASTER
+CREATE TABLE t5 (a INT);
+INSERT INTO t4 VALUES ('bbbbb');
+INSERT INTO t2 VALUES ('aaaaa');
+DELETE FROM t4;
+DELETE FROM t2;
+DROP TABLE t5;
+SET GLOBAL debug= @old_debug;
+DROP TABLE t1, t2, t4;
+RESET MASTER;
+include/start_slave.inc
+#######################################################################
+####################### PART 2: SLAVE TESTS ###########################
+#######################################################################
+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;
+call mtr.add_suppression("Slave I/O: Relay log write failure: could not queue event from master.*");
+call mtr.add_suppression("Error writing file .*");
+call mtr.add_suppression("Could not open .*");
+call mtr.add_suppression("MSYQL_BIN_LOG::open failed to sync the index file.");
+call mtr.add_suppression("Can't generate a unique log-filename .*");
+###################### TEST #13
+SET @old_debug=@@global.debug;
+include/stop_slave.inc
+SET GLOBAL debug="+d,error_unique_log_filename";
+START SLAVE io_thread;
+Last_IO_Error = Relay log write failure: could not queue event from master
+SET GLOBAL debug="-d,error_unique_log_filename";
+SET GLOBAL debug=@old_debug;
+###################### TEST #14
+SET @old_debug=@@global.debug;
+include/stop_slave.inc
+SET GLOBAL debug="+d,fault_injection_new_file_rotate_event";
+START SLAVE io_thread;
+Last_IO_Error = Relay log write failure: could not queue event from master
+SET GLOBAL debug="-d,fault_injection_new_file_rotate_event";
+SET GLOBAL debug=@old_debug;
+###################### TEST #15
+SET @old_debug=@@global.debug;
+include/stop_slave.inc
+SET GLOBAL debug="+d,fault_injection_registering_index";
+START SLAVE io_thread;
+Last_IO_Error = Relay log write failure: could not queue event from master
+SET GLOBAL debug="-d,fault_injection_registering_index";
+SET GLOBAL debug=@old_debug;
+###################### TEST #16
+SET @old_debug=@@global.debug;
+include/stop_slave.inc
+SET GLOBAL debug="+d,fault_injection_openning_index";
+START SLAVE io_thread;
+Last_IO_Error = Relay log write failure: could not queue event from master
+SET GLOBAL debug="-d,fault_injection_openning_index";
+SET GLOBAL debug=@old_debug;
+include/stop_slave.inc
+SET GLOBAL debug=@old_debug;
+RESET SLAVE;
+RESET MASTER;
+include/start_slave.inc
diff --git a/mysql-test/suite/rpl/t/rpl_binlog_errors-master.opt b/mysql-test/suite/rpl/t/rpl_binlog_errors-master.opt
new file mode 100644
index 00000000000..f8e46a44854
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_binlog_errors-master.opt
@@ -0,0 +1 @@
+--max_binlog_size=4096
diff --git a/mysql-test/suite/rpl/t/rpl_binlog_errors.test b/mysql-test/suite/rpl/t/rpl_binlog_errors.test
new file mode 100644
index 00000000000..ae83659a38a
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_binlog_errors.test
@@ -0,0 +1,413 @@
+# BUG#46166: MYSQL_BIN_LOG::new_file_impl is not propagating error
+#            when generating new name.
+#  
+# WHY
+# ===
+#
+# We want to check whether error is reported or not when
+# new_file_impl fails (this may happen when rotation is not
+# possible because there is some problem finding an 
+# unique filename).
+#
+# HOW
+# ===
+# 
+# Test cases are documented inline.
+
+-- source include/master-slave.inc
+-- source include/have_innodb.inc
+-- source include/have_debug.inc
+
+-- echo #######################################################################
+-- echo ####################### PART 1: MASTER TESTS ##########################
+-- echo #######################################################################
+
+
+### ACTION: stopping slave as it is not needed for the first part of
+###         the test
+
+-- connection slave
+-- source include/stop_slave.inc
+-- connection master
+
+call mtr.add_suppression("Can't generate a unique log-filename");
+call mtr.add_suppression("Writing one row to the row-based binary log failed.*");
+call mtr.add_suppression("Error writing file .*");
+
+SET @old_debug= @@global.debug;
+
+### ACTION: create a large file (> 4096 bytes) that will be later used
+###         in LOAD DATA INFILE to check binlog errors in its vacinity
+-- let $load_file= $MYSQLTEST_VARDIR/tmp/bug_46166.data
+-- let $MYSQLD_DATADIR= `select @@datadir`
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval SELECT repeat('x',8192) INTO OUTFILE '$load_file'
+
+### ACTION: create a small file (< 4096 bytes) that will be later used
+###         in LOAD DATA INFILE to check for absence of binlog errors
+###         when file loading this file does not force flushing and
+###         rotating the binary log
+-- let $load_file2= $MYSQLTEST_VARDIR/tmp/bug_46166-2.data
+-- let $MYSQLD_DATADIR= `select @@datadir`
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval SELECT repeat('x',10) INTO OUTFILE '$load_file2'
+
+RESET MASTER;
+
+-- echo ###################### TEST #1
+
+### ASSERTION: no problem flushing logs (should show two binlogs)
+FLUSH LOGS;
+-- echo # assert: must show two binlogs
+-- source include/show_binary_logs.inc
+
+-- echo ###################### TEST #2
+
+### ASSERTION: check that FLUSH LOGS actually fails and reports
+###            failure back to the user if find_uniq_filename fails
+###            (should show just one binlog)
+
+RESET MASTER;
+SET GLOBAL debug="+d,error_unique_log_filename";
+-- error ER_NO_UNIQUE_LOGFILE
+FLUSH LOGS;
+-- echo # assert: must show one binlog
+-- source include/show_binary_logs.inc
+
+### ACTION: clean up and move to next test
+SET GLOBAL debug="";
+RESET MASTER;
+
+-- echo ###################### TEST #3
+
+### ACTION: create some tables (t1, t2, t4) and insert some values in
+###         table t1
+CREATE TABLE t1 (a int);
+CREATE TABLE t2 (a TEXT) Engine=InnoDB;
+CREATE TABLE t4 (a TEXT);
+INSERT INTO t1 VALUES (1);
+RESET MASTER;
+
+### ASSERTION: we force rotation of the binary log because it exceeds
+###            the max_binlog_size option (should show two binary
+###            logs)
+
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA INFILE '$load_file' INTO TABLE t2
+
+# shows two binary logs
+-- echo # assert: must show two binlog
+-- source include/show_binary_logs.inc
+
+# clean up the table and the binlog to be used in next part of test
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+
+-- echo ###################### TEST #4
+
+### ASSERTION: load the big file into a transactional table and check
+###            that it reports error. The table will contain the
+###            changes performed despite the fact that it reported an
+###            error.
+
+SET GLOBAL debug="+d,error_unique_log_filename";
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- error ER_NO_UNIQUE_LOGFILE
+-- eval LOAD DATA INFILE '$load_file' INTO TABLE t2
+
+# show table 
+-- echo # assert: must show one entry
+SELECT count(*) FROM t2;
+
+# clean up the table and the binlog to be used in next part of test
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+
+-- echo ###################### TEST #5
+
+### ASSERTION: load the small file into a transactional table and
+###            check that it succeeds
+
+SET GLOBAL debug="+d,error_unique_log_filename";
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA INFILE '$load_file2' INTO TABLE t2
+
+# show table 
+-- echo # assert: must show one entry
+SELECT count(*) FROM t2;
+
+# clean up the table and the binlog to be used in next part of test
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+
+-- echo ###################### TEST #6
+
+### ASSERTION: check that even if one is using a transactional table
+###            and explicit transactions (no autocommit) if rotation
+###            fails we get the error. Transaction is not rolledback
+###            because rotation happens after the commit.
+
+SET GLOBAL debug="+d,error_unique_log_filename";
+SET AUTOCOMMIT=0;
+INSERT INTO t2 VALUES ('muse');
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA INFILE '$load_file' INTO TABLE t2
+INSERT INTO t2 VALUES ('muse');
+-- error ER_NO_UNIQUE_LOGFILE
+COMMIT;
+
+### ACTION: Show the contents of the table after the test 
+-- echo # assert: must show three entries
+SELECT count(*) FROM t2;
+
+### ACTION: clean up and move to the next test 
+SET AUTOCOMMIT= 1;
+SET GLOBAL debug="-d,error_unique_log_filename";
+DELETE FROM t2;
+RESET MASTER;
+
+-- echo ###################### TEST #7
+
+### ASSERTION: check that on a non-transactional table, if rotation
+###            fails then an error is reported and an incident event
+###            is written to the current binary log.
+
+SET GLOBAL debug="+d,error_unique_log_filename";
+SELECT count(*) FROM t4;
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- error ER_NO_UNIQUE_LOGFILE
+-- eval LOAD DATA INFILE '$load_file' INTO TABLE t4
+
+-- echo # assert: must show 1 entry
+SELECT count(*) FROM t4;
+
+-- echo ### check that the incident event is written to the current log
+SET GLOBAL debug="-d,error_unique_log_filename";
+-- let $binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1)
+-- let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1)
+
+# 53 is the size of the incident event, so we start from 22 bytes before the
+# current position
+-- let $binlog_start = `SELECT $binlog_start - 53`
+FLUSH LOGS;
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR $binlog_start <binlog_start> $binlog_file BINLOG_FILE
+-- replace_column 2 # 4 # 5 #
+-- eval SHOW BINLOG EVENTS IN '$binlog_file' FROM $binlog_start LIMIT 1
+
+# clean up and move to next test
+DELETE FROM t4;
+RESET MASTER;
+
+-- echo ###################### TEST #8
+
+### ASSERTION: check that statements end up in error but they succeed
+###            on changing the data. 
+
+SET GLOBAL debug="+d,error_unique_log_filename";
+-- echo # must show 0 entries
+SELECT count(*) FROM t4;
+SELECT count(*) FROM t2;
+
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- error ER_NO_UNIQUE_LOGFILE
+-- eval LOAD DATA INFILE '$load_file' INTO TABLE t4
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- error ER_NO_UNIQUE_LOGFILE
+-- eval LOAD DATA INFILE '$load_file' INTO TABLE t2
+-- error ER_NO_UNIQUE_LOGFILE
+INSERT INTO t2 VALUES ('aaa'), ('bbb'), ('ccc');
+
+-- echo # INFO: Count(*) Before Offending DELETEs
+-- echo # assert: must show 1 entry
+SELECT count(*) FROM t4;
+-- echo # assert: must show 4 entries
+SELECT count(*) FROM t2;
+
+-- error ER_NO_UNIQUE_LOGFILE
+DELETE FROM t4;
+-- error ER_NO_UNIQUE_LOGFILE
+DELETE FROM t2;
+
+-- echo # INFO: Count(*) After Offending DELETEs
+-- echo # assert: must show zero entries
+SELECT count(*) FROM t4;
+SELECT count(*) FROM t2;
+
+# remove fault injection
+SET GLOBAL debug="-d,error_unique_log_filename";
+
+-- echo ###################### TEST #9
+
+### ASSERTION: check that if we disable binlogging, then statements
+###            succeed.
+SET GLOBAL debug="+d,error_unique_log_filename";
+SET SQL_LOG_BIN=0;
+INSERT INTO t2 VALUES ('aaa'), ('bbb'), ('ccc'), ('ddd');
+INSERT INTO t4 VALUES ('eee'), ('fff'), ('ggg'), ('hhh');
+-- echo # assert: must show four entries
+SELECT count(*) FROM t2;
+SELECT count(*) FROM t4;
+DELETE FROM t2;
+DELETE FROM t4;
+-- echo # assert: must show zero entries
+SELECT count(*) FROM t2;
+SELECT count(*) FROM t4;
+SET SQL_LOG_BIN=1;
+SET GLOBAL debug="-d,error_unique_log_filename";
+
+-- echo ###################### TEST #10
+
+### ASSERTION: check that error is reported if there is a failure
+###            while registering the index file and the binary log
+###            file or failure to write the rotate event.
+
+call mtr.add_suppression("MSYQL_BIN_LOG::open failed to sync the index file.");
+call mtr.add_suppression("Could not open .*");
+
+RESET MASTER;
+SHOW WARNINGS;
+
+# +d,fault_injection_registering_index => injects fault on MYSQL_BIN_LOG::open
+SET GLOBAL debug="+d,fault_injection_registering_index";
+-- error ER_CANT_OPEN_FILE
+FLUSH LOGS;
+SET GLOBAL debug="-d,fault_injection_registering_index";
+
+-- error ER_NO_BINARY_LOGGING
+SHOW BINARY LOGS;
+
+# issue some statements and check that they don't fail
+CREATE TABLE t5 (a INT);
+INSERT INTO t4 VALUES ('bbbbb');
+INSERT INTO t2 VALUES ('aaaaa');
+DELETE FROM t4;
+DELETE FROM t2;
+DROP TABLE t5;
+
+-- echo ###################### TEST #11
+
+### ASSERTION: check that error is reported if there is a failure
+###            while opening the index file and the binary log file or
+###            failure to write the rotate event.
+
+# restart the server so that we have binlog again
+--source include/restart_mysqld.inc
+
+# +d,fault_injection_openning_index => injects fault on MYSQL_BIN_LOG::open_index_file
+SET GLOBAL debug="+d,fault_injection_openning_index";
+-- error ER_CANT_OPEN_FILE
+FLUSH LOGS;
+SET GLOBAL debug="-d,fault_injection_openning_index";
+
+-- error ER_FLUSH_MASTER_BINLOG_CLOSED
+RESET MASTER;
+
+# issue some statements and check that they don't fail
+CREATE TABLE t5 (a INT);
+INSERT INTO t4 VALUES ('bbbbb');
+INSERT INTO t2 VALUES ('aaaaa');
+DELETE FROM t4;
+DELETE FROM t2;
+DROP TABLE t5;
+
+# restart the server so that we have binlog again
+-- source include/restart_mysqld.inc
+
+-- echo ###################### TEST #12
+
+### ASSERTION: check that error is reported if there is a failure
+###            while writing the rotate event when creating a new log
+###            file.
+
+# +d,fault_injection_new_file_rotate_event => injects fault on MYSQL_BIN_LOG::MYSQL_BIN_LOG::new_file_impl
+SET GLOBAL debug="+d,fault_injection_new_file_rotate_event";
+-- error ER_ERROR_ON_WRITE
+FLUSH LOGS;
+SET GLOBAL debug="-d,fault_injection_new_file_rotate_event";
+
+-- error ER_FLUSH_MASTER_BINLOG_CLOSED
+RESET MASTER;
+
+# issue some statements and check that they don't fail
+CREATE TABLE t5 (a INT);
+INSERT INTO t4 VALUES ('bbbbb');
+INSERT INTO t2 VALUES ('aaaaa');
+DELETE FROM t4;
+DELETE FROM t2;
+DROP TABLE t5;
+
+# restart the server so that we have binlog again
+-- source include/restart_mysqld.inc
+
+## clean up
+SET GLOBAL debug= @old_debug;
+DROP TABLE t1, t2, t4;
+RESET MASTER;
+
+# restart slave again
+-- connection slave
+-- source include/start_slave.inc
+-- connection master
+
+-- echo #######################################################################
+-- echo ####################### PART 2: SLAVE TESTS ###########################
+-- echo #######################################################################
+
+### setup
+-- connection master
+# master-slave-reset starts the slave automatically
+-- source include/master-slave-reset.inc
+-- connection slave
+
+# slave suppressions
+
+call mtr.add_suppression("Slave I/O: Relay log write failure: could not queue event from master.*");
+call mtr.add_suppression("Error writing file .*");
+call mtr.add_suppression("Could not open .*");
+call mtr.add_suppression("MSYQL_BIN_LOG::open failed to sync the index file.");
+call mtr.add_suppression("Can't generate a unique log-filename .*");
+-- echo ###################### TEST #13
+
+#### ASSERTION: check against unique log filename error
+-- let $io_thd_injection_fault_flag= error_unique_log_filename
+-- let $slave_io_errno= 1595
+-- let $show_slave_io_error= 1
+-- source include/io_thd_fault_injection.inc
+
+-- echo ###################### TEST #14
+
+#### ASSERTION: check against rotate failing
+-- let $io_thd_injection_fault_flag= fault_injection_new_file_rotate_event
+-- let $slave_io_errno= 1595
+-- let $show_slave_io_error= 1
+-- source include/io_thd_fault_injection.inc
+
+-- echo ###################### TEST #15
+
+#### ASSERTION: check against relay log open failure
+-- let $io_thd_injection_fault_flag= fault_injection_registering_index
+-- let $slave_io_errno= 1595
+-- let $show_slave_io_error= 1
+-- source include/io_thd_fault_injection.inc
+
+-- echo ###################### TEST #16
+
+#### ASSERTION: check against relay log index open failure
+-- let $io_thd_injection_fault_flag= fault_injection_openning_index
+-- let $slave_io_errno= 1595
+-- let $show_slave_io_error= 1
+-- source include/io_thd_fault_injection.inc
+
+### clean up
+-- disable_warnings
+-- source include/stop_slave.inc
+-- enable_warnings
+SET GLOBAL debug=@old_debug;
+RESET SLAVE;
+RESET MASTER;
+-- source include/start_slave.inc
+-- connection master
+-- source include/master-slave-end.inc
diff --git a/sql/handler.cc b/sql/handler.cc
index 33c7f518838..ce4e4a9e3a9 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1192,7 +1192,11 @@ int ha_commit_trans(THD *thd, bool all)
     error=ha_commit_one_phase(thd, all) ? (cookie ? 2 : 1) : 0;
     DBUG_EXECUTE_IF("crash_commit_before_unlog", DBUG_SUICIDE(););
     if (cookie)
-      tc_log->unlog(cookie, xid);
+      if(tc_log->unlog(cookie, xid))
+      {
+        error= 2;
+        goto end;
+      }
     DBUG_EXECUTE_IF("crash_commit_after", DBUG_SUICIDE(););
 end:
     if (rw_trans)
diff --git a/sql/log.cc b/sql/log.cc
index cff69b4cf51..5fcb5fe0367 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1871,10 +1871,11 @@ static int find_uniq_filename(char *name)
   *end='.';
   length= (size_t) (end-start+1);
 
-  if (!(dir_info = my_dir(buff,MYF(MY_DONT_SORT))))
+  if ((DBUG_EVALUATE_IF("error_unique_log_filename", 1, 
+      !(dir_info = my_dir(buff,MYF(MY_DONT_SORT))))))
   {						// This shouldn't happen
     strmov(end,".1");				// use name+1
-    DBUG_RETURN(0);
+    DBUG_RETURN(1);
   }
   file_info= dir_info->dir_entry;
   for (i=dir_info->number_off_files ; i-- ; file_info++)
@@ -1888,8 +1889,7 @@ static int find_uniq_filename(char *name)
   my_dirend(dir_info);
 
   *end++='.';
-  sprintf(end,"%06ld",max_found+1);
-  DBUG_RETURN(0);
+  DBUG_RETURN((sprintf(end,"%06ld",max_found+1) < 0));
 }
 
 
@@ -2101,6 +2101,8 @@ int MYSQL_LOG::generate_new_name(char *new_name, const char *log_name)
     {
       if (find_uniq_filename(new_name))
       {
+        my_printf_error(ER_NO_UNIQUE_LOGFILE, ER(ER_NO_UNIQUE_LOGFILE),
+                        MYF(ME_FATALERROR), log_name);
 	sql_print_error(ER(ER_NO_UNIQUE_LOGFILE), log_name);
 	return 1;
       }
@@ -3066,7 +3068,8 @@ bool MYSQL_BIN_LOG::reset_logs(THD* thd)
   if (!thd->slave_thread)
     need_start_event=1;
   if (!open_index_file(index_file_name, 0, FALSE))
-    open(save_name, log_type, 0, io_cache_type, no_auto_events, max_size, 0, FALSE);
+    if ((error= open(save_name, log_type, 0, io_cache_type, no_auto_events, max_size, 0, FALSE)))
+      goto err;
   my_free((uchar*) save_name, MYF(0));
 
 err:
@@ -3728,17 +3731,23 @@ bool MYSQL_BIN_LOG::is_active(const char *log_file_name_arg)
   incapsulation 3) allows external access to the class without
   a lock (which is not possible with private new_file_without_locking
   method).
+
+  @retval
+    nonzero - error
 */
 
-void MYSQL_BIN_LOG::new_file()
+int MYSQL_BIN_LOG::new_file()
 {
-  new_file_impl(1);
+  return new_file_impl(1);
 }
 
-
-void MYSQL_BIN_LOG::new_file_without_locking()
+/*
+  @retval
+    nonzero - error
+ */
+int MYSQL_BIN_LOG::new_file_without_locking()
 {
-  new_file_impl(0);
+  return new_file_impl(0);
 }
 
 
@@ -3747,19 +3756,23 @@ void MYSQL_BIN_LOG::new_file_without_locking()
 
   @param need_lock		Set to 1 if caller has not locked LOCK_log
 
+  @retval
+    nonzero - error
+
   @note
     The new file name is stored last in the index file
 */
 
-void MYSQL_BIN_LOG::new_file_impl(bool need_lock)
+int MYSQL_BIN_LOG::new_file_impl(bool need_lock)
 {
-  char new_name[FN_REFLEN], *new_name_ptr, *old_name;
+  int error= 0, close_on_error= FALSE;
+  char new_name[FN_REFLEN], *new_name_ptr, *old_name, *file_to_open;
 
   DBUG_ENTER("MYSQL_BIN_LOG::new_file_impl");
   if (!is_open())
   {
     DBUG_PRINT("info",("log is closed"));
-    DBUG_VOID_RETURN;
+    DBUG_RETURN(error);
   }
 
   if (need_lock)
@@ -3797,7 +3810,7 @@ void MYSQL_BIN_LOG::new_file_impl(bool need_lock)
     We have to do this here and not in open as we want to store the
     new file name in the current binary log file.
   */
-  if (generate_new_name(new_name, name))
+  if ((error= generate_new_name(new_name, name)))
     goto end;
   new_name_ptr=new_name;
 
@@ -3811,7 +3824,13 @@ void MYSQL_BIN_LOG::new_file_impl(bool need_lock)
       */
       Rotate_log_event r(new_name+dirname_length(new_name),
                          0, LOG_EVENT_OFFSET, is_relay_log ? Rotate_log_event::RELAY_LOG : 0);
-      r.write(&log_file);
+      if(DBUG_EVALUATE_IF("fault_injection_new_file_rotate_event", (error=close_on_error=TRUE), FALSE) ||
+         (error= r.write(&log_file)))
+      {
+        close_on_error= TRUE;
+        my_printf_error(ER_ERROR_ON_WRITE, ER(ER_CANT_OPEN_FILE), MYF(ME_FATALERROR), name, errno);
+        goto end;
+      }
       bytes_written += r.data_written;
     }
     /*
@@ -3839,17 +3858,56 @@ void MYSQL_BIN_LOG::new_file_impl(bool need_lock)
   */
 
   /* reopen index binlog file, BUG#34582 */
-  if (!open_index_file(index_file_name, 0, FALSE))
-    open(old_name, log_type, new_name_ptr,
-         io_cache_type, no_auto_events, max_size, 1, FALSE);
+  file_to_open= index_file_name;
+  error= open_index_file(index_file_name, 0, FALSE);
+  if (!error)
+  {
+    /* reopen the binary log file. */
+    file_to_open= new_name_ptr;
+    error= open(old_name, log_type, new_name_ptr, io_cache_type,
+                no_auto_events, max_size, 1, FALSE);
+  }
+
+  /* handle reopening errors */
+  if (error)
+  {
+    my_printf_error(ER_CANT_OPEN_FILE, ER(ER_CANT_OPEN_FILE), 
+                    MYF(ME_FATALERROR), file_to_open, error);
+    close_on_error= TRUE;
+  }
+
   my_free(old_name,MYF(0));
 
 end:
+
+  if (error && close_on_error /* rotate or reopen failed */)
+  {
+    /* 
+      Close whatever was left opened.
+
+      We are keeping the behavior as it exists today, ie,
+      we disable logging and move on (see: BUG#51014).
+
+      TODO: as part of WL#1790 consider other approaches:
+       - kill mysql (safety);
+       - try multiple locations for opening a log file;
+       - switch server to protected/readonly mode
+       - ...
+    */
+    close(LOG_CLOSE_INDEX);
+    sql_print_error("Could not open %s for logging (error %d). "
+                     "Turning logging off for the whole duration "
+                     "of the MySQL server process. To turn it on "
+                     "again: fix the cause, shutdown the MySQL "
+                     "server and restart it.", 
+                     new_name_ptr, errno);
+  }
+
   if (need_lock)
     pthread_mutex_unlock(&LOCK_log);
   pthread_mutex_unlock(&LOCK_index);
 
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(error);
 }
 
 
@@ -3872,8 +3930,7 @@ bool MYSQL_BIN_LOG::append(Log_event* ev)
   bytes_written+= ev->data_written;
   DBUG_PRINT("info",("max_size: %lu",max_size));
   if ((uint) my_b_append_tell(&log_file) > max_size)
-    new_file_without_locking();
-
+    error= new_file_without_locking();
 err:
   pthread_mutex_unlock(&LOCK_log);
   signal_update();				// Safe as we don't call close
@@ -3902,8 +3959,7 @@ bool MYSQL_BIN_LOG::appendv(const char* buf, uint len,...)
   } while ((buf=va_arg(args,const char*)) && (len=va_arg(args,uint)));
   DBUG_PRINT("info",("max_size: %lu",max_size));
   if ((uint) my_b_append_tell(&log_file) > max_size)
-    new_file_without_locking();
-
+    error= new_file_without_locking();
 err:
   if (!error)
     signal_update();
@@ -4252,7 +4308,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
       if (!error)
       {
         signal_update();
-        rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
+        error= rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
       }
     }
 
@@ -4448,7 +4504,9 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
       if (flush_and_sync())
 	goto err;
       signal_update();
-      rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
+      if ((error= rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED)))
+        goto err;
+      
     }
     error=0;
 
@@ -4531,8 +4589,19 @@ bool general_log_write(THD *thd, enum enum_server_command command,
   return FALSE;
 }
 
-void MYSQL_BIN_LOG::rotate_and_purge(uint flags)
+/**
+  @note
+    If rotation fails, for instance the server was unable 
+    to create a new log file, we still try to write an 
+    incident event to the current log.
+
+  @retval
+    nonzero - error 
+*/
+int MYSQL_BIN_LOG::rotate_and_purge(uint flags)
 {
+  int error= 0;
+  DBUG_ENTER("MYSQL_BIN_LOG::rotate_and_purge");
 #ifdef HAVE_REPLICATION
   bool check_purge= false;
 #endif
@@ -4541,26 +4610,38 @@ void MYSQL_BIN_LOG::rotate_and_purge(uint flags)
   if ((flags & RP_FORCE_ROTATE) ||
       (my_b_tell(&log_file) >= (my_off_t) max_size))
   {
-    new_file_without_locking();
+    if ((error= new_file_without_locking()))
+      /** 
+         Be conservative... There are possible lost events (eg, 
+         failing to log the Execute_load_query_log_event
+         on a LOAD DATA while using a non-transactional
+         table)!
+
+         We give it a shot and try to write an incident event anyway
+         to the current log. 
+      */
+      if (!write_incident(current_thd, FALSE))
+        flush_and_sync();
+
 #ifdef HAVE_REPLICATION
     check_purge= true;
 #endif
   }
   if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
     pthread_mutex_unlock(&LOCK_log);
-
 #ifdef HAVE_REPLICATION
   /*
     NOTE: Run purge_logs wo/ holding LOCK_log
           as it otherwise will deadlock in ndbcluster_binlog_index_purge_file
   */
-  if (check_purge && expire_logs_days)
+  if (!error && check_purge && expire_logs_days)
   {
     time_t purge_time= my_time(0) - expire_logs_days*24*60*60;
     if (purge_time >= 0)
       purge_logs_before_date(purge_time);
   }
 #endif
+  DBUG_RETURN(error);
 }
 
 uint MYSQL_BIN_LOG::next_file_id()
@@ -4747,6 +4828,10 @@ bool MYSQL_BIN_LOG::write_incident(THD *thd, bool lock)
 {
   uint error= 0;
   DBUG_ENTER("MYSQL_BIN_LOG::write_incident");
+
+  if (!is_open())
+    DBUG_RETURN(error);
+
   LEX_STRING const write_error_msg=
     { C_STRING_WITH_LEN("error writing to the binary log") };
   Incident incident= INCIDENT_LOST_EVENTS;
@@ -4759,7 +4844,7 @@ bool MYSQL_BIN_LOG::write_incident(THD *thd, bool lock)
     if (!error && !(error= flush_and_sync()))
     {
       signal_update();
-      rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
+      error= rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
     }
     pthread_mutex_unlock(&LOCK_log);
   }
@@ -4871,7 +4956,8 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event,
       pthread_mutex_unlock(&LOCK_prep_xids);
     }
     else
-      rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED);
+      if (rotate_and_purge(RP_LOCK_LOG_IS_ALREADY_LOCKED))
+        goto err;
   }
   VOID(pthread_mutex_unlock(&LOCK_log));
 
@@ -5676,7 +5762,7 @@ int TC_LOG_MMAP::sync()
   cookie points directly to the memory where xid was logged.
 */
 
-void TC_LOG_MMAP::unlog(ulong cookie, my_xid xid)
+int TC_LOG_MMAP::unlog(ulong cookie, my_xid xid)
 {
   PAGE *p=pages+(cookie/tc_log_page_size);
   my_xid *x=(my_xid *)(data+cookie);
@@ -5694,6 +5780,7 @@ void TC_LOG_MMAP::unlog(ulong cookie, my_xid xid)
   if (p->waiters == 0)                 // the page is in pool and ready to rock
     pthread_cond_signal(&COND_pool);   // ping ... for overflow()
   pthread_mutex_unlock(&p->lock);
+  return 0;
 }
 
 void TC_LOG_MMAP::close()
@@ -5935,8 +6022,9 @@ int TC_LOG_BINLOG::log_xid(THD *thd, my_xid xid)
   DBUG_RETURN(!binlog_end_trans(thd, trx_data, &xle, TRUE));
 }
 
-void TC_LOG_BINLOG::unlog(ulong cookie, my_xid xid)
+int TC_LOG_BINLOG::unlog(ulong cookie, my_xid xid)
 {
+  DBUG_ENTER("TC_LOG_BINLOG::unlog");
   pthread_mutex_lock(&LOCK_prep_xids);
   DBUG_ASSERT(prepared_xids > 0);
   if (--prepared_xids == 0) {
@@ -5944,7 +6032,7 @@ void TC_LOG_BINLOG::unlog(ulong cookie, my_xid xid)
     pthread_cond_signal(&COND_prep_xids);
   }
   pthread_mutex_unlock(&LOCK_prep_xids);
-  rotate_and_purge(0);     // as ::write() did not rotate
+  DBUG_RETURN(rotate_and_purge(0));     // as ::write() did not rotate
 }
 
 int TC_LOG_BINLOG::recover(IO_CACHE *log, Format_description_log_event *fdle)
diff --git a/sql/log.h b/sql/log.h
index 8f1ed7ee90c..2b0bc6111b3 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -39,7 +39,7 @@ class TC_LOG
   virtual int open(const char *opt_name)=0;
   virtual void close()=0;
   virtual int log_xid(THD *thd, my_xid xid)=0;
-  virtual void unlog(ulong cookie, my_xid xid)=0;
+  virtual int unlog(ulong cookie, my_xid xid)=0;
 };
 
 class TC_LOG_DUMMY: public TC_LOG // use it to disable the logging
@@ -49,7 +49,7 @@ public:
   int open(const char *opt_name)        { return 0; }
   void close()                          { }
   int log_xid(THD *thd, my_xid xid)         { return 1; }
-  void unlog(ulong cookie, my_xid xid)  { }
+  int unlog(ulong cookie, my_xid xid)  { return 0; }
 };
 
 #ifdef HAVE_MMAP
@@ -94,7 +94,7 @@ class TC_LOG_MMAP: public TC_LOG
   int open(const char *opt_name);
   void close();
   int log_xid(THD *thd, my_xid xid);
-  void unlog(ulong cookie, my_xid xid);
+  int unlog(ulong cookie, my_xid xid);
   int recover();
 
   private:
@@ -283,8 +283,8 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
     new_file() is locking. new_file_without_locking() does not acquire
     LOCK_log.
   */
-  void new_file_without_locking();
-  void new_file_impl(bool need_lock);
+  int new_file_without_locking();
+  int new_file_impl(bool need_lock);
 
 public:
   MYSQL_LOG::generate_name;
@@ -314,7 +314,7 @@ public:
   int open(const char *opt_name);
   void close();
   int log_xid(THD *thd, my_xid xid);
-  void unlog(ulong cookie, my_xid xid);
+  int unlog(ulong cookie, my_xid xid);
   int recover(IO_CACHE *log, Format_description_log_event *fdle);
 #if !defined(MYSQL_CLIENT)
   int flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event);
@@ -354,7 +354,7 @@ public:
   bool open_index_file(const char *index_file_name_arg,
                        const char *log_name, bool need_mutex);
   /* Use this to start writing a new log file */
-  void new_file();
+  int new_file();
 
   void reset_gathered_updates(THD *thd);
   bool write(Log_event* event_info); // binary log write
@@ -379,7 +379,7 @@ public:
   void make_log_name(char* buf, const char* log_ident);
   bool is_active(const char* log_file_name);
   int update_log_index(LOG_INFO* linfo, bool need_update_threads);
-  void rotate_and_purge(uint flags);
+  int rotate_and_purge(uint flags);
   bool flush_and_sync();
   int purge_logs(const char *to_log, bool included,
                  bool need_mutex, bool need_update_threads,
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 04feb217d66..1838f0c924d 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1088,7 +1088,7 @@ uint cached_table_definitions(void);
 void kill_mysql(void);
 void close_connection(THD *thd, uint errcode, bool lock);
 bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables, 
-                          bool *write_to_binlog);
+                          int *write_to_binlog);
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
 bool check_access(THD *thd, ulong access, const char *db, ulong *save_priv,
 		  bool no_grant, bool no_errors, bool schema_db);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 3a33d06fb01..1a3a0886c1f 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -2839,7 +2839,7 @@ pthread_handler_t signal_hand(void *arg __attribute__((unused)))
     case SIGHUP:
       if (!abort_loop)
       {
-        bool not_used;
+        int not_used;
 	mysql_print_status();		// Print some debug info
 	reload_acl_and_cache((THD*) 0,
 			     (REFRESH_LOG | REFRESH_TABLES | REFRESH_FAST |
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index 43f4e7d849d..b797621d696 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -229,8 +229,7 @@ int injector::record_incident(THD *thd, Incident incident)
   Incident_log_event ev(thd, incident);
   if (int error= mysql_bin_log.write(&ev))
     return error;
-  mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
-  return 0;
+  return mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
 }
 
 int injector::record_incident(THD *thd, Incident incident, LEX_STRING const message)
@@ -238,6 +237,5 @@ int injector::record_incident(THD *thd, Incident incident, LEX_STRING const mess
   Incident_log_event ev(thd, incident, message);
   if (int error= mysql_bin_log.write(&ev))
     return error;
-  mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
-  return 0;
+  return mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
 }
diff --git a/sql/slave.cc b/sql/slave.cc
index 644aade517c..96319de5427 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -3398,8 +3398,7 @@ static int process_io_rotate(Master_info *mi, Rotate_log_event *rev)
     Rotate the relay log makes binlog format detection easier (at next slave
     start or mysqlbinlog)
   */
-  rotate_relay_log(mi); /* will take the right mutexes */
-  DBUG_RETURN(0);
+  DBUG_RETURN(rotate_relay_log(mi) /* will take the right mutexes */);
 }
 
 /*
@@ -4482,10 +4481,11 @@ err:
   is void).
 */
 
-void rotate_relay_log(Master_info* mi)
+int rotate_relay_log(Master_info* mi)
 {
   DBUG_ENTER("rotate_relay_log");
   Relay_log_info* rli= &mi->rli;
+  int error= 0;
 
   /* We don't lock rli->run_lock. This would lead to deadlocks. */
   pthread_mutex_lock(&mi->run_lock);
@@ -4501,7 +4501,8 @@ void rotate_relay_log(Master_info* mi)
   }
 
   /* If the relay log is closed, new_file() will do nothing. */
-  rli->relay_log.new_file();
+  if ((error= rli->relay_log.new_file()))
+    goto end;
 
   /*
     We harvest now, because otherwise BIN_LOG_HEADER_SIZE will not immediately
@@ -4519,7 +4520,7 @@ void rotate_relay_log(Master_info* mi)
   rli->relay_log.harvest_bytes_written(&rli->log_space_total);
 end:
   pthread_mutex_unlock(&mi->run_lock);
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(error);
 }
 
 
diff --git a/sql/slave.h b/sql/slave.h
index f356d28b626..da14b3ab4b0 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -189,7 +189,7 @@ int purge_relay_logs(Relay_log_info* rli, THD *thd, bool just_reset,
 		     const char** errmsg);
 void set_slave_thread_options(THD* thd);
 void set_slave_thread_default_charset(THD *thd, Relay_log_info const *rli);
-void rotate_relay_log(Master_info* mi);
+int rotate_relay_log(Master_info* mi);
 int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli);
 
 pthread_handler_t handle_slave_io(void *arg);
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 4b68f2a3821..f2aedf5d0aa 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -567,6 +567,13 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
                                                   transactional_table,
                                                   errcode);
       }
+
+      /*
+        Flushing the IO CACHE while writing the execute load query log event
+        may result in error (for instance, because the max_binlog_size has been 
+        reached, and rotation of the binary log failed).
+      */
+      error= error || mysql_bin_log.get_log_file()->error;
     }
     if (error)
       goto err;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index abfc2ec26b3..8225ea64fc0 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1474,7 +1474,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 #endif
   case COM_REFRESH:
   {
-    bool not_used;
+    int not_used;
     status_var_increment(thd->status_var.com_stat[SQLCOM_FLUSH]);
     ulong options= (ulong) (uchar) packet[0];
     if (check_global_access(thd,RELOAD_ACL))
@@ -3227,7 +3227,11 @@ end_with_restore_list:
       {
         Incident_log_event ev(thd, incident);
         (void) mysql_bin_log.write(&ev);        /* error is ignored */
-        mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
+        if (mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE))
+        {
+          res= 1;
+          break;
+        }
       }
       DBUG_PRINT("debug", ("Just after generate_incident()"));
     }
@@ -4051,7 +4055,7 @@ end_with_restore_list:
     lex->no_write_to_binlog= 1;
   case SQLCOM_FLUSH:
   {
-    bool write_to_binlog;
+    int write_to_binlog;
     if (check_global_access(thd,RELOAD_ACL))
       goto error;
 
@@ -4068,12 +4072,22 @@ end_with_restore_list:
       /*
         Presumably, RESET and binlog writing doesn't require synchronization
       */
-      if (!lex->no_write_to_binlog && write_to_binlog)
+
+      if (write_to_binlog > 0)  // we should write
+      { 
+        if (!lex->no_write_to_binlog)
+          res= write_bin_log(thd, FALSE, thd->query(), thd->query_length());
+      } else if (write_to_binlog < 0) 
       {
-        if ((res= write_bin_log(thd, FALSE, thd->query(), thd->query_length())))
-          break;
-      }
-      my_ok(thd);
+        /* 
+           We should not write, but rather report error because 
+           reload_acl_and_cache binlog interactions failed 
+         */
+        res= 1;
+      } 
+
+      if (!res)
+        my_ok(thd);
     } 
     
     break;
@@ -6855,7 +6869,10 @@ void add_join_natural(TABLE_LIST *a, TABLE_LIST *b, List<String> *using_fields,
   @param thd Thread handler (can be NULL!)
   @param options What should be reset/reloaded (tables, privileges, slave...)
   @param tables Tables to flush (if any)
-  @param write_to_binlog True if we can write to the binlog.
+  @param write_to_binlog < 0 if there was an error while interacting with the binary log inside
+                         reload_acl_and_cache, 
+                         0 if we should not write to the binary log, 
+                         > 0 if we can write to the binlog.
                
   @note Depending on 'options', it may be very bad to write the
     query to the binlog (e.g. FLUSH SLAVE); this is a
@@ -6869,11 +6886,11 @@ void add_join_natural(TABLE_LIST *a, TABLE_LIST *b, List<String> *using_fields,
 */
 
 bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
-                          bool *write_to_binlog)
+                          int *write_to_binlog)
 {
   bool result=0;
   select_errors=0;				/* Write if more errors */
-  bool tmp_write_to_binlog= 1;
+  int tmp_write_to_binlog= *write_to_binlog= 1;
 
   DBUG_ASSERT(!thd || !thd->in_sub_stmt);
 
@@ -6936,12 +6953,16 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
     tmp_write_to_binlog= 0;
     if( mysql_bin_log.is_open() )
     {
-      mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
+      if (mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE))
+        *write_to_binlog= -1;
     }
 #ifdef HAVE_REPLICATION
+    int rotate_error= 0;
     pthread_mutex_lock(&LOCK_active_mi);
-    rotate_relay_log(active_mi);
+    rotate_error= rotate_relay_log(active_mi);
     pthread_mutex_unlock(&LOCK_active_mi);
+    if (rotate_error)
+      *write_to_binlog= -1;
 #endif
 
     /* flush slow and general logs */
@@ -7052,7 +7073,8 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
 #endif
  if (options & REFRESH_USER_RESOURCES)
    reset_mqh((LEX_USER *) NULL, 0);             /* purecov: inspected */
- *write_to_binlog= tmp_write_to_binlog;
+ if (*write_to_binlog != -1)
+   *write_to_binlog= tmp_write_to_binlog;
  /*
    If the query was killed then this function must fail.
  */