diff --git a/client/client_priv.h b/client/client_priv.h
index ffdf0fc4b8f..54fed943313 100644
--- a/client/client_priv.h
+++ b/client/client_priv.h
@@ -100,6 +100,7 @@ enum options_client
   OPT_SKIP_ANNOTATE_ROWS_EVENTS,
   OPT_SSL_CRL, OPT_SSL_CRLPATH,
   OPT_PRINT_ROW_COUNT, OPT_PRINT_ROW_EVENT_POSITIONS,
+  OPT_SHUTDOWN_WAIT_FOR_SLAVES,
   OPT_MAX_CLIENT_OPTION /* should be always the last */
 };
 
diff --git a/client/mysqladmin.cc b/client/mysqladmin.cc
index 66678f3fd77..5c9f9cd05f3 100644
--- a/client/mysqladmin.cc
+++ b/client/mysqladmin.cc
@@ -40,7 +40,8 @@ ulonglong last_values[MAX_MYSQL_VAR+100];
 static int interval=0;
 static my_bool option_force=0,interrupted=0,new_line=0,
                opt_compress= 0, opt_local= 0, opt_relative= 0, opt_verbose= 0,
-               opt_vertical= 0, tty_password= 0, opt_nobeep;
+               opt_vertical= 0, tty_password= 0, opt_nobeep,
+               opt_shutdown_wait_for_slaves= 0;
 static my_bool debug_info_flag= 0, debug_check_flag= 0;
 static uint tcp_port = 0, option_wait = 0, option_silent=0, nr_iterations;
 static uint opt_count_iterations= 0, my_end_arg;
@@ -218,6 +219,11 @@ static struct my_option my_long_options[] =
   {"shutdown_timeout", OPT_SHUTDOWN_TIMEOUT, "", &opt_shutdown_timeout,
    &opt_shutdown_timeout, 0, GET_ULONG, REQUIRED_ARG,
    SHUTDOWN_DEF_TIMEOUT, 0, 3600*12, 0, 1, 0},
+  {"wait_for_all_slaves", OPT_SHUTDOWN_WAIT_FOR_SLAVES,
+   "Defers shutdown until after all binlogged events have been sent to "
+   "all connected slaves", &opt_shutdown_wait_for_slaves,
+   &opt_shutdown_wait_for_slaves, 0, GET_BOOL, NO_ARG, 0, 0, 0,
+   0, 0, 0},
   {"plugin_dir", OPT_PLUGIN_DIR, "Directory for client-side plugins.",
     &opt_plugin_dir, &opt_plugin_dir, 0,
    GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
@@ -693,7 +699,17 @@ static int execute_commands(MYSQL *mysql,int argc, char **argv)
 	  !stat(pidfile, &pidfile_status))
 	last_modified= pidfile_status.st_mtime;
 
