BUG#46166: MYSQL_BIN_LOG::new_file_impl is not propagating error

           when generating new name.
      
If find_uniq_filename returns an error, then this error is not
being propagated upwards, and execution does not report error to
the user (although a entry in the error log is generated).
                  
Additionally, some more errors were ignored in new_file_impl:
- when writing the rotate event
- when reopening the index and binary log file
                  
This patch addresses this by propagating the error up in the
execution stack. Furthermore, when rotation of the binary log
fails, an incident event is written, because there may be a
chance that some changes for a given statement, were not properly
logged. For example, in SBR, LOAD DATA INFILE statement requires
more than one event to be logged, should rotation fail while
logging part of the LOAD DATA events, then the logged data would
become inconsistent with the data in the storage engine.

17 files changed, 958 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 1525ca53bca..18381bc552c 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..46dcec2e2cd 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()
@@ -4759,7 +4840,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 +4952,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 +5758,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 +5776,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 +6018,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 +6028,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 709a49b2036..526133c20c6 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1063,7 +1063,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 7cf64134d70..c81bd36cfad 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;
@@ -6862,7 +6876,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
@@ -6876,11 +6893,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);
 
@@ -6943,12 +6960,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 */
@@ -7059,7 +7080,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.
  */