MDEV-7591 master crashed when slave specfied a future position with semi-repl plugin

cherry-pick the upstream fix

commit d4ba10184cd7bde9c31c610e664ecd0c93605c46
Author: Sujatha Sivakumar <sujatha.sivakumar@oracle.com>
Date:   Wed Jul 2 11:34:11 2014 +0530

    Bug#17453826:ASSERTION ERROR WHEN SETTING FUTURE BINLOG
    FILE/POS WITH SEMISYNC

    Problem:
    ========
    When DMLs are in progress on the master stopping a slave and
    setting ahead binlog name/pos will cause an assert on the
    master.
    ...

mysql-test/suite/rpl/r/semisync_future-7591.result

+include/master-slave.inc
+[connection master]
+call mtr.add_suppression("Timeout waiting for reply of binlog*");
+create table t1 (i int);
+set global rpl_semi_sync_master_enabled = ON;
+include/stop_slave.inc
+set global rpl_semi_sync_slave_enabled = ON;
+change master to master_log_file='master-bin.000002', master_log_pos = 320;
+start slave;
+include/wait_for_slave_io_error.inc [errno=1236]
+insert into t1 values (1);
+reset master;
+include/stop_slave.inc
+reset slave;
+include/start_slave.inc
+set global rpl_semi_sync_slave_enabled = OFF;
+drop table t1;
+set global rpl_semi_sync_master_enabled = OFF;
+include/rpl_end.inc

mysql-test/suite/rpl/t/semisync_future-7591.test

+--source include/have_semisync.inc
+--source include/master-slave.inc
+
+call mtr.add_suppression("Timeout waiting for reply of binlog*");
+create table t1 (i int);
+
+set global rpl_semi_sync_master_enabled = ON;
+
+--connection slave
+--source include/stop_slave.inc
+set global rpl_semi_sync_slave_enabled = ON;
+change master to master_log_file='master-bin.000002', master_log_pos = 320;
+
+start slave;
+--let $slave_io_errno=1236
+--source include/wait_for_slave_io_error.inc
+
+--connection master
+insert into t1 values (1);
+reset master;
+
+--connection slave
+--source include/stop_slave.inc
+reset slave;
+--source include/start_slave.inc
+
+set global rpl_semi_sync_slave_enabled = OFF;
+--connection master
+drop table t1;
+set global rpl_semi_sync_master_enabled = OFF;
+--source include/rpl_end.inc

sql/sql_repl.cc

@@ -1948,6 +1948,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
   slave_connection_state until_gtid_state_obj;
   rpl_gtid error_gtid;
   binlog_send_info info(thd, packet, flags, log_file_name);
+  bool has_transmit_started= false;
 
   int old_max_allowed_packet= thd->variables.max_allowed_packet;
 
@@ -2007,16 +2008,6 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
                     DBUG_SET("+d,corrupt_read_log_event2");
                   });
 
-  if (global_system_variables.log_warnings > 1)
-    sql_print_information("Start binlog_dump to slave_server(%lu), pos(%s, %lu)",
-                          thd->variables.server_id, log_ident, (ulong)pos);
-  if (RUN_HOOK(binlog_transmit, transmit_start, (thd, flags, log_ident, pos)))
-  {
-    errmsg= "Failed to run hook 'transmit_start'";
-    my_errno= ER_UNKNOWN_ERROR;
-    goto err;
-  }
-
 #ifndef DBUG_OFF
   if (opt_sporadic_binlog_dump_fail && (binlog_dump_count++ % 2))
   {
@@ -2113,6 +2104,17 @@ impossible position";
     goto err;
   }
 
+  if (global_system_variables.log_warnings > 1)
+    sql_print_information("Start binlog_dump to slave_server(%lu), pos(%s, %lu)",
+                          thd->variables.server_id, log_ident, (ulong)pos);
+  if (RUN_HOOK(binlog_transmit, transmit_start, (thd, flags, log_ident, pos)))
+  {
+    errmsg= "Failed to run hook 'transmit_start'";
+    my_errno= ER_UNKNOWN_ERROR;
+    goto err;
+  }
+  has_transmit_started= true;
+
   /* reset transmit packet for the fake rotate event below */
   if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
     goto err;
@@ -2680,7 +2682,8 @@ impossible position";
   end_io_cache(&log);
   mysql_file_close(file, MYF(MY_WME));
 
-  RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags));
+  if (has_transmit_started)
+    RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags));
   my_eof(thd);
   THD_STAGE_INFO(thd, stage_waiting_to_finalize_termination);
   mysql_mutex_lock(&LOCK_thread_count);
@@ -2749,7 +2752,8 @@ impossible position";
   else
     strcpy(error_text, errmsg);
   end_io_cache(&log);
-  RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags));
+  if (has_transmit_started)
+    RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags));
   /*
     Exclude  iteration through thread list
     this is needed for purge_logs() - it will iterate through