-      if (mysql_shutdown(mysql, SHUTDOWN_DEFAULT))
+      if (opt_shutdown_wait_for_slaves)
+      {
+        sprintf(buff, "SHUTDOWN WAIT FOR ALL SLAVES");
+        if (mysql_query(mysql, buff))
+        {
+          my_printf_error(0, "%s failed; error: '%-.200s'",
+                          error_flags, buff, mysql_error(mysql));
+          return -1;
+        }
+      }
+      else if (mysql_shutdown(mysql, SHUTDOWN_DEFAULT))
       {
 	my_printf_error(0, "shutdown failed; error: '%s'", error_flags,
 			mysql_error(mysql));
diff --git a/mysql-test/mysql-test-run.pl b/mysql-test/mysql-test-run.pl
index 56e265d9eef..a68d2f89bea 100755
--- a/mysql-test/mysql-test-run.pl
+++ b/mysql-test/mysql-test-run.pl
@@ -4550,7 +4550,8 @@ sub extract_warning_lines ($$) {
      qr/InnoDB: See also */,
      qr/InnoDB: Cannot open .*ib_buffer_pool.* for reading: No such file or directory*/,
      qr/InnoDB: Table .*mysql.*innodb_table_stats.* not found./,
-     qr/InnoDB: User stopword table .* does not exist./
+     qr/InnoDB: User stopword table .* does not exist./,
+     qr/Dump thread [0-9]+ last sent to server [0-9]+ binlog file:pos .+/
 
     );
 
diff --git a/mysql-test/suite/rpl/include/rpl_shutdown_wait_slaves.inc b/mysql-test/suite/rpl/include/rpl_shutdown_wait_slaves.inc
new file mode 100644
index 00000000000..476b41033e8
--- /dev/null
+++ b/mysql-test/suite/rpl/include/rpl_shutdown_wait_slaves.inc
@@ -0,0 +1,83 @@
+--connection server_1
+
+CREATE TABLE t1 (a INT) ENGINE=innodb;
+
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+
+--connection server_3
+--sync_with_master
+
+--connection server_4
+--source include/stop_slave.inc
+
+--connection server_1
+--disable_query_log
+--let $count=1000
+while ($count)
+{
+ INSERT INTO t1 SET a=1;
+ --dec $count
+}
+--enable_query_log
+--save_master_pos
+
+# Shutdown master and restart server_4 who will be waiting for the master
+# to start replication at its shutdown beginning phase.
+# The being forked out server_4 dump thread must relate to a record
+# in slave_list, and it won't start sending out binlog events
+# until has received a signal from the shutdown thread.
+# This also proves delivery to a  started-in-middle-of-shutdown slave.
+--connection server_1
+SET @@GLOBAL.debug_dbug="+d,simulate_delay_at_shutdown";
+
+--connection server_4
+--source include/start_slave.inc
+
+--connection server_1
+--write_file $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+wait
+EOF
+# --shutdown_server 60
+--send SHUTDOWN WAIT FOR ALL SLAVES
+--reap
+--source include/wait_until_disconnected.inc
+#
+# MDEV-18450 liveness condition:
+# Despite shutdown even "late" slave #4 is in sync
+#
+--connection server_4
+--sync_with_master
+
+--connection server_3
+--sync_with_master
+
+--connection server_2
+--sync_with_master
+
+--connection server_1
+--append_file $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+restart
+EOF
+
+--connection default
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+--connection server_1
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+
+#
+# Cleanup
+#
+--connection server_1
+DROP TABLE t1;
+
+--connection server_2
+--source include/start_slave.inc
+--connection server_3
+--source include/start_slave.inc
+--connection server_4
+--source include/start_slave.inc
diff --git a/mysql-test/suite/rpl/r/rpl_shutdown_wait_semisync_slaves.result b/mysql-test/suite/rpl/r/rpl_shutdown_wait_semisync_slaves.result
new file mode 100644
index 00000000000..e310ba3326d
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_shutdown_wait_semisync_slaves.result
@@ -0,0 +1,58 @@
+include/rpl_init.inc [topology=1->2, 1->3, 1->4]
+connection server_1;
+call mtr.add_suppression("Timeout waiting for reply of binlog");
+SET @@GLOBAL.rpl_semi_sync_master_enabled = 1;
+connection server_2;
+set global rpl_semi_sync_slave_enabled = 1;
+include/stop_slave.inc
+include/start_slave.inc
+set global rpl_semi_sync_slave_enabled = 1;
+connection server_3;
+set global rpl_semi_sync_slave_enabled = 1;
+include/stop_slave.inc
+include/start_slave.inc
+set global rpl_semi_sync_slave_enabled = 1;
+connection server_1;
+CREATE TABLE t1 (a INT) ENGINE=innodb;
+connection server_2;
+connection server_3;
+connection server_4;
+include/stop_slave.inc
+connection server_1;
+connection server_1;
+SET @@GLOBAL.debug_dbug="+d,simulate_delay_at_shutdown";
+connection server_4;
+include/start_slave.inc
+connection server_1;
+SHUTDOWN WAIT FOR ALL SLAVES;
+connection server_4;
+connection server_3;
+connection server_2;
+connection server_1;
+connection default;
+connection server_1;
+connection server_1;
+DROP TABLE t1;
+connection server_2;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_3;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_4;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_2;
+include/stop_slave.inc
+include/start_slave.inc
+SET @@GLOBAL.rpl_semi_sync_slave_enabled = 0;;
+connection server_3;
+include/stop_slave.inc
+include/start_slave.inc
+SET @@GLOBAL.rpl_semi_sync_slave_enabled = 0;;
+connection server_1;
+SET @@GLOBAL.rpl_semi_sync_master_enabled = 0;
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/r/rpl_shutdown_wait_slaves.result b/mysql-test/suite/rpl/r/rpl_shutdown_wait_slaves.result
new file mode 100644
index 00000000000..0fcff8fa349
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_shutdown_wait_slaves.result
@@ -0,0 +1,35 @@
+include/rpl_init.inc [topology=1->2, 1->3, 1->4]
+connection server_1;
+CREATE TABLE t1 (a INT) ENGINE=innodb;
+connection server_2;
+connection server_3;
+connection server_4;
+include/stop_slave.inc
+connection server_1;
+connection server_1;
+SET @@GLOBAL.debug_dbug="+d,simulate_delay_at_shutdown";
+connection server_4;
+include/start_slave.inc
+connection server_1;
+SHUTDOWN WAIT FOR ALL SLAVES;
+connection server_4;
+connection server_3;
+connection server_2;
+connection server_1;
+connection default;
+connection server_1;
+connection server_1;
+DROP TABLE t1;
+connection server_2;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_3;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+connection server_4;
+include/start_slave.inc
+Warnings:
+Note	1254	Slave is already running
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.cnf b/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.cnf
new file mode 100644
index 00000000000..8ff9df0384d
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.cnf
@@ -0,0 +1,16 @@
+!include ../my.cnf
+
+[mysqld.1]
+log_warnings=3
+[mysqld.2]
+
+[mysqld.3]
+
+[mysqld.4]
+
+[ENV]
+SERVER_MYPORT_3=		@mysqld.3.port
+SERVER_MYSOCK_3=		@mysqld.3.socket
+
+SERVER_MYPORT_4=		@mysqld.4.port
+SERVER_MYSOCK_4=		@mysqld.4.socket
diff --git a/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.test b/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.test
new file mode 100644
index 00000000000..2c63df30fde
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_semisync_slaves.test
@@ -0,0 +1,46 @@
+#
+# MDEV-18450 "Slow" shutdown to wait for slaves that are to be fed
+# with everything in the master binlog before shutdown completes.
+#
+# This is a semisync version of basic tests.
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--let $rpl_topology=1->2, 1->3, 1->4
+--source include/rpl_init.inc
+
+--connection server_1
+call mtr.add_suppression("Timeout waiting for reply of binlog");
+--let $sav_enabled_master=`SELECT @@GLOBAL.rpl_semi_sync_master_enabled`
+SET @@GLOBAL.rpl_semi_sync_master_enabled = 1;
+
+--let slaves= 3
+--let i= 2
+while (`SELECT $i <= $slaves`)
+{
+  --connection server_$i
+  --let $sav_enabled_slave=`SELECT @@GLOBAL.rpl_semi_sync_slave_enabled`
+  set global rpl_semi_sync_slave_enabled = 1;
+
+  source include/stop_slave.inc;
+  source include/start_slave.inc;
+  set global rpl_semi_sync_slave_enabled = 1;
+
+  --inc $i
+}
+
+--source include/rpl_shutdown_wait_slaves.inc
+--let i= 2
+while (`SELECT $i <= $slaves`)
+{
+  --connection server_$i
+  source include/stop_slave.inc;
+  source include/start_slave.inc;
+  --eval SET @@GLOBAL.rpl_semi_sync_slave_enabled = $sav_enabled_slave;
+
+  --inc $i
+}
+
+--connection server_1
+--eval SET @@GLOBAL.rpl_semi_sync_master_enabled = $sav_enabled_master
+
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.cnf b/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.cnf
new file mode 100644
index 00000000000..8ff9df0384d
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.cnf
@@ -0,0 +1,16 @@
+!include ../my.cnf
+
+[mysqld.1]
+log_warnings=3
+[mysqld.2]
+
+[mysqld.3]
+
+[mysqld.4]
+
+[ENV]
+SERVER_MYPORT_3=		@mysqld.3.port
+SERVER_MYSOCK_3=		@mysqld.3.socket
+
+SERVER_MYPORT_4=		@mysqld.4.port
+SERVER_MYSOCK_4=		@mysqld.4.socket
diff --git a/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.test b/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.test
new file mode 100644
index 00000000000..97363206776
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_shutdown_wait_slaves.test
@@ -0,0 +1,11 @@
+#
+# MDEV-18450 "Slow" shutdown to wait for slaves that are to be fed
+# with everything in the master binlog before shutdown completes.
+#
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--let $rpl_topology=1->2, 1->3, 1->4
+--source include/rpl_init.inc
+
+--source include/rpl_shutdown_wait_slaves.inc
+--source include/rpl_end.inc
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index bf1780ed082..fa9387ac51a 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -466,6 +466,7 @@ uint protocol_version;
 uint lower_case_table_names;
 ulong tc_heuristic_recover= 0;
 Atomic_counter<uint32_t> thread_count;
+bool shutdown_wait_for_slaves;
 int32 slave_open_temp_tables;
 ulong thread_created;
 ulong back_log, connect_timeout, concurrency, server_id;
@@ -1519,19 +1520,9 @@ static void end_ssl();
 ** Code to end mysqld
 ****************************************************************************/
 
-static my_bool kill_all_threads(THD *thd, void *)
+/* common callee of two shutdown phases */
+static void kill_thread(THD *thd)
 {
-  DBUG_PRINT("quit", ("Informing thread %ld that it's time to die",
-                      (ulong) thd->thread_id));
-  /* We skip slave threads on this first loop through. */
-  if (thd->slave_thread)
-    return 0;
-
-  if (DBUG_EVALUATE_IF("only_kill_system_threads", !thd->system_thread, 0))
-    return 0;
-
-  thd->set_killed(KILL_SERVER_HARD);
-  MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
   if (WSREP(thd)) mysql_mutex_lock(&thd->LOCK_thd_data);
   mysql_mutex_lock(&thd->LOCK_thd_kill);
   if (thd->mysys_var)
@@ -1557,16 +1548,73 @@ static my_bool kill_all_threads(THD *thd, void *)
   }
   mysql_mutex_unlock(&thd->LOCK_thd_kill);
   if (WSREP(thd)) mysql_mutex_unlock(&thd->LOCK_thd_data);
+}
+
+
+/**
+  First shutdown everything but slave threads and binlog dump connections
+*/
+static my_bool kill_thread_phase_1(THD *thd, void *)
+{
+  DBUG_PRINT("quit", ("Informing thread %ld that it's time to die",
+                      (ulong) thd->thread_id));
+  if (thd->slave_thread || thd->is_binlog_dump_thread())
+    return 0;
+
+  if (DBUG_EVALUATE_IF("only_kill_system_threads", !thd->system_thread, 0))
+    return 0;
+
+  thd->set_killed(KILL_SERVER_HARD);
+  MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
+  kill_thread(thd);
   return 0;
 }
 
 
