From f4569b9b1df0da425ca166d8f62d44559763a8b0 Mon Sep 17 00:00:00 2001 From: Bret McGuire Date: Wed, 8 Jun 2022 16:55:10 -0500 Subject: [PATCH 1/2] Add linkage to conviction policy to timeout handler in ResponseFuture --- cassandra/cluster.py | 4 ++- tests/unit/test_response_future.py | 51 +++++++++++++++++++++++++++--- 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index c836fb4302..f37c52fadb 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -4383,7 +4383,9 @@ def _on_timeout(self, _attempts=0): host = str(connection.endpoint) if connection else 'unknown' errors = {host: "Request timed out while waiting for schema agreement. See Session.execute[_async](timeout) and Cluster.max_schema_agreement_wait."} - self._set_final_exception(OperationTimedOut(errors, self._current_host)) + final_exc = OperationTimedOut(errors, self._current_host) + self.session.cluster.signal_connection_failure(self._current_host, final_exc, is_host_addition=False) + self._set_final_exception(final_exc) def _on_speculative_execute(self): self._timer = None diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index dbd8764ad9..cff568b336 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -19,8 +19,8 @@ from mock import Mock, MagicMock, ANY from cassandra import ConsistencyLevel, Unavailable, SchemaTargetType, SchemaChangeType, OperationTimedOut -from cassandra.cluster import Session, ResponseFuture, NoHostAvailable, ProtocolVersion -from cassandra.connection import Connection, ConnectionException +from cassandra.cluster import Session, ResponseFuture, NoHostAvailable, ProtocolVersion, Cluster +from cassandra.connection import Connection, ConnectionException, DefaultEndPoint from cassandra.protocol import (ReadTimeoutErrorMessage, WriteTimeoutErrorMessage, UnavailableErrorMessage, ResultMessage, QueryMessage, OverloadedErrorMessage, IsBootstrappingErrorMessage, @@ -28,8 +28,8 @@ RESULT_KIND_ROWS, RESULT_KIND_SET_KEYSPACE, RESULT_KIND_SCHEMA_CHANGE, RESULT_KIND_PREPARED, ProtocolHandler) -from cassandra.policies import RetryPolicy -from cassandra.pool import NoConnectionsAvailable +from cassandra.policies import RetryPolicy, ConvictionPolicy +from cassandra.pool import NoConnectionsAvailable, Host from cassandra.query import SimpleStatement @@ -160,6 +160,49 @@ def test_heartbeat_defunct_deadlock(self): rf._on_timeout() self.assertRaisesRegexp(OperationTimedOut, "Connection defunct by heartbeat", rf.result) + def test_timeout_updates_conviction_policy(self): + """ + PYTHON-539 + + Timeouts from ResponseFuture should notify the host's conviction policy, giving + the driver a mechanism to take action on repeated/systemic timeouts. + """ + conviction_policy = MagicMock(spec=ConvictionPolicy) + conviction_policy.add_failure.return_value = False + + host = Host(DefaultEndPoint("ip1"), lambda h: conviction_policy) + + connection = MagicMock(spec=Connection) + connection._requests = {1:False} + connection.orphaned_request_ids = set() + connection.orphaned_threshold = 2 + + pool = Mock() + pool.is_shutdown = False + pool.borrow_connection.return_value = [connection, 1] + + session = self.make_basic_session() + session.cluster._default_load_balancing_policy.make_query_plan.return_value = [host] + session._pools.get.return_value = pool + + # An extra bit of connective tissue. session.cluster is a mock but we want to use the + # actual impl in Cluster in order to get into the host (and from there to the conviction + # policy). As of this writing Cluster.signal_connection_failure is effectively static + # if the return value from add_failure() on the conviction poilcy is false so we can + # create this linkage _using the actual impl in Cluster_ via a mock side_effect. + def foo(*args, **kwargs): + Cluster.signal_connection_failure(None, *args, **kwargs) + session.cluster.signal_connection_failure.side_effect = foo + + query = SimpleStatement("SELECT * FROM foo") + message = QueryMessage(query=query, consistency_level=ConsistencyLevel.ONE) + + rf = ResponseFuture(session, message, query, 1) + rf.send_request() + rf._on_timeout() + + host.conviction_policy.add_failure.assert_called_once() + def test_read_timeout_error_message(self): session = self.make_session() query = SimpleStatement("SELECT * FROM foo") From 4b88eb9e1ac645b2096d6335e4704ae7ab679835 Mon Sep 17 00:00:00 2001 From: Bret McGuire Date: Thu, 9 Jun 2022 15:24:12 -0500 Subject: [PATCH 2/2] Python 2.7 complains if we pass anything other than a Cluster object when calling Cluster.signal_connection_failure (which we do in the unit test). --- tests/unit/test_response_future.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index cff568b336..e6689da5e4 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -191,7 +191,7 @@ def test_timeout_updates_conviction_policy(self): # if the return value from add_failure() on the conviction poilcy is false so we can # create this linkage _using the actual impl in Cluster_ via a mock side_effect. def foo(*args, **kwargs): - Cluster.signal_connection_failure(None, *args, **kwargs) + Cluster.signal_connection_failure(Cluster(), *args, **kwargs) session.cluster.signal_connection_failure.side_effect = foo query = SimpleStatement("SELECT * FROM foo")