int slave_start_workers(Relay_log_info *rli, ulong n, bool *mts_inited)
{
  uint i;
  int error= 0;

  mysql_mutex_assert_owner(&rli->run_lock);

  if (n == 0 && rli->mts_recovery_group_cnt == 0)
  {
    reset_dynamic(&rli->workers);
    goto end;
  }

  *mts_inited= true;

  /*
    The requested through argument number of Workers can be different 
     from the previous time which ended with an error. Thereby
     the effective number of configured Workers is max of the two.
  */
  rli->init_workers(max(n, rli->recovery_parallel_workers));

  // CGAP dynarray holds id:s of partitions of the Current being executed Group
  my_init_dynamic_array(&rli->curr_group_assigned_parts,
                        sizeof(db_worker_hash_entry*),
                        SLAVE_INIT_DBS_IN_GROUP, 1);
  rli->last_assigned_worker= NULL;     // associated with curr_group_assigned
  my_init_dynamic_array(&rli->curr_group_da, sizeof(Log_event*), 8, 2);
  // Least_occupied_workers array to hold items size of Slave_jobs_queue::len
  my_init_dynamic_array(&rli->least_occupied_workers, sizeof(ulong), n, 0); 

  /* 
     GAQ  queue holds seqno:s of scheduled groups. C polls workers in 
     @c opt_mts_checkpoint_period to update GAQ (see @c next_event())
     The length of GAQ is set to be equal to checkpoint_group.
     Notice, the size matters for mts_checkpoint_routine's progress loop.
  */

  rli->gaq= new Slave_committed_queue(rli->get_group_master_log_name(),
                                      sizeof(Slave_job_group),
                                      rli->checkpoint_group, n);
  if (!rli->gaq->inited)
    return 1;

  // length of WQ is actually constant though can be made configurable
  rli->mts_slave_worker_queue_len_max= mts_slave_worker_queue_len_max;
  rli->mts_pending_jobs_size= 0;
  rli->mts_pending_jobs_size_max= ::opt_mts_pending_jobs_size_max;
  rli->mts_wq_underrun_w_id= MTS_WORKER_UNDEF;
  rli->mts_wq_excess_cnt= 0;
  rli->mts_wq_overrun_cnt= 0;
  rli->mts_wq_oversize= FALSE;
  rli->mts_coordinator_basic_nap= mts_coordinator_basic_nap;
  rli->mts_worker_underrun_level= mts_worker_underrun_level;
  rli->curr_group_seen_begin= rli->curr_group_seen_gtid= false;
  rli->curr_group_isolated= FALSE;
  rli->checkpoint_seqno= 0;
  rli->mts_last_online_stat= my_time(0);
  rli->mts_group_status= Relay_log_info::MTS_NOT_IN_GROUP;
  /*
    dyn memory to consume by Coordinator per event
  */
  init_alloc_root(key_memory_rli_mts_coor,
                  &rli->mts_coor_mem_root, NAME_LEN,
                  (MAX_DBS_IN_EVENT_MTS / 2) * NAME_LEN);

  if (init_hash_workers(n))  // MTS: mapping_db_to_worker
  {
    sql_print_error("Failed to init partitions hash");
    error= 1;
    goto err;
  }

  for (i= 0; i < n; i++)
  {
    if ((error= slave_start_single_worker(rli, i)))
      goto err;
  }

end:
  /*
    Free the buffer that was being used to report worker's status through
    the table performance_schema.table_replication_execute_status_by_worker
    between stop slave and next start slave.
  */
  for (int i= rli->workers_copy_pfs.size() - 1; i >= 0; i--)
    delete rli->workers_copy_pfs[i];
  rli->workers_copy_pfs.clear();

  rli->slave_parallel_workers= n;
  // Effective end of the recovery right now when there is no gaps
  if (!error && rli->mts_recovery_group_cnt == 0)
  {
    if ((error= rli->mts_finalize_recovery()))
      (void) Rpl_info_factory::reset_workers(rli);
    if (!error)
      error= rli->flush_info(TRUE);
  }

err:
  return error;
}