-static my_bool warn_threads_still_active(THD *thd, void *)
+/**
+  Last shutdown binlog dump connections
+*/
+static my_bool kill_thread_phase_2(THD *thd, void *)
 {
-  sql_print_warning("%s: Thread %llu (user : '%s') did not exit\n", my_progname,
-                    (ulonglong) thd->thread_id,
-                    (thd->main_security_ctx.user ?
-                     thd->main_security_ctx.user : ""));
+  if (shutdown_wait_for_slaves)
+  {
+    thd->set_killed(KILL_SERVER);
+  }
+  else
+  {
+    thd->set_killed(KILL_SERVER_HARD);
+    MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (thd));
+  }
+  kill_thread(thd);
+  return 0;
+}
+
+
+/* associated with the kill thread phase 1 */
+static my_bool warn_threads_active_after_phase_1(THD *thd, void *)
+{
+  if (!thd->is_binlog_dump_thread())
+    sql_print_warning("%s: Thread %llu (user : '%s') did not exit\n", my_progname,
+                      (ulonglong) thd->thread_id,
+                      (thd->main_security_ctx.user ?
+                       thd->main_security_ctx.user : ""));
+  return 0;
+}
+
+
+/* associated with the kill thread phase 2 */
+static my_bool warn_threads_active_after_phase_2(THD *thd, void *)
+{
+  mysql_mutex_lock(&thd->LOCK_thd_data);
+  // dump thread may not have yet (or already) current_linfo set
+  sql_print_warning("Dump thread %llu last sent to server %lu "
+                    "binlog file:pos %s:%llu",
+                    thd->thread_id, thd->variables.server_id,
+                    thd->current_linfo ?
+                    my_basename(thd->current_linfo->log_file_name) : "NULL",
+                    thd->current_linfo ? thd->current_linfo->pos : 0);
+  mysql_mutex_unlock(&thd->LOCK_thd_data);
+
   return 0;
 }
 
