TDL-16984: Implement request timeouts

README.md

@@ -77,7 +77,8 @@ Create a config file containing the database connection credentials, e.g.:
   "host": "localhost",
   "port": "3306",
   "user": "root",
-  "password": "password"
+  "password": "password",
+  "request_timeout": 300
 }
 ```
 

tap_mysql/__init__.py

@@ -719,6 +719,17 @@ def main():
     os.environ['TZ'] = 'UTC'
 
     mysql_conn = MySQLConnection(args.config)
+
+    # add timeout error decorator on 'cursor.execute'
+    # In connection.py's 'make_connection_wrapper', the 'cursorclass' in config is set to the value from kwargs and in binlog.py's,
+    # 'sync_binlog_stream' when initializing 'BinLogStreamReader' we are passing connection_settings={}, as per the code at:
+    #       https://github.com/noplay/python-mysql-replication/blob/main/pymysqlreplication/binlogstream.py#L282
+    # the 'self.__connection_settings' will be {} and hence default cursor 'pymysql.cursors.SSCursor.execute' will be set
+    pymysql.cursors.SSCursor.execute = common.backoff_timeout_error(pymysql.cursors.SSCursor.execute)
+    # add decorator for 'cursorclass' from config
+    if args.config.get("cursorclass"):
+        args.config.get("cursorclass").execute = common.backoff_timeout_error(args.config.get("cursorclass").execute)
+
     log_server_params(mysql_conn)
 
     if args.discover:

tap_mysql/connection.py

@@ -16,6 +16,27 @@
 # We need to hold onto this for self-signed SSL
 match_hostname = ssl.match_hostname
 
+def get_request_timeout():
+    args = singer.utils.parse_args([])
+    # get the value of request timeout from config
+    config_request_timeout = args.config.get("request_timeout")
+
+    # return default value if timeout from config is none or empty
+    if not config_request_timeout:
+        return READ_TIMEOUT_SECONDS
+
+    if isinstance(config_request_timeout, int): # pylint: disable=no-else-return
+        # return value from config
+        return config_request_timeout
+    elif isinstance(config_request_timeout, str) and config_request_timeout.isdigit():
+        # return default value if timeout from config is "0" and integer casted value of valid value
+        return int(config_request_timeout) if int(config_request_timeout) else READ_TIMEOUT_SECONDS
+
+    # raise Exception as MySql dose not support float values
+    # Document: https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_net_read_timeout
+    raise Exception("Unsupported value of timeout, please use string or integer type values.")
+
+
 @backoff.on_exception(backoff.expo,
                       (pymysql.err.OperationalError),
                       max_tries=5,
@@ -36,7 +57,7 @@ def connect_with_backoff(connection):
              warnings.append('Could not set session.wait_timeout. Error: ({}) {}'.format(*e.args))
 
         try:
-            cur.execute("SET @@session.net_read_timeout={}".format(READ_TIMEOUT_SECONDS))
+            cur.execute("SET @@session.net_read_timeout={}".format(get_request_timeout()))
         except pymysql.err.InternalError as e:
              warnings.append('Could not set session.net_read_timeout. Error: ({}) {}'.format(*e.args))
 
@@ -91,7 +112,7 @@ def __init__(self, config):
             "port": int(config["port"]),
             "cursorclass": config.get("cursorclass") or pymysql.cursors.SSCursor,
             "connect_timeout": CONNECT_TIMEOUT_SECONDS,
-            "read_timeout": READ_TIMEOUT_SECONDS,
+            "read_timeout": get_request_timeout(),
             "charset": "utf8",
         }
 

tap_mysql/sync_strategies/common.py

@@ -3,6 +3,8 @@
 
 import copy
 import datetime
+import functools
+import backoff
 import singer
 import time
 import tzlocal
@@ -47,6 +49,37 @@ def monkey_patch_date(date_str):
 #--------------------------------------------------------------------------------------------
 #--------------------------------------------------------------------------------------------
 
+# boolean function to check if the error is 'timeout' error or not
+def is_timeout_error():
+    """
+        This function checks whether the URLError contains 'timed out' substring and return boolean
+        values accordingly, to decide whether to backoff or not.
+    """
+    def gen_fn(exc):
+        if str(exc).__contains__('timed out'):
+            # retry if the error string contains 'timed out'
+            return False
+        return True
+
+    return gen_fn
+
+def reconnect(details):
+    # get connection as 1st param will be 'self' and reconnect
+    connection = details.get("args")[0].connection
+    connection.ping(reconnect=True)
+
+def backoff_timeout_error(fnc):
+    @backoff.on_exception(backoff.expo,
+                          (pymysql.err.OperationalError),
+                          giveup=is_timeout_error(),
+                          on_backoff=reconnect,
+                          max_tries=5,
+                          factor=2)
+    @functools.wraps(fnc)
+    def wrapper(*args, **kwargs):
+        return fnc(*args, **kwargs)
+    return wrapper
+
 def escape(string):
     if '`' in string:
         raise Exception("Can't escape identifier {} because it contains a backtick"

tests/nosetests/test_date_types.py

@@ -1,4 +1,5 @@
 import unittest
+from unittest import mock
 import pymysql
 import tap_mysql
 import copy
@@ -36,7 +37,9 @@ def accumulate_singer_messages(message):
 
 class TestDateTypes(unittest.TestCase):
 
-    def setUp(self):
+    @mock.patch("singer.utils.parse_args")
+    def setUp(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         self.conn = test_utils.get_test_connection()
         self.state = {}
 
@@ -85,7 +88,9 @@ def setUp(self):
                                                'version',
                                                singer.utils.now())
 
-    def test_initial_full_table(self):
+    @mock.patch("singer.utils.parse_args")
+    def test_initial_full_table(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         state = {}
         expected_log_file, expected_log_pos = binlog.fetch_current_log_file_and_pos(self.conn)
 

tests/nosetests/test_full_table_interruption.py

@@ -1,5 +1,6 @@
 import copy
 import os
+from unittest import mock
 import pymysql
 import unittest
 import singer
@@ -133,7 +134,9 @@ def init_tables(conn):
 
 
 class BinlogInterruption(unittest.TestCase):
-    def setUp(self):
+    @mock.patch("singer.utils.parse_args")
+    def setUp(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         self.conn = test_utils.get_test_connection()
         self.catalog = init_tables(self.conn)
 
@@ -164,7 +167,9 @@ def setUp(self):
         global SINGER_MESSAGES
         SINGER_MESSAGES.clear()
 
-    def test_table_2_interrupted(self):
+    @mock.patch("singer.utils.parse_args")
+    def test_table_2_interrupted(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         singer.write_message = singer_write_message_no_table_2
 
         state = {}
@@ -291,7 +296,9 @@ def test_table_2_interrupted(self):
         self.assertIsNotNone(table_2_bookmark.get('log_file'))
         self.assertIsNotNone(table_2_bookmark.get('log_pos'))
 
-    def test_table_3_interrupted(self):
+    @mock.patch("singer.utils.parse_args")
+    def test_table_3_interrupted(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         singer.write_message = singer_write_message_no_table_3
 
         state = {}
@@ -424,7 +431,9 @@ def test_table_3_interrupted(self):
         self.assertIsNotNone(table_3_bookmark.get('log_pos'))
 
 class FullTableInterruption(unittest.TestCase):
-    def setUp(self):
+    @mock.patch("singer.utils.parse_args")
+    def setUp(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         self.conn = test_utils.get_test_connection()
         self.catalog = init_tables(self.conn)
 
@@ -448,7 +457,9 @@ def setUp(self):
         global SINGER_MESSAGES
         SINGER_MESSAGES.clear()
 
-    def test_table_2_interrupted(self):
+    @mock.patch("singer.utils.parse_args")
+    def test_table_2_interrupted(self, mocked_parse_args):
+        mocked_parse_args.return_value = test_utils.get_args({})
         singer.write_message = singer_write_message_no_table_2
 
         state = {}