cluster: split timeout host endpoint handling

Author: dkropachev

cassandra/cluster.py

@@ -4601,7 +4601,13 @@ def _on_timeout(self, _attempts=0):
         errors = self._errors
         if not errors:
             if self.is_schema_agreed:
-                key = str(self._get_host_endpoint(self._current_host)) if self._current_host else 'no host queried before timeout'
+                if self._current_host is None:
+                    key = 'no host queried before timeout'
+                elif self._connection is not None and self._connection.is_control_connection:
+                    control_host = self.session.cluster.get_control_connection_host()
+                    key = str(control_host.endpoint) if control_host is not None else str(self._connection.endpoint)
+                else:
+                    key = str(self._current_host.endpoint)
                 errors = {key: "Client request timeout. See Session.execute[_async](timeout)"}
             else:
                 connection = self.session.cluster.control_connection._connection
@@ -4669,10 +4675,6 @@ def send_request(self, error_no_hosts=True):
                 "Unable to complete the operation against any hosts", self._errors))
         return False
 
-    @staticmethod
-    def _get_host_endpoint(host):
-        return getattr(host, 'endpoint', host)
-
     def _has_usable_node_pool(self):
         try:
             pools = tuple(self.session._pools.values())

tests/unit/test_response_future.py

@@ -644,6 +644,32 @@ def send_msg(message, request_id, cb, **kwargs):
         with pytest.raises(OperationTimedOut):
             rf.result()
 
+    def test_control_connection_fallback_timeout_without_metadata_host_uses_connection_endpoint(self):
+        session = self.make_basic_session()
+        session.cluster.allow_control_connection_query_fallback = ControlConnectionQueryFallback.Fallback
+        session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1']
+        session._pools = {}
+        session.cluster.get_control_connection_host.return_value = None
+        connection = self.make_control_connection()
+        session.cluster.control_connection._connection = connection
+
+        def send_msg(message, request_id, cb, **kwargs):
+            connection._requests[request_id] = (cb, kwargs.get('decoder'), kwargs.get('result_metadata'))
+            return 128
+
+        connection.send_msg.side_effect = send_msg
+
+        rf = self.make_response_future(session)
+        rf.send_request()
+        rf._on_timeout()
+
+        with pytest.raises(OperationTimedOut) as exc_info:
+            rf.result()
+
+        assert exc_info.value.errors == {
+            'control-host': 'Client request timeout. See Session.execute[_async](timeout)'
+        }
+
     def test_first_pool_shutdown(self):
         session = self.make_basic_session()
         session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1', 'ip2']