@@ -1650,6 +1698,21 @@ void kill_mysql(THD *thd)
   {
     my_free(user);
   }
+
+  DBUG_EXECUTE_IF("mysql_admin_shutdown_wait_for_slaves",
+                  thd->lex->is_shutdown_wait_for_slaves= true;);
+  DBUG_EXECUTE_IF("simulate_delay_at_shutdown",
+                  {
+                    DBUG_ASSERT(binlog_dump_thread_count == 3);
+                    const char act[]=
+                      "now "
+                      "SIGNAL greetings_from_kill_mysql";
+                    DBUG_ASSERT(!debug_sync_set_action(thd,
+                                                       STRING_WITH_LEN(act)));
+                  };);
+
+  if (thd->lex->is_shutdown_wait_for_slaves)
+    shutdown_wait_for_slaves= true;
   break_connect_loop();
 }
 
@@ -1693,7 +1756,7 @@ static void close_connections(void)
     This will give the threads some time to gracefully abort their
     statements and inform their clients that the server is about to die.
   */
-  server_threads.iterate(kill_all_threads);
+  server_threads.iterate(kill_thread_phase_1);
 
   Events::deinit();
   slave_prepare_for_shutdown();
@@ -1716,11 +1779,11 @@ static void close_connections(void)
   */
   DBUG_PRINT("info", ("thread_count: %u", uint32_t(thread_count)));
 
