bool change_master(THD* thd, Master_info* mi)
{
  int thread_mask;
  const char* errmsg= 0;
  bool need_relay_log_purge= 1;
  char *var_master_log_name= NULL, *var_group_master_log_name= NULL;
  bool ret= false;
  char saved_host[HOSTNAME_LENGTH + 1], saved_bind_addr[HOSTNAME_LENGTH + 1];
  uint saved_port= 0;
  char saved_log_name[FN_REFLEN];
  my_off_t saved_log_pos= 0;
  my_bool save_relay_log_purge= relay_log_purge;
  bool mts_remove_workers= false;

  DBUG_ENTER("change_master");

  lock_slave_threads(mi);
  init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
  LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
  if (thread_mask) // We refuse if any slave thread is running
  {
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
    ret= true;
    goto err;
  }
  thread_mask= SLAVE_IO | SLAVE_SQL;

  THD_STAGE_INFO(thd, stage_changing_master);
  /* 
    We need to check if there is an empty master_host. Otherwise
    change master succeeds, a master.info file is created containing 
    empty master_host string and when issuing: start slave; an error
    is thrown stating that the server is not configured as slave.
    (See BUG#28796).
  */
  if(lex_mi->host && !*lex_mi->host) 
  {
    my_error(ER_WRONG_ARGUMENTS, MYF(0), "MASTER_HOST");
    unlock_slave_threads(mi);
    DBUG_RETURN(TRUE);
  }
  if (global_init_info(mi, false, thread_mask))
  {
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
    ret= true;
    goto err;
  }
  if (mi->rli->mts_recovery_group_cnt)
  {
    /*
      Change-Master can't be done if there is a mts group gap.
      That requires mts-recovery which START SLAVE provides.
    */
    DBUG_ASSERT(mi->rli->recovery_parallel_workers);

    my_message(ER_MTS_CHANGE_MASTER_CANT_RUN_WITH_GAPS,
               ER(ER_MTS_CHANGE_MASTER_CANT_RUN_WITH_GAPS), MYF(0));
    ret= true;
    goto err;
  }
  else
  {
    /*
      Lack of mts group gaps makes Workers info stale
      regardless of need_relay_log_purge computation.
    */
    if (mi->rli->recovery_parallel_workers)
      mts_remove_workers= true;
  }
  /*
    We cannot specify auto position and set either the coordinates
    on master or slave. If we try to do so, an error message is
    printed out.
  */
  if (lex_mi->log_file_name != NULL || lex_mi->pos != 0 || 
      lex_mi->relay_log_name != NULL || lex_mi->relay_log_pos != 0)
  {
    if (lex_mi->auto_position == LEX_MASTER_INFO::LEX_MI_ENABLE ||
        (lex_mi->auto_position != LEX_MASTER_INFO::LEX_MI_DISABLE &&
         mi->is_auto_position()))
    {
      my_message(ER_BAD_SLAVE_AUTO_POSITION,
                 ER(ER_BAD_SLAVE_AUTO_POSITION), MYF(0));
      ret= true;
      goto err;
    }
  }

  // CHANGE MASTER TO MASTER_AUTO_POSITION = 1 requires GTID_MODE = ON
  if (lex_mi->auto_position == LEX_MASTER_INFO::LEX_MI_ENABLE && gtid_mode != 3)
  {
    my_message(ER_AUTO_POSITION_REQUIRES_GTID_MODE_ON,
               ER(ER_AUTO_POSITION_REQUIRES_GTID_MODE_ON), MYF(0));
    ret= true;
    goto err;
  }

  /*
    Data lock not needed since we have already stopped the running threads,
    and we have the hold on the run locks which will keep all threads that
    could possibly modify the data structures from running
  */

  /*
    Before processing the command, save the previous state.
  */
  strmake(saved_host, mi->host, HOSTNAME_LENGTH);
  strmake(saved_bind_addr, mi->bind_addr, HOSTNAME_LENGTH);
  saved_port= mi->port;
  strmake(saved_log_name, mi->get_master_log_name(), FN_REFLEN - 1);
  saved_log_pos= mi->get_master_log_pos();

  /*
    If the user specified host or port without binlog or position,
    reset binlog's name to FIRST and position to 4.
  */

  if ((lex_mi->host && strcmp(lex_mi->host, mi->host)) ||
      (lex_mi->port && lex_mi->port != mi->port))
  {
    /*
      This is necessary because the primary key, i.e. host or port, has
      changed.

      The repository does not support direct changes on the primary key,
      so the row is dropped and re-inserted with a new primary key. If we
      don't do that, the master info repository we will end up with several
      rows.
    */
    if (mi->clean_info())
    {
      ret= true;
      goto err;
    }
    mi->master_uuid[0]= 0;
    mi->master_id= 0;
  }

  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
  {
    var_master_log_name= const_cast<char*>(mi->get_master_log_name());
    var_master_log_name[0]= '\0';
    mi->set_master_log_pos(BIN_LOG_HEADER_SIZE);
  }

  if (lex_mi->log_file_name)
    mi->set_master_log_name(lex_mi->log_file_name);
  if (lex_mi->pos)
  {
    mi->set_master_log_pos(lex_mi->pos);
  }
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->get_master_log_pos()));

  if (lex_mi->user || lex_mi->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
    push_warning(thd, Sql_condition::SL_NOTE,
                 ER_INSECURE_CHANGE_MASTER,
                 ER(ER_INSECURE_CHANGE_MASTER));
  }

  if (lex_mi->user)
    mi->set_user(lex_mi->user);

  if (lex_mi->password)
  {
    if (mi->set_password(lex_mi->password, strlen(lex_mi->password)))
    {
      /*
        After implementing WL#5769, we should create a better error message
        to denote that the call may have failed due to an error while trying
        to encrypt/store the password in a secure key store.
      */
      my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
      ret= false;
      goto err;
    }
  }
  if (lex_mi->host)
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
  if (lex_mi->bind_addr)
    strmake(mi->bind_addr, lex_mi->bind_addr, sizeof(mi->bind_addr)-1);
  if (lex_mi->port)
    mi->port = lex_mi->port;
  if (lex_mi->connect_retry)
    mi->connect_retry = lex_mi->connect_retry;
  if (lex_mi->retry_count_opt !=  LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->retry_count = lex_mi->retry_count;
  if (lex_mi->heartbeat_opt != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->heartbeat_period = lex_mi->heartbeat_period;
  else
    mi->heartbeat_period= min<float>(SLAVE_MAX_HEARTBEAT_PERIOD,
                                     (slave_net_timeout/2.0));
  mi->received_heartbeats= LL(0); // counter lives until master is CHANGEd
  /*
    reset the last time server_id list if the current CHANGE MASTER 
    is mentioning IGNORE_SERVER_IDS= (...)
  */
  if (lex_mi->repl_ignore_server_ids_opt == LEX_MASTER_INFO::LEX_MI_ENABLE)
    reset_dynamic(&(mi->ignore_server_ids->dynamic_ids));
  for (uint i= 0; i < lex_mi->repl_ignore_server_ids.elements; i++)
  {
    ulong s_id;
    get_dynamic(&lex_mi->repl_ignore_server_ids, (uchar*) &s_id, i);
    if (s_id == ::server_id && replicate_same_server_id)
    {
      my_error(ER_SLAVE_IGNORE_SERVER_IDS, MYF(0), static_cast<int>(s_id));
      ret= TRUE;
      goto err;
    }
    else
    {
      if (bsearch((const ulong *) &s_id,
                  mi->ignore_server_ids->dynamic_ids.buffer,
                  mi->ignore_server_ids->dynamic_ids.elements, sizeof(ulong),
                  (int (*) (const void*, const void*))
                  change_master_server_id_cmp) == NULL)
        insert_dynamic(&(mi->ignore_server_ids->dynamic_ids), (uchar*) &s_id);
    }
  }
  sort_dynamic(&(mi->ignore_server_ids->dynamic_ids), (qsort_cmp) change_master_server_id_cmp);

  if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE);

  if (lex_mi->sql_delay != -1)
    mi->rli->set_sql_delay(lex_mi->sql_delay);

  if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->ssl_verify_server_cert=
      (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::LEX_MI_ENABLE);

  if (lex_mi->ssl_ca)
    strmake(mi->ssl_ca, lex_mi->ssl_ca, sizeof(mi->ssl_ca)-1);
  if (lex_mi->ssl_capath)
    strmake(mi->ssl_capath, lex_mi->ssl_capath, sizeof(mi->ssl_capath)-1);
  if (lex_mi->ssl_cert)
    strmake(mi->ssl_cert, lex_mi->ssl_cert, sizeof(mi->ssl_cert)-1);
  if (lex_mi->ssl_cipher)
    strmake(mi->ssl_cipher, lex_mi->ssl_cipher, sizeof(mi->ssl_cipher)-1);
  if (lex_mi->ssl_key)
    strmake(mi->ssl_key, lex_mi->ssl_key, sizeof(mi->ssl_key)-1);
  if (lex_mi->ssl_crl)
    strmake(mi->ssl_crl, lex_mi->ssl_crl, sizeof(mi->ssl_crl)-1);
  if (lex_mi->ssl_crlpath)
    strmake(mi->ssl_crlpath, lex_mi->ssl_crlpath, sizeof(mi->ssl_crlpath)-1);
#ifndef HAVE_OPENSSL
  if (lex_mi->ssl || lex_mi->ssl_ca || lex_mi->ssl_capath ||
      lex_mi->ssl_cert || lex_mi->ssl_cipher || lex_mi->ssl_key ||
      lex_mi->ssl_verify_server_cert || lex_mi->ssl_crl || lex_mi->ssl_crlpath)
    push_warning(thd, Sql_condition::SL_NOTE,
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
#endif

  if (lex_mi->relay_log_name)
  {
    need_relay_log_purge= 0;
    char relay_log_name[FN_REFLEN];

    mi->rli->relay_log.make_log_name(relay_log_name, lex_mi->relay_log_name);
    mi->rli->set_group_relay_log_name(relay_log_name);
    mi->rli->set_event_relay_log_name(relay_log_name);
    mi->rli->is_group_master_log_pos_invalid= true;
  }

  if (lex_mi->relay_log_pos)
  {
    need_relay_log_purge= 0;
    mi->rli->set_group_relay_log_pos(lex_mi->relay_log_pos);
    mi->rli->set_event_relay_log_pos(lex_mi->relay_log_pos);
    mi->rli->is_group_master_log_pos_invalid= true;
  }

  /*
   Consider a situation like this

     mysql> STOP SLAVE;
     mysql> CHANGE MASTER TO RELAY_LOG_POS=4;
     mysql> CHANGE_MASTER TO MASTER_HOST='something.example.com',
                             MASTER_PORT=23789,
                             MASTER_USER='repl_user';
     mysql> START SLAVE;

   In this case the group_master_log_pos is valid after the second change
   master so this flag should be unset.
   */
  if (!lex_mi->relay_log_name && !lex_mi->relay_log_pos)
    mi->rli->is_group_master_log_pos_invalid= false;

  /*
    If user did specify neither host nor port nor any log name nor any log
    pos, i.e. he specified only user/password/master_connect_retry, he probably
    wants replication to resume from where it had left, i.e. from the
    coordinates of the **SQL** thread (imagine the case where the I/O is ahead
    of the SQL; restarting from the coordinates of the I/O would lose some
    events which is probably unwanted when you are just doing minor changes
    like changing master_connect_retry).
    A side-effect is that if only the I/O thread was started, this thread may
    restart from ''/4 after the CHANGE MASTER. That's a minor problem (it is a
    much more unlikely situation than the one we are fixing here).
    Note: coordinates of the SQL thread must be read here, before the
    'if (need_relay_log_purge)' block which resets them.
  */
  if (!lex_mi->host && !lex_mi->port &&
      !lex_mi->log_file_name && !lex_mi->pos &&
      need_relay_log_purge)
   {
     /*
       Sometimes mi->rli->master_log_pos == 0 (it happens when the SQL thread is
       not initialized), so we use a max().
       What happens to mi->rli->master_log_pos during the initialization stages
       of replication is not 100% clear, so we guard against problems using
       max().
      */
     mi->set_master_log_pos(max<ulonglong>(BIN_LOG_HEADER_SIZE,
                                           mi->rli->get_group_master_log_pos()));
     mi->set_master_log_name(mi->rli->get_group_master_log_name());
  }

  /*
    Sets if the slave should connect to the master and look for
    GTIDs.
  */
  if (lex_mi->auto_position != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->set_auto_position(
      (lex_mi->auto_position == LEX_MASTER_INFO::LEX_MI_ENABLE));

  /*
    Relay log's IO_CACHE may not be inited, if rli->inited==0 (server was never
    a slave before).
  */
  if (flush_master_info(mi, true))
  {
    my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
    ret= TRUE;
    goto err;
  }
  if (need_relay_log_purge)
  {
    relay_log_purge= 1;
    THD_STAGE_INFO(thd, stage_purging_old_relay_logs);
    if (mi->rli->purge_relay_logs(thd,
                                  0 /* not only reset, but also reinit */,
                                  &errmsg))
    {
      my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
      ret= TRUE;
      goto err;
    }
  }
  else
  {
    const char* msg;
    relay_log_purge= 0;
    /* Relay log is already initialized */
    
    if (mi->rli->init_relay_log_pos(mi->rli->get_group_relay_log_name(),
                                    mi->rli->get_group_relay_log_pos(),
                                    true/*need_data_lock=true*/,
                                    &msg, 0))
    {
      my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
      ret= TRUE;
      goto err;
    }
  }
  relay_log_purge= save_relay_log_purge;

  /*
    Coordinates in rli were spoilt by the 'if (need_relay_log_purge)' block,
    so restore them to good values. If we left them to ''/0, that would work;
    but that would fail in the case of 2 successive CHANGE MASTER (without a
    START SLAVE in between): because first one would set the coords in mi to
    the good values of those in rli, the set those in rli to ''/0, then
    second CHANGE MASTER would set the coords in mi to those of rli, i.e. to
    ''/0: we have lost all copies of the original good coordinates.
    That's why we always save good coords in rli.
  */
  if (need_relay_log_purge)
  {
    mi->rli->set_group_master_log_pos(mi->get_master_log_pos());
    DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->get_master_log_pos()));
    mi->rli->set_group_master_log_name(mi->get_master_log_name());
  }
  var_group_master_log_name=  const_cast<char *>(mi->rli->get_group_master_log_name());
  if (!var_group_master_log_name[0]) // uninitialized case
    mi->rli->set_group_master_log_pos(0);

  mysql_mutex_lock(&mi->rli->data_lock);
  mi->rli->abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
  /* Clear the errors, for a clean start */
  mi->rli->clear_error();
  if (mi->rli->workers_array_initialized)
  {
    for(uint i= 0; i < mi->rli->get_worker_count(); i++)
    {
      mi->rli->get_worker(i)->clear_error();
    }
  }

  mi->rli->clear_until_condition();

  sql_print_information("'CHANGE MASTER TO executed'. "
    "Previous state master_host='%s', master_port= %u, master_log_file='%s', "
    "master_log_pos= %ld, master_bind='%s'. "
    "New state master_host='%s', master_port= %u, master_log_file='%s', "
    "master_log_pos= %ld, master_bind='%s'.", 
    saved_host, saved_port, saved_log_name, (ulong) saved_log_pos,
    saved_bind_addr, mi->host, mi->port, mi->get_master_log_name(),
    (ulong) mi->get_master_log_pos(), mi->bind_addr);

  /*
    If we don't write new coordinates to disk now, then old will remain in
    relay-log.info until START SLAVE is issued; but if mysqld is shutdown
    before START SLAVE, then old will remain in relay-log.info, and will be the
    in-memory value at restart (thus causing errors, as the old relay log does
    not exist anymore).

    Notice that the rli table is available exclusively as slave is not
    running.
  */
  DBUG_ASSERT(!mi->rli->slave_running);
  if ((ret= mi->rli->flush_info(true)))
    my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush relay info file.");
  mysql_cond_broadcast(&mi->data_cond);
  mysql_mutex_unlock(&mi->rli->data_lock);

err:
  unlock_slave_threads(mi);
  if (ret == FALSE)
  {
    if (!mts_remove_workers)
      my_ok(thd);
    else
      if (!Rpl_info_factory::reset_workers(mi->rli))
        my_ok(thd);
      else
        my_error(ER_MTS_RESET_WORKERS, MYF(0));
  }
  DBUG_RETURN(ret);
}