apply_event_and_update_pos(Log_event** ptr_ev, THD* thd, Relay_log_info* rli)
{
  int exec_res= 0;
  bool skip_event= FALSE;
  Log_event *ev= *ptr_ev;
  Log_event::enum_skip_reason reason= Log_event::EVENT_SKIP_NOT;

  DBUG_ENTER("apply_event_and_update_pos");

  DBUG_PRINT("exec_event",("%s(type_code: %d; server_id: %d)",
                           ev->get_type_str(), ev->get_type_code(),
                           ev->server_id));
  DBUG_PRINT("info", ("thd->options: %s%s; rli->last_event_start_time: %lu",
                      FLAGSTR(thd->variables.option_bits, OPTION_NOT_AUTOCOMMIT),
                      FLAGSTR(thd->variables.option_bits, OPTION_BEGIN),
                      (ulong) rli->last_event_start_time));

  /*
    Execute the event to change the database and update the binary
    log coordinates, but first we set some data that is needed for
    the thread.

    The event will be executed unless it is supposed to be skipped.

    Queries originating from this server must be skipped.  Low-level
    events (Format_description_log_event, Rotate_log_event,
    Stop_log_event) from this server must also be skipped. But for
    those we don't want to modify 'group_master_log_pos', because
    these events did not exist on the master.
    Format_description_log_event is not completely skipped.

    Skip queries specified by the user in 'slave_skip_counter'.  We
    can't however skip events that has something to do with the log
    files themselves.

    Filtering on own server id is extremely important, to ignore
    execution of events created by the creation/rotation of the relay
    log (remember that now the relay log starts with its Format_desc,
    has a Rotate etc).
  */
  /*
     Set the unmasked and actual server ids from the event
   */
  thd->server_id = ev->server_id; // use the original server id for logging
  thd->unmasked_server_id = ev->unmasked_server_id;
  thd->set_time();                            // time the query
  thd->lex->set_current_select(0);
  if (!ev->when.tv_sec)
    my_micro_time_to_timeval(my_micro_time(), &ev->when);
  ev->thd = thd; // because up to this point, ev->thd == 0

  if (!(rli->is_mts_recovery() && bitmap_is_set(&rli->recovery_groups,
                                                rli->mts_recovery_index)))
  {
    reason= ev->shall_skip(rli);
  }
#ifndef DBUG_OFF
  if (rli->is_mts_recovery())
  {
    DBUG_PRINT("mts", ("Mts is recovering %d, number of bits set %d, "
                       "bitmap is set %d, index %lu.\n",
                       rli->is_mts_recovery(),
                       bitmap_bits_set(&rli->recovery_groups),
                       bitmap_is_set(&rli->recovery_groups,
                                     rli->mts_recovery_index),
                       rli->mts_recovery_index));
  }
#endif
  if (reason == Log_event::EVENT_SKIP_COUNT)
  {
    sql_slave_skip_counter= --rli->slave_skip_counter;
    skip_event= TRUE;
  }
  if (reason == Log_event::EVENT_SKIP_NOT)
  {
    // Sleeps if needed, and unlocks rli->data_lock.
    if (sql_delay_event(ev, thd, rli))
      DBUG_RETURN(SLAVE_APPLY_EVENT_AND_UPDATE_POS_OK);

    exec_res= ev->apply_event(rli);

    if (!exec_res && (ev->worker != rli))
    {
      if (ev->worker)
      {
        Slave_job_item item= {ev}, *job_item= &item;
        Slave_worker *w= (Slave_worker *) ev->worker;
        // specially marked group typically with OVER_MAX_DBS_IN_EVENT_MTS db:s
        bool need_sync= ev->is_mts_group_isolated();

        // all events except BEGIN-query must be marked with a non-NULL Worker
        DBUG_ASSERT(((Slave_worker*) ev->worker) == rli->last_assigned_worker);

        DBUG_PRINT("Log_event::apply_event:",
                   ("-> job item data %p to W_%lu", job_item->data, w->id));

        // Reset mts in-group state
        if (rli->mts_group_status == Relay_log_info::MTS_END_GROUP)
        {
          // CGAP cleanup
          for (uint i= rli->curr_group_assigned_parts.elements; i > 0; i--)
            delete_dynamic_element(&rli->
                                   curr_group_assigned_parts, i - 1);
          // reset the B-group and Gtid-group marker
          rli->curr_group_seen_begin= rli->curr_group_seen_gtid= false;
          if (is_mts_db_partitioned(rli)||
              (!is_mts_db_partitioned(rli) &&
             !static_cast<Mts_submode_logical_clock*>
                (rli->current_mts_submode)->defer_new_group))
            rli->last_assigned_worker= NULL;
        }
        /* 
           Stroring GAQ index of the group that the event belongs to
           in the event. Deferred events are handled similarly below.
        */
        ev->mts_group_idx= rli->gaq->assigned_group_index;

        bool append_item_to_jobs_error= false;
        if (rli->curr_group_da.elements > 0)
        {
          /*
            the current event sorted out which partion the current group
            belongs to. It's time now to processed deferred array events.
          */
          for (uint i= 0; i < rli->curr_group_da.elements; i++)
          { 
            Slave_job_item da_item;
            get_dynamic(&rli->curr_group_da, (uchar*) &da_item.data, i);
            DBUG_PRINT("mts", ("Assigning job %llu to worker %lu",
                      ((Log_event* )da_item.data)->log_pos, w->id));
            static_cast<Log_event*>(da_item.data)->mts_group_idx=
              rli->gaq->assigned_group_index; // similarly to above
            if (!append_item_to_jobs_error)
              append_item_to_jobs_error= append_item_to_jobs(&da_item, w, rli);
            if (append_item_to_jobs_error)
              delete static_cast<Log_event*>(da_item.data);
          }
          if (rli->curr_group_da.elements > rli->curr_group_da.max_element)
          {
            // reallocate to less mem
            rli->curr_group_da.elements= rli->curr_group_da.max_element;
            rli->curr_group_da.max_element= 0;
            freeze_size(&rli->curr_group_da); // restores max_element
          }
          rli->curr_group_da.elements= 0;
        }
        if (append_item_to_jobs_error)
          DBUG_RETURN(SLAVE_APPLY_EVENT_AND_UPDATE_POS_APPEND_JOB_ERROR);

        DBUG_PRINT("mts", ("Assigning job %llu to worker %lu\n",
                   ((Log_event* )job_item->data)->log_pos, w->id));

        /* Notice `ev' instance can be destoyed after `append()' */
        if (append_item_to_jobs(job_item, w, rli))
          DBUG_RETURN(SLAVE_APPLY_EVENT_AND_UPDATE_POS_APPEND_JOB_ERROR);
        if (need_sync)
        {
          /*
            combination of over-max db:s and end of the current group
            forces to wait for the assigned groups completion by assigned
            to the event worker.
            Indeed MTS group status could be safely set to MTS_NOT_IN_GROUP
            after wait_() returns.
            No need to know a possible error out of synchronization call.
          */
          (void)rli->current_mts_submode->wait_for_workers_to_finish(rli);
        }

      }
      *ptr_ev= NULL; // announcing the event is passed to w-worker

      if (rli->is_parallel_exec() && rli->mts_events_assigned % 1024 == 1)
      {
        time_t my_now= my_time(0);

        if ((my_now - rli->mts_last_online_stat) >=
            mts_online_stat_period)
        {
          sql_print_information("Multi-threaded slave statistics: "
                                "seconds elapsed = %lu; "
                                "events assigned = %llu; "
                                "worker queues filled over overrun level = %lu; "
                                "waited due a Worker queue full = %lu; "
                                "waited due the total size = %lu; "
                                "slept when Workers occupied = %lu ",
                                static_cast<unsigned long>
                                (my_now - rli->mts_last_online_stat),
                                rli->mts_events_assigned,
                                rli->mts_wq_overrun_cnt,
                                rli->mts_wq_overfill_cnt,
                                rli->wq_size_waits_cnt,
                                rli->mts_wq_no_underrun_cnt);
          rli->mts_last_online_stat= my_now;
        }
      }
    }
  }
  else
    mysql_mutex_unlock(&rli->data_lock);
 
  DBUG_PRINT("info", ("apply_event error = %d", exec_res));
  if (exec_res == 0)
  {
    /*
      Positions are not updated here when an XID is processed. To make
      a slave crash-safe, positions must be updated while processing a
      XID event and as such do not need to be updated here again.

      However, if the event needs to be skipped, this means that it
      will not be processed and then positions need to be updated here.

      See sql/rpl_rli.h for further details.
    */
    int error= 0;
    if (*ptr_ev &&
        (ev->get_type_code() != XID_EVENT ||
         skip_event || (rli->is_mts_recovery() && !is_gtid_event(ev) &&
         (ev->ends_group() || !rli->mts_recovery_group_seen_begin) &&
          bitmap_is_set(&rli->recovery_groups, rli->mts_recovery_index))))
    {
#ifndef DBUG_OFF
      /*
        This only prints information to the debug trace.
        
        TODO: Print an informational message to the error log?
      */
      static const char *const explain[] = {
        // EVENT_SKIP_NOT,
        "not skipped",
        // EVENT_SKIP_IGNORE,
        "skipped because event should be ignored",
        // EVENT_SKIP_COUNT
        "skipped because event skip counter was non-zero"
      };
      DBUG_PRINT("info", ("OPTION_BEGIN: %d; IN_STMT: %d",
                          test(thd->variables.option_bits & OPTION_BEGIN),
                          rli->get_flag(Relay_log_info::IN_STMT)));
      DBUG_PRINT("skip_event", ("%s event was %s",
                                ev->get_type_str(), explain[reason]));
#endif

      error= ev->update_pos(rli);

#ifndef DBUG_OFF
      DBUG_PRINT("info", ("update_pos error = %d", error));
      if (!rli->belongs_to_client())
      {
        char buf[22];
        DBUG_PRINT("info", ("group %s %s",
                            llstr(rli->get_group_relay_log_pos(), buf),
                            rli->get_group_relay_log_name()));
        DBUG_PRINT("info", ("event %s %s",
                            llstr(rli->get_event_relay_log_pos(), buf),
                            rli->get_event_relay_log_name()));
      }
#endif
    }
    else
    {
      DBUG_ASSERT(*ptr_ev == ev || rli->is_parallel_exec() ||
		  (!ev->worker &&
		   (ev->get_type_code() == INTVAR_EVENT ||
		    ev->get_type_code() == RAND_EVENT ||
		    ev->get_type_code() == USER_VAR_EVENT)));

      rli->inc_event_relay_log_pos();
    }

    if (!error && rli->is_mts_recovery() &&
        ev->get_type_code() != ROTATE_EVENT &&
        ev->get_type_code() != FORMAT_DESCRIPTION_EVENT &&
        ev->get_type_code() != PREVIOUS_GTIDS_LOG_EVENT)
    {
      if (ev->starts_group())
      {
        rli->mts_recovery_group_seen_begin= true;
      }
      else if ((ev->ends_group() || !rli->mts_recovery_group_seen_begin) &&
               !is_gtid_event(ev))
      {
        rli->mts_recovery_index++;
        if (--rli->mts_recovery_group_cnt == 0)
        {
          rli->mts_recovery_index= 0;
          sql_print_information("Slave: MTS Recovery has completed at "
                                "relay log %s, position %llu "
                                "master log %s, position %llu.",
                                rli->get_group_relay_log_name(),
                                rli->get_group_relay_log_pos(),
                                rli->get_group_master_log_name(),
                                rli->get_group_master_log_pos());
#ifndef DBUG_OFF
          /* 
             Few tests wait for UNTIL_SQL_AFTER_MTS_GAPS completion.
             Due to exisiting convention the status won't change 
             prior to slave restarts.
             So making of UNTIL_SQL_AFTER_MTS_GAPS completion isdone here,
             and only in the debug build to make the test to catch the change
             despite a faulty design of UNTIL checking before execution.
          */
          if (rli->until_condition == Relay_log_info::UNTIL_SQL_AFTER_MTS_GAPS)
          {
            rli->until_condition= Relay_log_info::UNTIL_DONE;
          }
#endif
          // reset the Worker tables to remove last slave session time info
          if ((error= rli->mts_finalize_recovery()))
          {
            (void) Rpl_info_factory::reset_workers(rli);
          }
        }
        rli->mts_recovery_group_seen_begin= false;
        if (!error)
          error= rli->flush_info(true);
      }
    }

    if (error)
    {
      /*
        The update should not fail, so print an error message and
        return an error code.
        
        TODO: Replace this with a decent error message when merged
        with BUG#24954 (which adds several new error message).
      */
      char buf[22];
      rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR,
                  "It was not possible to update the positions"
                  " of the relay log information: the slave may"
                  " be in an inconsistent state."
                  " Stopped in %s position %s",
                  rli->get_group_relay_log_name(),
                  llstr(rli->get_group_relay_log_pos(), buf));
      DBUG_RETURN(SLAVE_APPLY_EVENT_AND_UPDATE_POS_UPDATE_POS_ERROR);
    }
  }

  DBUG_RETURN(exec_res ? SLAVE_APPLY_EVENT_AND_UPDATE_POS_APPLY_ERROR :
                         SLAVE_APPLY_EVENT_AND_UPDATE_POS_OK);
}