-  for (int i= 0; thread_count && i < 1000; i++)
+  for (int i= 0; (thread_count - binlog_dump_thread_count) && i < 1000; i++)
     my_sleep(20000);
 
   if (global_system_variables.log_warnings)
-    server_threads.iterate(warn_threads_still_active);
+    server_threads.iterate(warn_threads_active_after_phase_1);
 
 #ifdef WITH_WSREP
   if (wsrep_inited == 1)
@@ -1732,9 +1795,23 @@ static void close_connections(void)
   DBUG_PRINT("quit", ("Waiting for threads to die (count=%u)",
                       uint32_t(thread_count)));
 
-  while (thread_count)
+  while (thread_count - binlog_dump_thread_count)
     my_sleep(1000);
 
+  /* Kill phase 2 */
+  server_threads.iterate(kill_thread_phase_2);
+  for (uint64 i= 0; thread_count; i++)
+  {
+    /*
+      This time the warnings are emitted within the loop to provide a
+      dynamic view on the shutdown status through the errorlog.
+    */
+    if (global_system_variables.log_warnings > 2 && i % 60000 == 0)
+      server_threads.iterate(warn_threads_active_after_phase_2);
+    my_sleep(1000);
+  }
+  /* End of kill phase 2 */
+
   DBUG_PRINT("quit",("close_connections thread"));
   DBUG_VOID_RETURN;
 }
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index f13ff6c9163..e7873b185c5 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -147,11 +147,11 @@ int THD::register_slave(uchar *packet, size_t packet_length)
   if (!(si->master_id= uint4korr(p)))
     si->master_id= global_system_variables.server_id;
 
-  binlog_dump_thread_count++;
   unregister_slave();
   mysql_mutex_lock(&LOCK_thd_data);
   slave_info= si;
   mysql_mutex_unlock(&LOCK_thd_data);
+  binlog_dump_thread_count++;
   return 0;
 
 err:
