BUG20653441: Fix connection hanging when a query is killed

peeyushgupta · peeyushgupta · commit 6320cf85abb4 · 2015-06-22T14:48:34.000+05:30
Connection was hanging when a query is killed from a different
connection. We fix this by correctly reading packets from MySQL
server which contain error information.

A test case has been added for BUG#20653441
diff --git a/lib/mysql/connector/errors.py b/lib/mysql/connector/errors.py
@@ -143,8 +143,11 @@ def get_exception(packet):
     """
     errno = errmsg = None
 
-    if packet[4] != 255:
-        raise ValueError("Packet is not an error packet")
+    try:
+        if packet[4] != 255:
+            raise ValueError("Packet is not an error packet")
+    except IndexError as err:
+        return InterfaceError("Failed getting Error information (%r)" % err)
 
     sqlstate = None
     try:
diff --git a/lib/mysql/connector/protocol.py b/lib/mysql/connector/protocol.py
@@ -33,6 +33,7 @@
 from . import errors, utils
 from .authentication import get_auth_plugin
 from .catch23 import PY2, struct_unpack
+from .errors import get_exception
 
 
 class MySQLProtocol(object):
@@ -335,8 +336,10 @@ def read_text_result(self, sock, count=1):
                 rowdata = utils.read_lc_string_list(packet[4:])
             if eof is None and rowdata is not None:
                 rows.append(rowdata)
+            elif eof is None and rowdata is None:
+                raise get_exception(packet)
             i += 1
-        return (rows, eof)
+        return rows, eof
 
     def _parse_binary_integer(self, packet, field):
         """Parse an integer from a binary packet"""
@@ -466,6 +469,8 @@ def read_binary_result(self, sock, columns, count=1):
                 values = self._parse_binary_values(columns, packet[5:])
             if eof is None and values is not None:
                 rows.append(values)
+            elif eof is None and values is None:
+                raise get_exception(packet)
             i += 1
         return (rows, eof)
 
diff --git a/tests/test_bugs.py b/tests/test_bugs.py
@@ -3578,3 +3578,75 @@ def test_set(self):
             exp.append((row.id, row.name, row.dept))
         self.assertEqual(exp, data[1:])
         cur.close()
+
+
+class BugOra20653441(tests.MySQLConnectorTests):
+
+    """BUG#20653441: PYTHON CONNECTOR HANGS IF A QUERY IS KILLED (ERROR 1317)"""
+
+    def setUp(self):
+        self.table_name = 'Bug20653441'
+
+    def _setup(self):
+        self.cnx.cmd_query("DROP TABLE IF EXISTS {0}".format(self.table_name))
+        table = (
+            "CREATE TABLE {table} ("
+            " id INT UNSIGNED NOT NULL AUTO_INCREMENT,"
+            " c1 VARCHAR(255) DEFAULT '{default}',"
+            " PRIMARY KEY (id)"
+            ")"
+        ).format(table=self.table_name, default='a' * 255)
+        self.cnx.cmd_query(table)
+
+        stmt = "INSERT INTO {table} (id) VALUES {values}".format(
+            table=self.table_name,
+            values=','.join(['(NULL)'] * 1024)
+        )
+        self.cnx.cmd_query(stmt)
+
+    def tearDown(self):
+        try:
+            cnx = connection.MySQLConnection(**tests.get_mysql_config())
+            cnx.cmd_query(
+                "DROP TABLE IF EXISTS {0}".format(self.table_name))
+            cnx.close()
+        except:
+            pass
+
+    @foreach_cnx()
+    def test_kill_query(self):
+        self._setup()
+
+        def kill(connection_id):
+            """Kill query using separate connection"""
+            killer = connection.MySQLConnection(**tests.get_mysql_config())
+            time.sleep(1)
+            killer.cmd_query("KILL QUERY {0}".format(connection_id))
+            killer.close()
+
+        def sleepy_select(cnx):
+            """Execute a SELECT statement which takes a while to complete"""
+            cur = cnx.cursor()
+            # Ugly query ahead!
+            stmt = "SELECT x1.*, x2.* from {table} as x1, {table} as x2".format(
+                table=self.table_name)
+            cur.execute(stmt)
+            # Save the error so we can check in the calling thread
+            cnx.test_error = None
+
+            try:
+                cur.fetchall()
+            except errors.Error as err:
+                cnx.test_error = err
+
+        worker = Thread(target=sleepy_select, args=[self.cnx])
+        killer = Thread(target=kill, args=[self.cnx.connection_id])
+        worker.start()
+        killer.start()
+        worker.join()
+        killer.join()
+
+        self.assertTrue(isinstance(self.cnx.test_error, errors.DatabaseError))
+        self.assertEqual(str(self.cnx.test_error),
+                         "1317 (70100): Query execution was interrupted")
+        self.cnx.close()
