Introduce metadata_request_timeout configuration option

This option allows user to control timeout for driver internal queries.
Idea is to make driver queries more resilient and being independent of
user queries.

Author: dkropachev

cassandra/cluster.py

@@ -19,6 +19,7 @@
 from __future__ import absolute_import
 
 import atexit
+import datetime
 from binascii import hexlify
 from collections import defaultdict
 from collections.abc import Mapping
@@ -1033,6 +1034,12 @@ def default_retry_policy(self, policy):
     or to disable the shardaware port (advanced shardaware)
     """
 
+    metadata_request_timeout = datetime.timedelta(seconds=2)
+    """
+    Timeout for all queries used by driver it self.
+    Supported only by Scylla clusters.
+    """
+
     @property
     def schema_metadata_enabled(self):
         """
@@ -1148,7 +1155,9 @@ def __init__(self,
                  client_id=None,
                  cloud=None,
                  scylla_cloud=None,
-                 shard_aware_options=None):
+                 shard_aware_options=None,
+                 metadata_request_timeout=None,
+                 ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
         extablishing connection pools or refreshing metadata.
@@ -1240,6 +1249,8 @@ def __init__(self,
         self.no_compact = no_compact
 
         self.auth_provider = auth_provider
+        if metadata_request_timeout is not None:
+            self.metadata_request_timeout = metadata_request_timeout
 
         if load_balancing_policy is not None:
             if isinstance(load_balancing_policy, type):
@@ -3549,6 +3560,7 @@ class PeersQueryType(object):
     _is_shutdown = False
     _timeout = None
     _protocol_version = None
+    _metadata_request_timeout = None
 
     _schema_event_refresh_window = None
     _topology_event_refresh_window = None
@@ -3648,7 +3660,7 @@ def _reconnect_internal(self):
         (conn, _) = self._connect_host_in_lbp()
         if conn is not None:
             return conn
-        
+
         # Try to re-resolve hostnames as a fallback when all hosts are unreachable
         self._cluster._resolve_hostnames()
 
@@ -3693,7 +3705,10 @@ def _try_connect(self, host):
         # If sharding information is available, it's a ScyllaDB cluster, so do not use peers_v2 table.
         if connection.features.sharding_info is not None:
             self._uses_peers_v2 = False
-        
+
+        # Cassandra does not support "USING TIMEOUT"
+        self._metadata_request_timeout = None if connection.features.sharding_info is None \
+            else datetime.timedelta(seconds=self._cluster.control_connection_timeout)
         self._tablets_routing_v1 = connection.features.tablets_routing_v1
 
         # use weak references in both directions
@@ -3830,7 +3845,12 @@ def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_w
             log.debug("Skipping schema refresh due to lack of schema agreement")
             return False
 
-        self._cluster.metadata.refresh(connection, self._timeout, fetch_size=self._schema_meta_page_size, **kwargs)
+        self._cluster.metadata.refresh(
+            connection,
+            self._timeout,
+            fetch_size=self._schema_meta_page_size,
+            metadata_request_timeout=self._metadata_request_timeout,
+            **kwargs)
 
         return True
 

cassandra/metadata.py

@@ -134,11 +134,12 @@ def export_schema_as_string(self):
         """
         return "\n\n".join(ks.export_as_string() for ks in self.keyspaces.values())
 
-    def refresh(self, connection, timeout, target_type=None, change_type=None, fetch_size=None, **kwargs):
+    def refresh(self, connection, timeout, target_type=None, change_type=None, fetch_size=None,
+                metadata_request_timeout=None, **kwargs):
 
         server_version = self.get_host(connection.original_endpoint).release_version
         dse_version = self.get_host(connection.original_endpoint).dse_version
-        parser = get_schema_parser(connection, server_version, dse_version, timeout, fetch_size)
+        parser = get_schema_parser(connection, server_version, dse_version, timeout, metadata_request_timeout, fetch_size)
 
         if not target_type:
             self._rebuild_all(parser)
@@ -1946,11 +1947,11 @@ def export_as_string(self):
 
 
 class _SchemaParser(object):
-
-    def __init__(self, connection, timeout, fetch_size):
+    def __init__(self, connection, timeout, fetch_size, metadata_request_timeout):
         self.connection = connection
         self.timeout = timeout
         self.fetch_size = fetch_size
+        self.metadata_request_timeout = metadata_request_timeout
 
     def _handle_results(self, success, result, expected_failures=tuple(), query_msg=None, timeout=None):
         """
@@ -2054,8 +2055,8 @@ class SchemaParserV22(_SchemaParser):
         "compression",
         "default_time_to_live")
 
-    def __init__(self, connection, timeout, fetch_size):
-        super(SchemaParserV22, self).__init__(connection, timeout, fetch_size)
+    def __init__(self, connection, timeout, fetch_size, metadata_request_timeout):
+        super(SchemaParserV22, self).__init__(connection, timeout, fetch_size, metadata_request_timeout)
         self.keyspaces_result = []
         self.tables_result = []
         self.columns_result = []
@@ -2575,8 +2576,8 @@ class SchemaParserV3(SchemaParserV22):
         'read_repair_chance',
         'speculative_retry')
 
