int start_slave(THD* thd , Master_info* mi,  bool net_report)
{
  int slave_errno= 0;
  int thread_mask;
  DBUG_ENTER("start_slave");

  if (check_access(thd, SUPER_ACL, any_db, NULL, NULL, 0, 0))
    DBUG_RETURN(1);

  if (thd->lex->slave_connection.user ||
      thd->lex->slave_connection.password)
  {
#if defined(HAVE_OPENSSL) && !defined(EMBEDDED_LIBRARY)
    if (thd->vio_ok() && !thd->net.vio->ssl_arg)
      push_warning(thd, Sql_condition::SL_NOTE,
                   ER_INSECURE_PLAIN_TEXT,
                   ER(ER_INSECURE_PLAIN_TEXT));
#endif
#if !defined(HAVE_OPENSSL) && !defined(EMBEDDED_LIBRARY)
    push_warning(thd, Sql_condition::SL_NOTE,
                 ER_INSECURE_PLAIN_TEXT,
                 ER(ER_INSECURE_PLAIN_TEXT));
#endif
  }

  lock_slave_threads(mi);  // this allows us to cleanly read slave_running
  // Get a mask of _stopped_ threads
  init_thread_mask(&thread_mask,mi,1 /* inverse */);
  /*
    Below we will start all stopped threads.  But if the user wants to
    start only one thread, do as if the other thread was running (as we
    don't wan't to touch the other thread), so set the bit to 0 for the
    other thread
  */
  if (thd->lex->slave_thd_opt)
    thread_mask&= thd->lex->slave_thd_opt;
  if (thread_mask) //some threads are stopped, start them
  {
    if (global_init_info(mi, false, thread_mask))
      slave_errno=ER_MASTER_INFO;
    else if (server_id_supplied && *mi->host)
    {
      /*
        If we will start IO thread we need to take care of possible
        options provided through the START SLAVE if there is any.
      */
      if (thread_mask & SLAVE_IO)
      {
        if (thd->lex->slave_connection.user)
        {
          mi->set_start_user_configured(true);
          mi->set_user(thd->lex->slave_connection.user);
        }
        if (thd->lex->slave_connection.password)
        {
          mi->set_start_user_configured(true);
          mi->set_password(thd->lex->slave_connection.password,
                           strlen(thd->lex->slave_connection.password));
        }
        if (thd->lex->slave_connection.plugin_auth)
          mi->set_plugin_auth(thd->lex->slave_connection.plugin_auth);
        if (thd->lex->slave_connection.plugin_dir)
          mi->set_plugin_dir(thd->lex->slave_connection.plugin_dir);
      }
 
      /*
        If we will start SQL thread we will care about UNTIL options If
        not and they are specified we will ignore them and warn user
        about this fact.
      */
      if (thread_mask & SLAVE_SQL)
      {
        /*
          To cache the MTS system var values and used them in the following
          runtime. The system var:s can change meanwhile but having no other
          effects.
        */
        mi->rli->opt_slave_parallel_workers= opt_mts_slave_parallel_workers;
#ifndef DBUG_OFF
        if (!DBUG_EVALUATE_IF("check_slave_debug_group", 1, 0))
#endif
          mi->rli->checkpoint_group= opt_mts_checkpoint_group;

        mysql_mutex_lock(&mi->rli->data_lock);

        if (thd->lex->mi.pos)
        {
          if (thd->lex->mi.relay_log_pos)
            slave_errno= ER_BAD_SLAVE_UNTIL_COND;
          mi->rli->until_condition= Relay_log_info::UNTIL_MASTER_POS;
          mi->rli->until_log_pos= thd->lex->mi.pos;
          /*
             We don't check thd->lex->mi.log_file_name for NULL here
             since it is checked in sql_yacc.yy
          */
          strmake(mi->rli->until_log_name, thd->lex->mi.log_file_name,
                  sizeof(mi->rli->until_log_name)-1);
        }
        else if (thd->lex->mi.relay_log_pos)
        {
          if (thd->lex->mi.pos)
            slave_errno= ER_BAD_SLAVE_UNTIL_COND;
          mi->rli->until_condition= Relay_log_info::UNTIL_RELAY_POS;
          mi->rli->until_log_pos= thd->lex->mi.relay_log_pos;
          strmake(mi->rli->until_log_name, thd->lex->mi.relay_log_name,
                  sizeof(mi->rli->until_log_name)-1);
        }
        else if (thd->lex->mi.gtid)
        {
          global_sid_lock->wrlock();
          mi->rli->clear_until_condition();
          if (mi->rli->until_sql_gtids.add_gtid_text(thd->lex->mi.gtid)
              != RETURN_STATUS_OK)
            slave_errno= ER_BAD_SLAVE_UNTIL_COND;
          else {
            mi->rli->until_condition=
              LEX_MASTER_INFO::UNTIL_SQL_BEFORE_GTIDS == thd->lex->mi.gtid_until_condition
              ? Relay_log_info::UNTIL_SQL_BEFORE_GTIDS
              : Relay_log_info::UNTIL_SQL_AFTER_GTIDS;
            if ((mi->rli->until_condition ==
               Relay_log_info::UNTIL_SQL_AFTER_GTIDS) &&
               mi->rli->opt_slave_parallel_workers != 0)
            {
              mi->rli->opt_slave_parallel_workers= 0;
              push_warning_printf(thd, Sql_condition::SL_NOTE,
                                  ER_MTS_FEATURE_IS_NOT_SUPPORTED,
                                  ER(ER_MTS_FEATURE_IS_NOT_SUPPORTED),
                                  "UNTIL condtion",
                                  "Slave is started in the sequential execution mode.");
            }
          }
          global_sid_lock->unlock();
        }
        else if (thd->lex->mi.until_after_gaps)
        {
            mi->rli->until_condition= Relay_log_info::UNTIL_SQL_AFTER_MTS_GAPS;
            mi->rli->opt_slave_parallel_workers=
              mi->rli->recovery_parallel_workers;
        }
        else
          mi->rli->clear_until_condition();

        if (mi->rli->until_condition == Relay_log_info::UNTIL_MASTER_POS ||
            mi->rli->until_condition == Relay_log_info::UNTIL_RELAY_POS)
        {
          /* Preparing members for effective until condition checking */
          const char *p= fn_ext(mi->rli->until_log_name);
          char *p_end;
          if (*p)
          {
            //p points to '.'
            mi->rli->until_log_name_extension= strtoul(++p,&p_end, 10);
            /*
              p_end points to the first invalid character. If it equals
              to p, no digits were found, error. If it contains '\0' it
              means  conversion went ok.
            */
            if (p_end==p || *p_end)
              slave_errno=ER_BAD_SLAVE_UNTIL_COND;
          }
          else
            slave_errno=ER_BAD_SLAVE_UNTIL_COND;

          /* mark the cached result of the UNTIL comparison as "undefined" */
          mi->rli->until_log_names_cmp_result=
            Relay_log_info::UNTIL_LOG_NAMES_CMP_UNKNOWN;

          /* Issuing warning then started without --skip-slave-start */
          if (!opt_skip_slave_start)
            push_warning(thd, Sql_condition::SL_NOTE,
                         ER_MISSING_SKIP_SLAVE,
                         ER(ER_MISSING_SKIP_SLAVE));
          if (mi->rli->opt_slave_parallel_workers != 0)
          {
            mi->rli->opt_slave_parallel_workers= 0;
            push_warning_printf(thd, Sql_condition::SL_NOTE,
                                ER_MTS_FEATURE_IS_NOT_SUPPORTED,
                                ER(ER_MTS_FEATURE_IS_NOT_SUPPORTED),
                                "UNTIL condtion",
                                "Slave is started in the sequential execution mode.");
          }
        }

        mysql_mutex_unlock(&mi->rli->data_lock);

        /* MTS technical limitation no support of trans retry */
        if (mi->rli->opt_slave_parallel_workers != 0 && slave_trans_retries != 0)
        {
          push_warning_printf(thd, Sql_condition::SL_NOTE,
                              ER_MTS_FEATURE_IS_NOT_SUPPORTED,
                              ER(ER_MTS_FEATURE_IS_NOT_SUPPORTED),
                              "slave_transaction_retries",
                              "In the event of a transient failure, the slave will "
                              "not retry the transaction and will stop.");
        }
      }
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos || thd->lex->mi.gtid)
        push_warning(thd, Sql_condition::SL_NOTE, ER_UNTIL_COND_IGNORED,
                     ER(ER_UNTIL_COND_IGNORED));

      if (!slave_errno)
        slave_errno = start_slave_threads(false/*need_lock_slave=false*/,
                                          true/*wait_for_start=true*/,
                                          mi,
                                          thread_mask);
    }
    else
      slave_errno = ER_BAD_SLAVE;
  }
  else
  {
    /* no error if all threads are already started, only a warning */
    push_warning(thd, Sql_condition::SL_NOTE, ER_SLAVE_WAS_RUNNING,
                 ER(ER_SLAVE_WAS_RUNNING));
  }

  /*
    Clean up start information if there was an attempt to start
    the IO thread to avoid any security issue.
  */
  if (slave_errno &&
      (thread_mask & SLAVE_IO) == SLAVE_IO)
    mi->reset_start_info();

  unlock_slave_threads(mi);

  if (slave_errno)
  {
    if (net_report)
      my_message(slave_errno, ER(slave_errno), MYF(0));
    DBUG_RETURN(1);
  }
  else if (net_report)
    my_ok(thd);

  DBUG_RETURN(0);
}