[storage/federated] PS-8747: FEDERATED engine not handling wait_timeo…

…ut (percona#5058)

https://jira.percona.com/browse/PS-8747

mysql-test/suite/federated/r/federated_debug.result

@@ -27,6 +27,29 @@ a
 # Drop tables on master and slave
 DROP TABLE t1;
 DROP TABLE t1;
+#
+# PS-8747: Got an error writing communication packets error during FEDERATED engine reconnection
+#
+SET @OLD_SLAVE_WAIT_TIMEOUT= @@GLOBAL.WAIT_TIMEOUT;
+SET @@GLOBAL.WAIT_TIMEOUT= 4;
+CREATE TABLE test.t1 (id int PRIMARY KEY);
+CREATE SERVER test FOREIGN DATA WRAPPER test1 OPTIONS(
+user 'root',
+password '',
+host '127.0.0.1',
+port SLAVE_PORT,
+database 'test');
+CREATE TABLE test.t1 (id int PRIMARY KEY) ENGINE=FEDERATED CONNECTION='test';
+SELECT * FROM test.t1;
+id
+SET @debug_save= @@GLOBAL.DEBUG;
+SET @@GLOBAL.DEBUG='+d,PS-8747_wait_for_disconnect_after_check';
+INSERT INTO test.t1 SELECT tt.* FROM SEQUENCE_TABLE(20000) AS tt;
+SET GLOBAL DEBUG= @debug_save;
+DROP TABLE test.t1;
+DROP SERVER test;
+DROP TABLE test.t1;
+SET @@GLOBAL.WAIT_TIMEOUT= @OLD_SLAVE_WAIT_TIMEOUT;
 # Federated cleanup
 DROP TABLE IF EXISTS federated.t1;
 DROP DATABASE federated;

mysql-test/suite/federated/t/federated_debug.test

@@ -34,6 +34,41 @@ DROP TABLE t1;
 connection slave;
 DROP TABLE t1;
 
+--echo #
+--echo # PS-8747: Got an error writing communication packets error during FEDERATED engine reconnection
+--echo #
+connection slave;
+SET @OLD_SLAVE_WAIT_TIMEOUT= @@GLOBAL.WAIT_TIMEOUT;
+SET @@GLOBAL.WAIT_TIMEOUT= 4;
+CREATE TABLE test.t1 (id int PRIMARY KEY);
+
+connection master;
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval CREATE SERVER test FOREIGN DATA WRAPPER test1 OPTIONS(
+  user 'root',
+  password '',
+  host '127.0.0.1',
+  port $SLAVE_MYPORT,
+  database 'test');
+CREATE TABLE test.t1 (id int PRIMARY KEY) ENGINE=FEDERATED CONNECTION='test';
+
+# The following SELECT will setup internal client-server connection inside Federated SE
+SELECT * FROM test.t1;
+
+SET @debug_save= @@GLOBAL.DEBUG;
+SET @@GLOBAL.DEBUG='+d,PS-8747_wait_for_disconnect_after_check';
+
+# Send data which will not fit into one communication packet, so client will try to send them
+# before flush and reconnection.
+INSERT INTO test.t1 SELECT tt.* FROM SEQUENCE_TABLE(20000) AS tt;
+SET GLOBAL DEBUG= @debug_save;
+
+DROP TABLE test.t1;
+DROP SERVER test;
+connection slave;
+DROP TABLE test.t1;
+SET @@GLOBAL.WAIT_TIMEOUT= @OLD_SLAVE_WAIT_TIMEOUT;
+
 connection default;
 --echo # Federated cleanup
 source suite/federated/include/federated_cleanup.inc;

sql-common/client.cc

@@ -1384,6 +1384,24 @@ bool cli_advanced_command(MYSQL *mysql, enum enum_server_command command,
   if ((command != COM_QUIT) && mysql->reconnect && !vio_is_connected(net->vio))
     net->error = NET_ERROR_SOCKET_UNUSABLE;
 
+  /*
+    If the connection becomes dead after the above check,
+    and the packet we are sending is bigger than 16k (net_buffer_length default)
+    the client will try to send partial data. In such a case net_write_command()
+    will detect connection loss  when trying to send the packet over the network
+    and report the error via my_error() from net_write_raw_loop(). But we don't
+    want this error to be propagated back to the client session.
+    net_write_command() should fail silently and the below reconnection logic
+    should kick in.
+  */
+  DBUG_EXECUTE_IF(
+      "PS-8747_wait_for_disconnect_after_check",
+      DBUG_SET("-d,PS-8747_wait_for_disconnect_after_check");
+      net->error = 0;  // Even if above check detected we are already
+                       // disconnected pretend we know nothing about this.
+      sleep(10);       // MTR test will set interactive_timeout/wait_timeout = 4
+  );
+
   if (net_write_command(net, (uchar)command, header, header_length, arg,
                         arg_length)) {
     DBUG_PRINT("error",

storage/federated/ha_federated.cc

@@ -3067,6 +3067,20 @@ int ha_federated::real_query(const char *query, size_t length) {
 
   if (!query || !length) goto end;
 
+  /* Here we operate on internal proxy -> server connection which uses
+     client code. But it is compiled in context of server code, so
+     all things like:
+     #ifdef MYSQL_SERVER
+         my_error(net->last_errno, MYF(0));
+     #endif
+     are compiled in.
+     proxy -> server connection has reconnection logic, and all internal errors
+     should be handled silently as long as the overal result of the operation
+     is success.
+     If this connection fails and cannot be recovered, the function will return
+     error, which will be handled by the server.
+  */
+
   rc = mysql_real_query(mysql, query, static_cast<ulong>(length));
 
   // Simulate as errors happened within the previous query
@@ -3102,6 +3116,7 @@ int ha_federated::real_query(const char *query, size_t length) {
     Diagnostics_area *da = current_thd->get_stmt_da();
     if (da->is_set()) {
       const uint err = da->mysql_errno();
+
       if ((err == ER_NET_PACKETS_OUT_OF_ORDER || err == ER_NET_ERROR_ON_WRITE ||
            err == ER_NET_WRITE_INTERRUPTED || err == ER_NET_READ_ERROR ||
            err == ER_NET_READ_INTERRUPTED) &&
@@ -3115,6 +3130,18 @@ int ha_federated::real_query(const char *query, size_t length) {
     }
   }
 end:
+  /* Reconnection can take place above or inside cli_advanced_command().
+   If it happens in cli_advanced_command(), diagnositcs are may contain
+   the information about the error which triggered the reconnection.
+   But as the reconnection was fine, the information about transient
+   problems should not be propagated to the client.
+   */
+  if (!rc) {
+    Diagnostics_area *da = current_thd->get_stmt_da();
+    da->reset_condition_info(current_thd);
+    da->reset_diagnostics_area();
+  }
+
   return rc;
 }