From 53c479f4ca7c800bceb3a8a81fefd1f6c21c83c0 Mon Sep 17 00:00:00 2001 From: Nadav Har'El Date: Mon, 26 Oct 2026 13:27:05 +0200 Subject: [PATCH] cluster: report a timeout waiting for schema agreement, not a defunct connection This patch fixes the misleading error of a request that times out while the driver waits for schema agreement after a schema change. The driver reported "Connection defunct by heartbeat", although no connection failed and the server had applied the schema change. When a request's response arrives, the connection removes the request from its queue. For a schema change result, the driver then waits for schema agreement in another thread before it completes the request, and the request's timer keeps running. When the timer fires during that wait, _on_timeout() doesn't find the request in the connection's queue. The KeyError branch for that case was added (PYTHON-1044) for a connection that a heartbeat failure defuncted, which also empties the queue, so it always reported "Connection defunct by heartbeat". The existing message for this case, "Request timed out while waiting for schema agreement", was never reached. The fix checks is_schema_agreed in that branch, which is false only while a schema change result waits for schema agreement, and then reports the schema agreement message. The defunct connection case keeps its message. This misleading message appeared in many Scylla CI failures, for example a slow DROP KEYSPACE in SCT, and sent their analysis looking for connection problems. Fixes SCYLLADB-1317 --- cassandra/cluster.py | 19 +++++++++--- tests/unit/test_response_future.py | 49 ++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 5 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 1962f95a65..e6a6b7d6f7 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -7403,8 +7403,14 @@ def _on_timeout(self, _attempts=0): # We should still raise OperationTimedOut to reject the future so that the main event thread will not # wait for it endlessly except KeyError: - key = "Connection defunct by heartbeat" - errors = {key: "Client request timeout. See Session.execute[_async](timeout)"} + if not self.is_schema_agreed: + # The request was also removed when its response arrived: + # a schema change result, which now waits for schema + # agreement. Report that instead of a defunct connection. + errors = self._schema_agreement_timeout_errors() + else: + key = "Connection defunct by heartbeat" + errors = {key: "Client request timeout. See Session.execute[_async](timeout)"} self._set_final_exception(OperationTimedOut(errors, self._current_host, timeout=self.timeout, in_flight=connection.in_flight)) @@ -7436,14 +7442,17 @@ def _on_timeout(self, _attempts=0): key = str(self._current_host.endpoint) errors = {key: "Client request timeout. See Session.execute[_async](timeout)"} else: - connection = self.session.cluster.control_connection._connection - 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."} + errors = self._schema_agreement_timeout_errors() self._set_final_exception(OperationTimedOut(errors, self._current_host, timeout=self.timeout, in_flight=conn_in_flight)) + def _schema_agreement_timeout_errors(self): + connection = self.session.cluster.control_connection._connection + host = str(connection.endpoint) if connection else 'unknown' + return {host: "Request timed out while waiting for schema agreement. See Session.execute[_async](timeout) and Cluster.max_schema_agreement_wait."} + def _on_speculative_execute(self): self._timer = None if not self._event.is_set(): diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index 1e7a568168..555a9e4cbb 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -749,6 +749,55 @@ def test_heartbeat_defunct_deadlock(self): assert exc_info.value.timeout == 1 assert exc_info.value.in_flight == 5 + def test_timeout_while_waiting_for_schema_agreement(self): + """ + A schema change result removes the request from the connection's request + queue, like a defunct connection does, and then waits for schema agreement. + A timeout during that wait must be reported as a timeout while waiting for + schema agreement - the server did apply the schema change - not as a + connection defunct by heartbeat. + """ + + connection = MagicMock(spec=Connection) + connection._requests = {} + connection.in_flight = 5 + connection.orphaned_request_ids = set() + connection.features.tablets_routing_v2 = False + connection.features.tablets_routing_v1 = False + + 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 = [Mock(), Mock()] + session._pools.get.return_value = pool + + query = SimpleStatement("DROP KEYSPACE ks") + message = QueryMessage(query=query, consistency_level=ConsistencyLevel.ONE) + + rf = ResponseFuture(session, message, query, 1) + rf.send_request() + + # Simulate the arrival of the schema change result: the connection has + # removed the request from its queue, and the result waits for schema + # agreement in session.submit(), which this mock session never runs. + connection._requests = {} + result = Mock(spec=ResultMessage, + kind=RESULT_KIND_SCHEMA_CHANGE, + schema_change_event={'target_type': SchemaTargetType.KEYSPACE, + 'change_type': SchemaChangeType.DROPPED, + 'keyspace': "ks"}) + rf._set_result(None, connection, pool, result) + session.submit.assert_called_once() + + # Simulate ResponseFuture timing out + rf._on_timeout() + with pytest.raises(OperationTimedOut, match="waiting for schema agreement") as exc_info: + rf.result() + assert "Connection defunct by heartbeat" not in str(exc_info.value) + assert exc_info.value.timeout == 1 + def test_read_timeout_error_message(self): session = self.make_session() query = SimpleStatement("SELECT * FROM foo")