static int queue_event(Master_info* mi,const char* buf, ulong event_len)
{
  int error= 0;
  String error_msg;
  ulong inc_pos= 0;
  Relay_log_info *rli= mi->rli;
  mysql_mutex_t *log_lock= rli->relay_log.get_log_lock();
  ulong s_id;
  bool unlock_data_lock= TRUE;
  /*
    FD_q must have been prepared for the first R_a event
    inside get_master_version_and_clock()
    Show-up of FD:s affects checksum_alg at once because
    that changes FD_queue.
  */
  uint8 checksum_alg= mi->checksum_alg_before_fd != BINLOG_CHECKSUM_ALG_UNDEF ? 
    mi->checksum_alg_before_fd :
    mi->rli->relay_log.relay_log_checksum_alg;

  char *save_buf= NULL; // needed for checksumming the fake Rotate event
  char rot_buf[LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN + FN_REFLEN];
  Gtid gtid= { 0, 0 };
  Log_event_type event_type= (Log_event_type)buf[EVENT_TYPE_OFFSET];

  DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_OFF || 
              checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF || 
              checksum_alg == BINLOG_CHECKSUM_ALG_CRC32); 

  DBUG_ENTER("queue_event");
  /*
    FD_queue checksum alg description does not apply in a case of
    FD itself. The one carries both parts of the checksum data.
  */
  if (event_type == FORMAT_DESCRIPTION_EVENT)
  {
    checksum_alg= get_checksum_alg(buf, event_len);
  }
  else if (event_type == START_EVENT_V3)
  {
    // checksum behaviour is similar to the pre-checksum FD handling
    mi->checksum_alg_before_fd= BINLOG_CHECKSUM_ALG_UNDEF;
    mysql_mutex_lock(&mi->data_lock);
    mi->get_mi_description_event()->checksum_alg=
      mi->rli->relay_log.relay_log_checksum_alg= checksum_alg=
      BINLOG_CHECKSUM_ALG_OFF;
    mysql_mutex_unlock(&mi->data_lock);
  }

  // does not hold always because of old binlog can work with NM 
  // DBUG_ASSERT(checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF);

  // should hold unless manipulations with RL. Tests that do that
  // will have to refine the clause.
  DBUG_ASSERT(mi->rli->relay_log.relay_log_checksum_alg !=
              BINLOG_CHECKSUM_ALG_UNDEF);
              
  // Emulate the network corruption
  DBUG_EXECUTE_IF("corrupt_queue_event",
    if (event_type != FORMAT_DESCRIPTION_EVENT)
    {
      char *debug_event_buf_c = (char*) buf;
      int debug_cor_pos = rand() % (event_len - BINLOG_CHECKSUM_LEN);
      debug_event_buf_c[debug_cor_pos] =~ debug_event_buf_c[debug_cor_pos];
      DBUG_PRINT("info", ("Corrupt the event at queue_event: byte on position %d", debug_cor_pos));
      DBUG_SET("");
    }
  );
                                              
  if (event_checksum_test((uchar *) buf, event_len, checksum_alg))
  {
    error= ER_NETWORK_READ_EVENT_CHECKSUM_FAILURE;
    unlock_data_lock= FALSE;
    goto err;
  }

  mysql_mutex_lock(&mi->data_lock);

  if (mi->get_mi_description_event()->binlog_version < 4 &&
      event_type != FORMAT_DESCRIPTION_EVENT /* a way to escape */)
  {
    int ret= queue_old_event(mi,buf,event_len);
    mysql_mutex_unlock(&mi->data_lock);
    DBUG_RETURN(ret);
  }

  switch (event_type) {
  case STOP_EVENT:
    /*
      We needn't write this event to the relay log. Indeed, it just indicates a
      master server shutdown. The only thing this does is cleaning. But
      cleaning is already done on a per-master-thread basis (as the master
      server is shutting down cleanly, it has written all DROP TEMPORARY TABLE
      prepared statements' deletion are TODO only when we binlog prep stmts).

      We don't even increment mi->get_master_log_pos(), because we may be just after
      a Rotate event. Btw, in a few milliseconds we are going to have a Start
      event from the next binlog (unless the master is presently running
      without --log-bin).
    */
    goto err;
  case ROTATE_EVENT:
  {
    Rotate_log_event rev(buf, checksum_alg != BINLOG_CHECKSUM_ALG_OFF ?
                         event_len - BINLOG_CHECKSUM_LEN : event_len,
                         mi->get_mi_description_event());

    if (unlikely(process_io_rotate(mi, &rev)))
    {
      error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
      goto err;
    }
    /* 
       Checksum special cases for the fake Rotate (R_f) event caused by the protocol
       of events generation and serialization in RL where Rotate of master is 
       queued right next to FD of slave.
       Since it's only FD that carries the alg desc of FD_s has to apply to R_m.
       Two special rules apply only to the first R_f which comes in before any FD_m.
       The 2nd R_f should be compatible with the FD_s that must have taken over
       the last seen FD_m's (A).
       
       RSC_1: If OM \and fake Rotate \and slave is configured to
              to compute checksum for its first FD event for RL
              the fake Rotate gets checksummed here.
    */
    if (uint4korr(&buf[0]) == 0 && checksum_alg == BINLOG_CHECKSUM_ALG_OFF &&
        mi->rli->relay_log.relay_log_checksum_alg != BINLOG_CHECKSUM_ALG_OFF)
    {
      ha_checksum rot_crc= my_checksum(0L, NULL, 0);
      event_len += BINLOG_CHECKSUM_LEN;
      memcpy(rot_buf, buf, event_len - BINLOG_CHECKSUM_LEN);
      int4store(&rot_buf[EVENT_LEN_OFFSET],
                uint4korr(rot_buf + EVENT_LEN_OFFSET) + BINLOG_CHECKSUM_LEN);
      rot_crc= my_checksum(rot_crc, (const uchar *) rot_buf,
                           event_len - BINLOG_CHECKSUM_LEN);
      int4store(&rot_buf[event_len - BINLOG_CHECKSUM_LEN], rot_crc);
      DBUG_ASSERT(event_len == uint4korr(&rot_buf[EVENT_LEN_OFFSET]));
      DBUG_ASSERT(mi->get_mi_description_event()->checksum_alg ==
                  mi->rli->relay_log.relay_log_checksum_alg);
      /* the first one */
      DBUG_ASSERT(mi->checksum_alg_before_fd != BINLOG_CHECKSUM_ALG_UNDEF);
      save_buf= (char *) buf;
      buf= rot_buf;
    }
    else
      /*
        RSC_2: If NM \and fake Rotate \and slave does not compute checksum
        the fake Rotate's checksum is stripped off before relay-logging.
      */
      if (uint4korr(&buf[0]) == 0 && checksum_alg != BINLOG_CHECKSUM_ALG_OFF &&
          mi->rli->relay_log.relay_log_checksum_alg == BINLOG_CHECKSUM_ALG_OFF)
      {
        event_len -= BINLOG_CHECKSUM_LEN;
        memcpy(rot_buf, buf, event_len);
        int4store(&rot_buf[EVENT_LEN_OFFSET],
                  uint4korr(rot_buf + EVENT_LEN_OFFSET) - BINLOG_CHECKSUM_LEN);
        DBUG_ASSERT(event_len == uint4korr(&rot_buf[EVENT_LEN_OFFSET]));
        DBUG_ASSERT(mi->get_mi_description_event()->checksum_alg ==
                    mi->rli->relay_log.relay_log_checksum_alg);
        /* the first one */
        DBUG_ASSERT(mi->checksum_alg_before_fd != BINLOG_CHECKSUM_ALG_UNDEF);
        save_buf= (char *) buf;
        buf= rot_buf;
      }
    /*
      Now the I/O thread has just changed its mi->get_master_log_name(), so
      incrementing mi->get_master_log_pos() is nonsense.
    */
    inc_pos= 0;
    break;
  }
  case FORMAT_DESCRIPTION_EVENT:
  {
    /*
      Create an event, and save it (when we rotate the relay log, we will have
      to write this event again).
    */
    /*
      We are the only thread which reads/writes mi_description_event.
      The relay_log struct does not move (though some members of it can
      change), so we needn't any lock (no rli->data_lock, no log lock).
    */
    const char* errmsg;
    // mark it as undefined that is irrelevant anymore
    mi->checksum_alg_before_fd= BINLOG_CHECKSUM_ALG_UNDEF;
    Format_description_log_event *new_fdle=
      (Format_description_log_event*)
      Log_event::read_log_event(buf, event_len, &errmsg,
                                mi->get_mi_description_event(), 1);
    if (new_fdle == NULL)
    {
      error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
      goto err;
    }
    if (new_fdle->checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF)
      new_fdle->checksum_alg= BINLOG_CHECKSUM_ALG_OFF;
    mi->set_mi_description_event(new_fdle);

    /* installing new value of checksum Alg for relay log */
    mi->rli->relay_log.relay_log_checksum_alg= new_fdle->checksum_alg;

    /*
       Though this does some conversion to the slave's format, this will
       preserve the master's binlog format version, and number of event types.
    */
    /*
       If the event was not requested by the slave (the slave did not ask for
       it), i.e. has end_log_pos=0, we do not increment mi->get_master_log_pos()
    */
    inc_pos= uint4korr(buf+LOG_POS_OFFSET) ? event_len : 0;
    DBUG_PRINT("info",("binlog format is now %d",
                       mi->get_mi_description_event()->binlog_version));

  }
  break;

  case HEARTBEAT_LOG_EVENT:
  {
    /*
      HB (heartbeat) cannot come before RL (Relay)
    */
    char  llbuf[22];
    Heartbeat_log_event hb(buf,
                           mi->rli->relay_log.relay_log_checksum_alg
                           != BINLOG_CHECKSUM_ALG_OFF ?
                           event_len - BINLOG_CHECKSUM_LEN : event_len,
                           mi->get_mi_description_event());
    if (!hb.is_valid())
    {
      error= ER_SLAVE_HEARTBEAT_FAILURE;
      error_msg.append(STRING_WITH_LEN("inconsistent heartbeat event content;"));
      error_msg.append(STRING_WITH_LEN("the event's data: log_file_name "));
      error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident()));
      error_msg.append(STRING_WITH_LEN(" log_pos "));
      llstr(hb.log_pos, llbuf);
      error_msg.append(llbuf, strlen(llbuf));
      goto err;
    }
    mi->received_heartbeats++;
    mi->last_heartbeat= my_time(0);


    /*
      During GTID protocol, if the master skips transactions,
      a heartbeat event is sent to the slave at the end of last
      skipped transaction to update coordinates.

      I/O thread receives the heartbeat event and updates mi
      only if the received heartbeat position is greater than
      mi->get_master_log_pos(). This event is written to the
      relay log as an ignored Rotate event. SQL thread reads
      the rotate event only to update the coordinates corresponding
      to the last skipped transaction. Note that,
      we update only the positions and not the file names, as a ROTATE
      EVENT from the master prior to this will update the file name.
    */
    if (mi->is_auto_position()  && mi->get_master_log_pos() < hb.log_pos
        &&  mi->get_master_log_name() != NULL)
    {

      DBUG_ASSERT(memcmp(const_cast<char*>(mi->get_master_log_name()),
                         hb.get_log_ident(), hb.get_ident_len()) == 0);

      mi->set_master_log_pos(hb.log_pos);

      /*
         Put this heartbeat event in the relay log as a Rotate Event.
      */
      inc_pos= 0;
      memcpy(rli->ign_master_log_name_end, mi->get_master_log_name(),
             FN_REFLEN);
      rli->ign_master_log_pos_end = mi->get_master_log_pos();

      if (write_ignored_events_info_to_relay_log(mi->info_thd, mi))
        goto err;
    }

    /* 
       compare local and event's versions of log_file, log_pos.
       
       Heartbeat is sent only after an event corresponding to the corrdinates
       the heartbeat carries.
       Slave can not have a difference in coordinates except in the only
       special case when mi->get_master_log_name(), mi->get_master_log_pos() have never
       been updated by Rotate event i.e when slave does not have any history
       with the master (and thereafter mi->get_master_log_pos() is NULL).

       TODO: handling `when' for SHOW SLAVE STATUS' snds behind
    */
    if ((memcmp(const_cast<char *>(mi->get_master_log_name()),
                hb.get_log_ident(), hb.get_ident_len())
         && mi->get_master_log_name() != NULL)
        || ((mi->get_master_log_pos() != hb.log_pos && gtid_mode == 0) || 
            /*
              When Gtid mode is on only monotocity can be claimed.
              Todo: enhance HB event with the skipped events size
              and to convert HB.pos  == MI.pos to HB.pos - HB.skip_size == MI.pos
            */
            (mi->get_master_log_pos() > hb.log_pos)))
    {
      /* missed events of heartbeat from the past */
      error= ER_SLAVE_HEARTBEAT_FAILURE;
      error_msg.append(STRING_WITH_LEN("heartbeat is not compatible with local info;"));
      error_msg.append(STRING_WITH_LEN("the event's data: log_file_name "));
      error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident()));
      error_msg.append(STRING_WITH_LEN(" log_pos "));
      llstr(hb.log_pos, llbuf);
      error_msg.append(llbuf, strlen(llbuf));
      goto err;
    }
    goto skip_relay_logging;
  }
  break;

  case PREVIOUS_GTIDS_LOG_EVENT:
  {
    if (gtid_mode == 0)
    {
      error= ER_FOUND_GTID_EVENT_WHEN_GTID_MODE_IS_OFF;
      goto err;
    }
    /*
      This event does not have any meaning for the slave and
      was just sent to show the slave the master is making
      progress and avoid possible deadlocks.
      So at this point, the event is replaced by a rotate
      event what will make the slave to update what it knows
      about the master's coordinates.
    */
    inc_pos= 0;
    mi->set_master_log_pos(mi->get_master_log_pos() + event_len);
    memcpy(rli->ign_master_log_name_end, mi->get_master_log_name(), FN_REFLEN);
    rli->ign_master_log_pos_end= mi->get_master_log_pos();

    if (write_ignored_events_info_to_relay_log(mi->info_thd, mi))
      goto err;

    goto skip_relay_logging;
  }
  break;

  case GTID_LOG_EVENT:
  {
    if (gtid_mode == 0)
    {
      error= ER_FOUND_GTID_EVENT_WHEN_GTID_MODE_IS_OFF;
      goto err;
    }
    global_sid_lock->rdlock();
    Gtid_log_event gtid_ev(buf, checksum_alg != BINLOG_CHECKSUM_ALG_OFF ?
                           event_len - BINLOG_CHECKSUM_LEN : event_len,
                           mi->get_mi_description_event());
    gtid.sidno= gtid_ev.get_sidno(false);
    global_sid_lock->unlock();
    if (gtid.sidno < 0)
      goto err;
    gtid.gno= gtid_ev.get_gno();
    inc_pos= event_len;
  }
  break;

  case ANONYMOUS_GTID_LOG_EVENT:

  default:
    inc_pos= event_len;
  break;
  }

  /*
     If this event is originating from this server, don't queue it.
     We don't check this for 3.23 events because it's simpler like this; 3.23
     will be filtered anyway by the SQL slave thread which also tests the
     server id (we must also keep this test in the SQL thread, in case somebody
     upgrades a 4.0 slave which has a not-filtered relay log).

     ANY event coming from ourselves can be ignored: it is obvious for queries;
     for STOP_EVENT/ROTATE_EVENT/START_EVENT: these cannot come from ourselves
     (--log-slave-updates would not log that) unless this slave is also its
     direct master (an unsupported, useless setup!).
  */

  mysql_mutex_lock(log_lock);
  s_id= uint4korr(buf + SERVER_ID_OFFSET);

  /*
    If server_id_bits option is set we need to mask out irrelevant bits
    when checking server_id, but we still put the full unmasked server_id
    into the Relay log so that it can be accessed when applying the event
  */
  s_id&= opt_server_id_mask;

  if ((s_id == ::server_id && !mi->rli->replicate_same_server_id) ||
      /*
        the following conjunction deals with IGNORE_SERVER_IDS, if set
        If the master is on the ignore list, execution of
        format description log events and rotate events is necessary.
      */
      (mi->ignore_server_ids->dynamic_ids.elements > 0 &&
       mi->shall_ignore_server_id(s_id) &&
       /* everything is filtered out from non-master */
       (s_id != mi->master_id ||
        /* for the master meta information is necessary */
        (event_type != FORMAT_DESCRIPTION_EVENT &&
         event_type != ROTATE_EVENT))))
  {
    /*
      Do not write it to the relay log.
      a) We still want to increment mi->get_master_log_pos(), so that we won't
      re-read this event from the master if the slave IO thread is now
      stopped/restarted (more efficient if the events we are ignoring are big
      LOAD DATA INFILE).
      b) We want to record that we are skipping events, for the information of
      the slave SQL thread, otherwise that thread may let
      rli->group_relay_log_pos stay too small if the last binlog's event is
      ignored.
      But events which were generated by this slave and which do not exist in
      the master's binlog (i.e. Format_desc, Rotate & Stop) should not increment
      mi->get_master_log_pos().
      If the event is originated remotely and is being filtered out by
      IGNORE_SERVER_IDS it increments mi->get_master_log_pos()
      as well as rli->group_relay_log_pos.
    */
    if (!(s_id == ::server_id && !mi->rli->replicate_same_server_id) ||
        (event_type != FORMAT_DESCRIPTION_EVENT &&
         event_type != ROTATE_EVENT &&
         event_type != STOP_EVENT))
    {
      mi->set_master_log_pos(mi->get_master_log_pos() + inc_pos);
      memcpy(rli->ign_master_log_name_end, mi->get_master_log_name(), FN_REFLEN);
      DBUG_ASSERT(rli->ign_master_log_name_end[0]);
      rli->ign_master_log_pos_end= mi->get_master_log_pos();
    }
    rli->relay_log.signal_update(); // the slave SQL thread needs to re-check
    DBUG_PRINT("info", ("master_log_pos: %lu, event originating from %u server, ignored",
                        (ulong) mi->get_master_log_pos(), uint4korr(buf + SERVER_ID_OFFSET)));
  }
  else
  {
    /* write the event to the relay log */
    if (likely(rli->relay_log.append_buffer(buf, event_len, mi) == 0))
    {
      mi->set_master_log_pos(mi->get_master_log_pos() + inc_pos);
      DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->get_master_log_pos()));
      rli->relay_log.harvest_bytes_written(&rli->log_space_total);

      if (event_type == GTID_LOG_EVENT)
      {
        global_sid_lock->rdlock();
        int ret= rli->add_logged_gtid(gtid.sidno, gtid.gno);
        if (!ret)
          rli->set_last_retrieved_gtid(gtid);
        global_sid_lock->unlock();
        if (ret != 0)
          goto err;
      }
    }
    else
    {
      error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
    }
    rli->ign_master_log_name_end[0]= 0; // last event is not ignored
    if (save_buf != NULL)
      buf= save_buf;
  }
  mysql_mutex_unlock(log_lock);

skip_relay_logging:
  
err:
  if (unlock_data_lock)
    mysql_mutex_unlock(&mi->data_lock);
  DBUG_PRINT("info", ("error: %d", error));
  if (error)
    mi->report(ERROR_LEVEL, error, ER(error), 
               (error == ER_SLAVE_RELAY_LOG_WRITE_FAILURE)?
               "could not queue event from master" :
               error_msg.ptr());
  DBUG_RETURN(error);
}