cancel timers before defuncting connection

mambocab · mambocab · commit 8aeb926644bc · 2017-01-17T12:40:51.000-05:00
This allows us to revert some of the fix to PYTHON-367 without the timer heap growing out of control as described on that ticket. We cancel before defuncting to ensure that the timer doesn't hang around while running callbacks. Between this change and the delegation of callbacks to a secondary thread (see commit 5192302, part of the original PYTHON-367 fix), this commit should not cause a regression on PYTHON-367. Note that we don't cancel timers before defuncting and retrying -- we're not done with the logical request and still want that timer to fire. Now that timers are started on ResponseFuture initialization, we actually use the value of timeout in the tests, so we no longer pass that argument, allowing the test to use the default value.
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
@@ -3608,6 +3608,7 @@ def _set_result(self, host, connection, pool, response):
                 # we got some other kind of response message
                 msg = "Got unexpected message: %r" % (response,)
                 exc = ConnectionException(msg, host)
+                self._cancel_timer()
                 self._connection.defunct(exc)
                 self._set_final_exception(exc)
         except Exception as exc:
diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py
@@ -184,7 +184,7 @@ def test_default_legacy(self):
         self._verify_response_future_profile(rf, expected_profile)
 
     def test_default_profile(self):
-        non_default_profile = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(5)])
+        non_default_profile = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(3)])
         cluster = Cluster(execution_profiles={'non-default': non_default_profile})
         session = Session(cluster, hosts=[])
 
@@ -218,7 +218,7 @@ def test_statement_params_override_legacy(self):
         self._verify_response_future_profile(rf, expected_profile)
 
     def test_statement_params_override_profile(self):
-        non_default_profile = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(5)])
+        non_default_profile = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(3)])
         cluster = Cluster(execution_profiles={'non-default': non_default_profile})
         session = Session(cluster, hosts=[])
 
@@ -284,15 +284,15 @@ def test_no_legacy_with_profile(self):
 
     def test_profile_name_value(self):
 
-        internalized_profile = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(5)])
+        internalized_profile = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(3)])
         cluster = Cluster(execution_profiles={'by-name': internalized_profile})
         session = Session(cluster, hosts=[])
         self.assertEqual(cluster._config_mode, _ConfigMode.PROFILES)
 
         rf = session.execute_async("query", execution_profile='by-name')
         self._verify_response_future_profile(rf, internalized_profile)
 
-        by_value = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(5)])
+        by_value = ExecutionProfile(RoundRobinPolicy(), *[object() for _ in range(3)])
         rf = session.execute_async("query", execution_profile=by_value)
         self._verify_response_future_profile(rf, by_value)
 