-    def __init__(self, connection, timeout, fetch_size):
-        super(SchemaParserV3, self).__init__(connection, timeout, fetch_size)
+    def __init__(self, connection, timeout, fetch_size, metadata_request_timeout):
+        super(SchemaParserV3, self).__init__(connection, timeout, fetch_size, metadata_request_timeout)
         self.indexes_result = []
         self.keyspace_table_index_rows = defaultdict(lambda: defaultdict(list))
         self.keyspace_view_rows = defaultdict(list)
@@ -2860,8 +2861,8 @@ class SchemaParserV4(SchemaParserV3):
     _SELECT_VIRTUAL_TABLES = 'SELECT * from system_virtual_schema.tables'
     _SELECT_VIRTUAL_COLUMNS = 'SELECT * from system_virtual_schema.columns'
 
-    def __init__(self, connection, timeout, fetch_size):
-        super(SchemaParserV4, self).__init__(connection, timeout, fetch_size)
+    def __init__(self, connection, timeout, fetch_size, metadata_request_timeout):
+        super(SchemaParserV4, self).__init__(connection, timeout, fetch_size, metadata_request_timeout)
         self.virtual_keyspaces_rows = defaultdict(list)
         self.virtual_tables_rows = defaultdict(list)
         self.virtual_columns_rows = defaultdict(lambda: defaultdict(list))
@@ -2995,8 +2996,8 @@ class SchemaParserDSE68(SchemaParserDSE67):
 
     _table_metadata_class = TableMetadataDSE68
 
-    def __init__(self, connection, timeout, fetch_size):
-        super(SchemaParserDSE68, self).__init__(connection, timeout, fetch_size)
+    def __init__(self, connection, timeout, fetch_size, metadata_request_timeout):
+        super(SchemaParserDSE68, self).__init__(connection, timeout, fetch_size, metadata_request_timeout)
         self.keyspace_table_vertex_rows = defaultdict(lambda: defaultdict(list))
         self.keyspace_table_edge_rows = defaultdict(lambda: defaultdict(list))
 
@@ -3361,25 +3362,25 @@ def __init__(
         self.to_clustering_columns = to_clustering_columns
 
 
-def get_schema_parser(connection, server_version, dse_version, timeout, fetch_size=None):
+def get_schema_parser(connection, server_version, dse_version, timeout, metadata_request_timeout, fetch_size=None):
     version = Version(server_version)
     if dse_version:
         v = Version(dse_version)
         if v >= Version('6.8.0'):
-            return SchemaParserDSE68(connection, timeout, fetch_size)
+            return SchemaParserDSE68(connection, timeout, fetch_size, metadata_request_timeout)
         elif v >= Version('6.7.0'):
-            return SchemaParserDSE67(connection, timeout, fetch_size)
+            return SchemaParserDSE67(connection, timeout, fetch_size, metadata_request_timeout)
         elif v >= Version('6.0.0'):
-            return SchemaParserDSE60(connection, timeout, fetch_size)
+            return SchemaParserDSE60(connection, timeout, fetch_size, metadata_request_timeout)
 
     if version >= Version('4-a'):
-        return SchemaParserV4(connection, timeout, fetch_size)
+        return SchemaParserV4(connection, timeout, fetch_size, metadata_request_timeout)
     elif version >= Version('3.0.0'):
-        return SchemaParserV3(connection, timeout, fetch_size)
+        return SchemaParserV3(connection, timeout, fetch_size, metadata_request_timeout)
     else:
         # we could further specialize by version. Right now just refactoring the
         # multi-version parser we have as of C* 2.2.0rc1.
-        return SchemaParserV22(connection, timeout, fetch_size)
+        return SchemaParserV22(connection, timeout, fetch_size, metadata_request_timeout)
 
 
 def _cql_from_cass_type(cass_type):

tests/integration/standard/test_metadata.py

@@ -243,7 +243,8 @@ def test_basic_table_meta_properties(self):
             cc,
             self.cluster.metadata.get_host(cc.host).release_version,
             self.cluster.metadata.get_host(cc.host).dse_version,
-            1
+            1,
+            None,
         )
 
         for option in tablemeta.options:
@@ -1968,7 +1969,8 @@ def setup_class(cls):
             connection,
             cls.cluster.metadata.get_host(connection.host).release_version,
             cls.cluster.metadata.get_host(connection.host).dse_version,
-            timeout=20
+            20,
+            None,
         ).__class__
         cls.cluster.control_connection.reconnect = Mock()
 

tests/unit/test_metadata.py

@@ -618,7 +618,7 @@ def test_build_index_as_cql(self):
         column_meta.table.name = 'table_name_here'
         column_meta.table.keyspace_name = 'keyspace_name_here'
         column_meta.table.columns = {column_meta.name: column_meta}
-        parser = get_schema_parser(Mock(), '2.1.0', None, 0.1)
+        parser = get_schema_parser(Mock(), '2.1.0', None, 0.1, None)
 
         row = {'index_name': 'index_name_here', 'index_type': 'index_type_here'}
         index_meta = parser._build_index_metadata(column_meta, row)