correctness assert

thd_get_ha_data() can be used without a lock, but only from the
current thd thread, when calling from anoher thread it *must*
be protected by thd->LOCK_thd_data

* fix group commit code to take thd->LOCK_thd_data
* remove innobase_close_connection() from the innodb background thread,
  it's not needed after 8777540 and was failing the assert with
  current_thd==0

include/my_pthread.h

@@ -437,6 +437,7 @@ void safe_mutex_free_deadlock_data(safe_mutex_t *mp);
 #define my_cond_wait(A,B) safe_cond_wait((A), (B), __FILE__, __LINE__)
 #else
 
+#define safe_mutex_is_owner(mp) (1)
 #define safe_mutex_assert_owner(mp) do {} while (0)
 #define safe_mutex_assert_not_owner(mp) do {} while (0)
 #define safe_mutex_setflags(mp, F) do {} while (0)

sql/log.cc

@@ -8512,7 +8512,11 @@ MYSQL_BIN_LOG::trx_group_commit_leader(group_commit_entry *leader)
     ++num_commits;
     if (current->cache_mngr->using_xa && likely(!current->error) &&
         DBUG_EVALUATE_IF("skip_commit_ordered", 0, 1))
+    {
+      mysql_mutex_lock(&current->thd->LOCK_thd_data);
       run_commit_ordered(current->thd, current->all);
+      mysql_mutex_unlock(&current->thd->LOCK_thd_data);
+    }
     current->thd->wakeup_subsequent_commits(current->error);
 
     /*

sql/sql_class.cc

@@ -462,6 +462,7 @@ void thd_storage_lock_wait(THD *thd, long long value)
 extern "C"
 void *thd_get_ha_data(const THD *thd, const struct handlerton *hton)
 {
+  DBUG_ASSERT(thd == current_thd ||  mysql_mutex_is_owner(&thd->LOCK_thd_data));
   return thd->ha_data[hton->slot].ha_ptr;
 }
 

storage/innobase/handler/ha_innodb.cc

@@ -1484,10 +1484,6 @@ innobase_destroy_background_thd(
 /*============================*/
 	MYSQL_THD thd)
 {
-	/* need to close the connection explicitly, the server won't do it
-	if innodb is in the PLUGIN_IS_DYING state */
-	innobase_close_connection(innodb_hton_ptr, thd);
-	thd_set_ha_data(thd, innodb_hton_ptr, NULL);
 	destroy_background_thd(thd);
 }