@@ -161,6 +161,16 @@ err:
 }
 
 
+bool THD::is_binlog_dump_thread()
+{
+  mysql_mutex_lock(&LOCK_thd_data);
+  bool res= slave_info != NULL;
+  mysql_mutex_unlock(&LOCK_thd_data);
+
+  return res;
+}
+
+
 static my_bool show_slave_hosts_callback(THD *thd, Protocol *protocol)
 {
   my_bool res= FALSE;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index bbba5218989..e2bc6ef1d90 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -644,6 +644,7 @@ THD::THD(my_thread_id id, bool is_wsrep_applier, bool skip_global_sys_var_lock)
    m_tmp_tables_locked(false)
 #ifdef HAVE_REPLICATION
    ,
+   current_linfo(0),
    slave_info(0)
 #endif
 #ifdef WITH_WSREP
@@ -751,7 +752,6 @@ THD::THD(my_thread_id id, bool is_wsrep_applier, bool skip_global_sys_var_lock)
   progress.arena= 0;
   progress.report_to_client= 0;
   progress.max_counter= 0;
-  current_linfo =  0;
   slave_thread = 0;
   connection_name.str= 0;
   connection_name.length= 0;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 23c51067734..96f9e6643fe 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3194,12 +3194,6 @@ public:
   /** number of name_const() substitutions, see sp_head.cc:subst_spvars() */
   uint       query_name_consts;
 
-  /*
-    If we do a purge of binary logs, log index info of the threads
-    that are currently reading it needs to be adjusted. To do that
-    each thread that is using LOG_INFO needs to adjust the pointer to it
-  */
-  LOG_INFO*  current_linfo;
   NET*       slave_net;			// network connection from slave -> m.
 
   /*
@@ -4813,10 +4807,20 @@ private:
 
 public:
 #ifdef HAVE_REPLICATION
+  /*
+    If we do a purge of binary logs, log index info of the threads
+    that are currently reading it needs to be adjusted. To do that
+    each thread that is using LOG_INFO needs to adjust the pointer to it
+  */
+  LOG_INFO *current_linfo;
   Slave_info *slave_info;
 
+  void set_current_linfo(LOG_INFO *linfo);
+  void reset_current_linfo() { set_current_linfo(0); }
+
   int register_slave(uchar *packet, size_t packet_length);
   void unregister_slave();
+  bool is_binlog_dump_thread();
 #endif
 
   inline ulong wsrep_binlog_format() const
@@ -4975,18 +4979,6 @@ public:
        (THD_TRANS::DID_WAIT | THD_TRANS::CREATED_TEMP_TABLE |
         THD_TRANS::DROPPED_TEMP_TABLE | THD_TRANS::DID_DDL));
   }
-  /*
-    Reset current_linfo
-    Setting current_linfo to 0 needs to be done with LOCK_thd_data to
-    ensure that adjust_linfo_offsets doesn't use a structure that may
-    be deleted.
-  */
-  inline void reset_current_linfo()
-  {
-    mysql_mutex_lock(&LOCK_thd_data);
-    current_linfo= 0;
-    mysql_mutex_unlock(&LOCK_thd_data);
-  }
 
 
   uint get_net_wait_timeout()
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 61816bee803..8e983551dbe 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -3226,6 +3226,7 @@ public:
   /* The following is used by KILL */
   killed_state kill_signal;
   killed_type  kill_type;
+  bool is_shutdown_wait_for_slaves;
   /*
     This variable is used in post-parse stage to declare that sum-functions,
     or functions which have sense only if GROUP BY is present, are allowed.
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 5a520c0df02..91f43fdbaa6 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2168,6 +2168,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     */
     enum mysql_enum_shutdown_level level;
     level= (enum mysql_enum_shutdown_level) (uchar) packet[0];
+    thd->lex->is_shutdown_wait_for_slaves= false;  // "deferred" cleanup
     if (level == SHUTDOWN_DEFAULT)
       level= SHUTDOWN_WAIT_ALL_BUFFERS; // soon default will be configurable
     else if (level != SHUTDOWN_WAIT_ALL_BUFFERS)
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 7fc3bb5926d..4c7a768b9ce 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -506,6 +506,22 @@ static enum enum_binlog_checksum_alg get_binlog_checksum_value_at_connect(THD *
   DBUG_RETURN(ret);
 }
 
