Cherrypick from 5.6 : Bug#20009154 - FIX REGRESSION AFTER HA_FLUSH_LOGS() HOOK FOR NDB IS REMOVED.

Author: Maitrayi Sabaratnam

mysql-test/suite/ndb_binlog/r/ndb_binlog_flush_log_waits.result

@@ -0,0 +1,27 @@
+SET @default_max_binlog_size = @@global.max_binlog_size;
+set global max_binlog_size=1048576;
+create table test.tn (b varchar(1000)) engine=ndb;
+create procedure create_load(loops int)
+begin
+set @x = 0;
+repeat
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+insert into tn values (repeat('C', 100));
+delete from tn;
+set @x = @x + 1;
+until @x = loops
+end repeat;
+end%
+call create_load(20);
+FLUSH LOGS;
+drop procedure create_load;
+drop table tn;
+set global max_binlog_size= @default_max_binlog_size;

mysql-test/suite/ndb_binlog/t/ndb_binlog_flush_log_waits.test

@@ -0,0 +1,68 @@
+-- source include/have_ndb.inc
+-- source include/have_binlog_format_mixed_or_row.inc
+
+#
+# Test whether FLUSH LOGS flushes upto
+# session_last_committed_epoch to Binlog.
+#
+
+# Set Binlog size large enough to avoid size-based rotate
+SET @default_max_binlog_size = @@global.max_binlog_size;
+set global max_binlog_size=1048576; #1MB
+
+create table test.tn (b varchar(1000)) engine=ndb;
+
+# Method to generate load on NDB
+delimiter %;
+create procedure create_load(loops int)
+begin
+  set @x = 0;
+  repeat
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    insert into tn values (repeat('C', 100));
+    delete from tn;
+    set @x = @x + 1;
+  until @x = loops
+  end repeat;
+end%
+delimiter ;%
+
+# Generate much load on NDB in order to make Binlog to lag behind
+--disable_result_log
+call create_load(20);
+--enable_result_log
+
+# Get the last committed epoch of the session
+--source ndb_binlog_init_epoch_vals.inc
+let $session_epoch= query_get_value(select @init_session_epoch as e, e, 1);
+
+# Flush Binlog
+FLUSH LOGS;
+
+# Get the maximum epoch written to Binlog
+let $max_epoch_after_flush = `select max(epoch) as epoch from
+mysql.ndb_binlog_index`;
+
+if ($max_epoch_after_flush < $session_epoch)
+{
+  --echo
+  --echo Max epoch found in Binlog after FLUSH LOGS ($max_epoch_after_flush)
+  --echo is less than the last committed epoch of the session
+($session_epoch)
+  --echo
+  die "FLUSH LOGS failed to flush upto the last committed epoch of the
+session";
+}
+
+drop procedure create_load;
+drop table tn;
+
+set global max_binlog_size= @default_max_binlog_size;

sql/binlog.cc

@@ -5484,12 +5484,19 @@ void MYSQL_BIN_LOG::purge()
   @retval
     nonzero - error in rotating routine.
 */
-int MYSQL_BIN_LOG::rotate_and_purge(bool force_rotate)
+int MYSQL_BIN_LOG::rotate_and_purge(THD* thd, bool force_rotate)
 {
   int error= 0;
   DBUG_ENTER("MYSQL_BIN_LOG::rotate_and_purge");
   bool check_purge= false;
 
+  /*
+    Wait for handlerton to insert any pending information into the binlog.
+    For e.g. ha_ndbcluster which updates the binlog asynchronously this is
+    needed so that the user see its own commands in the binlog.
+  */
+  ha_binlog_wait(thd);
+
   DBUG_ASSERT(!is_relay_log);
   mysql_mutex_lock(&LOCK_log);
   error= rotate(force_rotate, &check_purge);

sql/binlog.h

@@ -619,7 +619,7 @@ class MYSQL_BIN_LOG: public TC_LOG, private MYSQL_LOG
   int remove_logs_from_index(LOG_INFO* linfo, bool need_update_threads);
   int rotate(bool force_rotate, bool* check_purge);
   void purge();
-  int rotate_and_purge(bool force_rotate);
+  int rotate_and_purge(THD* thd, bool force_rotate);
   /**
      Flush binlog cache and synchronize to disk.
 

sql/sql_reload.cc

@@ -163,7 +163,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long options,
       tmp_write_to_binlog= 0;
       if (mysql_bin_log.is_open())
       {
-        if (mysql_bin_log.rotate_and_purge(true))
+        if (mysql_bin_log.rotate_and_purge(thd, true))
           *write_to_binlog= -1;
       }
     }

sql/sys_vars.h

@@ -2198,7 +2198,7 @@ class Sys_var_gtid_purged : public sys_var
                           previous_gtid_logged, current_gtid_logged);
 
     // Rotate logs to have Previous_gtid_event on last binlog.
-    rotate_res= mysql_bin_log.rotate_and_purge(true);
+    rotate_res= mysql_bin_log.rotate_and_purge(thd, true);
     if (rotate_res)
     {
       error= true;