MDEV-4485: Incorrect error handling in record_gtid().

Fix the error handling when access to the table mysql.gtid_slave_pos
fails for whatever reason. Add some test cases.

mysql-test/suite/rpl/r/rpl_gtid_mdev4484.result

+include/master-slave.inc
+[connection master]
+CREATE TABLE t1 (i int) ENGINE=InnoDB;
+*** MDEV-4484, incorrect error handling when entries in gtid_slave_pos not found. ***
+TRUNCATE TABLE mysql.gtid_slave_pos;
+INSERT INTO t1 VALUES (1);
+include/stop_slave.inc
+SET @old_dbug= @@GLOBAL.debug_dbug;
+SET GLOBAL debug_dbug="+d,gtid_slave_pos_simulate_failed_delete";
+include/start_slave.inc
+SET sql_log_bin= 0;
+CALL mtr.add_suppression("Can't find file");
+SET sql_log_bin= 1;
+INSERT INTO t1 VALUES (2);
+include/wait_for_slave_sql_error.inc [errno=1942]
+STOP SLAVE IO_THREAD;
+SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
+ORDER BY domain_id, sub_id DESC LIMIT 1;
+domain_id	server_id	seq_no
+0	1	3
+SET GLOBAL debug_dbug= @old_dbug;
+include/start_slave.inc
+INSERT INTO t1 VALUES (3);
+SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
+ORDER BY domain_id, sub_id DESC LIMIT 1;
+domain_id	server_id	seq_no
+0	1	4
+SELECT * FROM t1 ORDER BY i;
+i
+1
+2
+3
+DROP TABLE t1;
+include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_gtid_mdev4484.test

+--source include/master-slave.inc
+--source include/have_innodb.inc
+--source include/have_debug.inc
+
+CREATE TABLE t1 (i int) ENGINE=InnoDB;
+
+--sync_slave_with_master
+
+--echo *** MDEV-4484, incorrect error handling when entries in gtid_slave_pos not found. ***
+TRUNCATE TABLE mysql.gtid_slave_pos;
+
+--connection master
+INSERT INTO t1 VALUES (1);
+--sync_slave_with_master
+
+# Inject an artificial error deleting entries, and check that the error handling code works.
+--connection slave
+--source include/stop_slave.inc
+SET @old_dbug= @@GLOBAL.debug_dbug;
+SET GLOBAL debug_dbug="+d,gtid_slave_pos_simulate_failed_delete";
+--source include/start_slave.inc
+SET sql_log_bin= 0;
+CALL mtr.add_suppression("Can't find file");
+SET sql_log_bin= 1;
+
+--connection master
+INSERT INTO t1 VALUES (2);
+
+--connection slave
+--let $slave_sql_errno= 1942
+--source include/wait_for_slave_sql_error.inc
+STOP SLAVE IO_THREAD;
+SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
+ ORDER BY domain_id, sub_id DESC LIMIT 1;
+SET GLOBAL debug_dbug= @old_dbug;
+--source include/start_slave.inc
+
+--connection master
+INSERT INTO t1 VALUES (3);
+--sync_slave_with_master
+
+--connection slave
+SELECT domain_id, server_id, seq_no FROM mysql.gtid_slave_pos
+ ORDER BY domain_id, sub_id DESC LIMIT 1;
+SELECT * FROM t1 ORDER BY i;
+
+
+# Clean up
+--connection master
+DROP TABLE t1;
+
+--source include/rpl_end.inc

sql/rpl_gtid.cc

@@ -179,6 +179,22 @@ rpl_slave_state::get_element(uint32 domain_id)
 }
 
 
+int
+rpl_slave_state::put_back_list(uint32 domain_id, list_element *list)
+{
+  element *e;
+  if (!(e= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0)))
+    return 1;
+  while (list)
+  {
+    list_element *next= list->next;
+    e->add(list);
+    list= next;
+  }
+  return 0;
+}
+
+
 int
 rpl_slave_state::truncate_state_table(THD *thd)
 {
@@ -330,7 +346,10 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
   table->field[3]->store(gtid->seq_no, true);
   DBUG_EXECUTE_IF("inject_crash_before_write_rpl_slave_state", DBUG_SUICIDE(););
   if ((err= table->file->ha_write_row(table->record[0])))
-      goto end;
+  {
+    table->file->print_error(err, MYF(0));
+    goto end;
+  }
 
   lock();
   if ((elem= get_element(gtid->domain_id)) == NULL)
@@ -351,23 +370,42 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
   bitmap_set_bit(table->read_set, table->field[1]->field_index);
 
   if ((err= table->file->ha_index_init(0, 0)))
+  {
+    table->file->print_error(err, MYF(0));
     goto end;
+  }
   while (elist)
   {
     uchar key_buffer[4+8];
 
+    DBUG_EXECUTE_IF("gtid_slave_pos_simulate_failed_delete",
+                    { err= ENOENT;
+                      table->file->print_error(err, MYF(0));
+                      /* `break' does not work in DBUG_EXECUTE_IF */
+                      goto dbug_break; });
+
     next= elist->next;
 
     table->field[1]->store(elist->sub_id, true);
     /* domain_id is already set in table->record[0] from write_row() above. */
     key_copy(key_buffer, table->record[0], &table->key_info[0], 0, false);
-    if ((err= table->file->ha_index_read_map(table->record[1], key_buffer,
-                                             HA_WHOLE_KEY, HA_READ_KEY_EXACT)) ||
-        (err= table->file->ha_delete_row(table->record[1])))
-      break;
+    if (table->file->ha_index_read_map(table->record[1], key_buffer,
+                                       HA_WHOLE_KEY, HA_READ_KEY_EXACT))
+      /* We cannot find the row, assume it is already deleted. */
+      ;
+    else if ((err= table->file->ha_delete_row(table->record[1])))
+      table->file->print_error(err, MYF(0));
+    /*
+      In case of error, we still discard the element from the list. We do
+      not want to endlessly error on the same element in case of table
+      corruption or such.
+    */
     my_free(elist);
     elist= next;
+    if (err)
+      break;
   }
+IF_DBUG(dbug_break:, )
   table->file->ha_index_end();
 
   if(!err && opt_bin_log &&
@@ -382,9 +420,16 @@ end:
     if (err)
     {
       /*
-        ToDo: If error, we need to put any remaining elist back into the HASH so
-        we can do another delete attempt later.
+        If error, we need to put any remaining elist back into the HASH so we
+        can do another delete attempt later.
       */
+      if (elist)
+      {
+        lock();
+        put_back_list(gtid->domain_id, elist);
+        unlock();
+      }
+
       ha_rollback_trans(thd, FALSE);
       close_thread_tables(thd);
     }

sql/rpl_gtid.h

@@ -103,6 +103,7 @@ struct rpl_slave_state
   void unlock() { DBUG_ASSERT(inited); mysql_mutex_unlock(&LOCK_slave_state); }
 
   element *get_element(uint32 domain_id);
+  int put_back_list(uint32 domain_id, list_element *list);
 
   void update_state_hash(uint64 sub_id, rpl_gtid *gtid);
   int record_and_update_gtid(THD *thd, Relay_log_info *rli);