+
+/**
+  Set current_linfo
+
+  Setting current_linfo needs to be done with LOCK_thd_data to ensure that
+  adjust_linfo_offsets doesn't use a structure that may be deleted.
+*/
+
+void THD::set_current_linfo(LOG_INFO *linfo)
+{
+  mysql_mutex_lock(&LOCK_thd_data);
+  current_linfo= linfo;
+  mysql_mutex_unlock(&LOCK_thd_data);
+}
+
+
 /*
   Adjust the position pointer in the binary log file for all running slaves
 
@@ -2125,9 +2141,8 @@ static int init_binlog_sender(binlog_send_info *info,
 
   // set current pos too
   linfo->pos= *pos;
-
   // note: publish that we use file, before we open it
-  thd->current_linfo= linfo;
+  thd->set_current_linfo(linfo);
 
   if (check_start_offset(info, linfo->log_file_name, *pos))
     return 1;
@@ -2365,14 +2380,15 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
   DBUG_RETURN(0);
 }
 
-static bool should_stop(binlog_send_info *info)
+static bool should_stop(binlog_send_info *info, bool kill_server_check= false)
 {
   return
-      info->net->error ||
-      info->net->vio == NULL ||
-      info->thd->killed ||
-      info->error != 0 ||
-      info->should_stop;
+    info->net->error ||
+    info->net->vio == NULL ||
+    (info->thd->killed &&
+     (info->thd->killed != KILL_SERVER || kill_server_check)) ||
+    info->error != 0 ||
+    info->should_stop;
 }
 
 /**
@@ -2393,7 +2409,7 @@ static int wait_new_events(binlog_send_info *info,         /* in */
                         &stage_master_has_sent_all_binlog_to_slave,
                         &old_stage);
 
-  while (!should_stop(info))
+  while (!should_stop(info, true))
   {
     *end_pos_ptr= mysql_bin_log.get_binlog_end_pos(binlog_end_pos_filename);
     if (strcmp(linfo->log_file_name, binlog_end_pos_filename) != 0)
@@ -2745,6 +2761,14 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
     info->error= ER_UNKNOWN_ERROR;
     goto err;
   }
+  DBUG_EXECUTE_IF("simulate_delay_at_shutdown",
+                 {
+                   const char act[]=
+                     "now "
+                     "WAIT_FOR greetings_from_kill_mysql";
+                   DBUG_ASSERT(!debug_sync_set_action(thd,
+                                                      STRING_WITH_LEN(act)));
+                 };);
 
   /*
     heartbeat_period from @master_heartbeat_period user variable
@@ -3952,7 +3976,7 @@ bool mysql_show_binlog_events(THD* thd)
       goto err;
     }
 
-    thd->current_linfo= &linfo;
+    thd->set_current_linfo(&linfo);
 
     if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0)
       goto err;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 970f5d50334..5c911928a9a 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -14717,6 +14717,15 @@ kill_expr:
 
 shutdown:
         SHUTDOWN { Lex->sql_command= SQLCOM_SHUTDOWN; }
+        shutdown_option {}
+        ;
+
+shutdown_option:
+        /*  Empty */    { Lex->is_shutdown_wait_for_slaves= false; }
+        | WAIT_SYM FOR_SYM ALL SLAVES
+        {
+          Lex->is_shutdown_wait_for_slaves= true;
+        }
         ;
 
 /* change database */
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index e709e7e3afa..b6ea6ba9970 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -14862,8 +14862,16 @@ kill_expr:
 
 shutdown:
         SHUTDOWN { Lex->sql_command= SQLCOM_SHUTDOWN; }
+        shutdown_option {}
         ;
 
+shutdown_option:
+        /*  Empty */    { Lex->is_shutdown_wait_for_slaves= false; }
+        | WAIT_SYM FOR_SYM ALL SLAVES
+        {
+          Lex->is_shutdown_wait_for_slaves= true;
+        }
+        ;
 /* change database */
 
 use: