Remove protocol versions >5 from MAX_SUPPORTED

Without removing associated code yet, this patch just ensure we begin
negotiating protocol version 5, then fallback to 4, reducing the amount of protocol errors
that the Scylla server sends when it sees the unsupported versions.

Refs: #244

Signed-off-by: Yaniv Kaul <[email protected]>

Author: mykaul

cassandra/__init__.py

@@ -180,9 +180,9 @@ class ProtocolVersion(object):
     DSE private protocol v2, supported in DSE 6.0+
     """
 
-    SUPPORTED_VERSIONS = (DSE_V2, DSE_V1, V6, V5, V4, V3, V2, V1)
+    SUPPORTED_VERSIONS = (V5, V4, V3, V2, V1)
     """
-    A tuple of all supported protocol versions
+    A tuple of all supported protocol versions for ScyllaDB, including future v5 support.
     """
 
     BETA_VERSIONS = (V6,)
@@ -237,10 +237,6 @@ def uses_keyspace_flag(cls, version):
     def has_continuous_paging_support(cls, version):
         return version >= cls.DSE_V1
 
-    @classmethod
-    def has_continuous_paging_next_pages(cls, version):
-        return version >= cls.DSE_V2
-
     @classmethod
     def has_checksumming_support(cls, version):
         return cls.V5 <= version < cls.DSE_V1

cassandra/cluster.py

@@ -51,7 +51,7 @@
 from cassandra.connection import (ConnectionException, ConnectionShutdown,
                                   ConnectionHeartbeat, ProtocolVersionUnsupported,
                                   EndPoint, DefaultEndPoint, DefaultEndPointFactory,
-                                  ContinuousPagingState, SniEndPointFactory, ConnectionBusy)
+                                  SniEndPointFactory, ConnectionBusy)
 from cassandra.cqltypes import UserType
 import cassandra.cqltypes as types
 from cassandra.encoder import Encoder
@@ -672,7 +672,7 @@ class Cluster(object):
     server will be automatically used.
     """
 
-    protocol_version = ProtocolVersion.DSE_V2
+    protocol_version = ProtocolVersion.V5
     """
     The maximum version of the native protocol to use.
 
@@ -2749,7 +2749,6 @@ def __init__(self, cluster, hosts, keyspace=None):
             raise NoHostAvailable(msg, [h.address for h in hosts])
 
         self.session_id = uuid.uuid4()
-        self._graph_paging_available = self._check_graph_paging_available()
 
         if self.cluster.column_encryption_policy is not None:
             try:
@@ -2946,26 +2945,10 @@ def execute_graph_async(self, query, parameters=None, trace=False, execution_pro
     def _maybe_set_graph_paging(self, execution_profile):
         graph_paging = execution_profile.continuous_paging_options
         if execution_profile.continuous_paging_options is _NOT_SET:
-            graph_paging = ContinuousPagingOptions() if self._graph_paging_available else None
+            graph_paging = None
 
         execution_profile.continuous_paging_options = graph_paging
 
-    def _check_graph_paging_available(self):
-        """Verify if we can enable graph paging. This executed only once when the session is created."""
-
-        if not ProtocolVersion.has_continuous_paging_next_pages(self._protocol_version):
-            return False
-
-        for host in self.cluster.metadata.all_hosts():
-            if host.dse_version is None:
-                return False
-
-            version = Version(host.dse_version)
-            if version < _GRAPH_PAGING_MIN_DSE_VERSION:
-                return False
-
-        return True
-
     def _resolve_execution_profile_options(self, execution_profile):
         """
         Determine the GraphSON protocol and row factory for a graph query. This is useful
@@ -3112,13 +3095,7 @@ def _create_response_future(self, query, parameters, trace, custom_payload,
         else:
             timestamp = None
 
-        supports_continuous_paging_state = (
-            ProtocolVersion.has_continuous_paging_next_pages(self._protocol_version)
-        )
-        if continuous_paging_options and supports_continuous_paging_state:
-            continuous_paging_state = ContinuousPagingState(continuous_paging_options.max_queue_size)
-        else:
-            continuous_paging_state = None
+        continuous_paging_state = None
 
         if isinstance(query, SimpleStatement):
             query_string = query.query_string

cassandra/connection.py

@@ -445,32 +445,6 @@ class CrcMismatchException(ConnectionException):
     pass
 
 
-class ContinuousPagingState(object):
-    """
-     A class for specifying continuous paging state, only supported starting with DSE_V2.
-    """
-
-    num_pages_requested = None
-    """
-    How many pages we have already requested
-    """
-
-    num_pages_received = None
-    """
-    How many pages we have already received
-    """
-
-    max_queue_size = None
-    """
-    The max queue size chosen by the user via the options
-    """
-
-    def __init__(self, max_queue_size):
-        self.num_pages_requested = max_queue_size  # the initial query requests max_queue_size
-        self.num_pages_received = 0
-        self.max_queue_size = max_queue_size
-
-
 class ContinuousPagingSession(object):
     def __init__(self, stream_id, decoder, row_factory, connection, state):
         self.stream_id = stream_id

cassandra/protocol.py

@@ -631,8 +631,6 @@ def _write_query_params(self, f, protocol_version):
     def _write_paging_options(self, f, paging_options, protocol_version):
         write_int(f, paging_options.max_pages)
         write_int(f, paging_options.max_pages_per_second)
-        if ProtocolVersion.has_continuous_paging_next_pages(protocol_version):
-            write_int(f, paging_options.max_queue_size)
 
 
 class QueryMessage(_QueryMessage):
@@ -1092,12 +1090,10 @@ def send_body(self, f, protocol_version):
         if self.op_type == ReviseRequestMessage.RevisionType.PAGING_BACKPRESSURE:
             if self.next_pages <= 0:
                 raise UnsupportedOperation("Continuous paging backpressure requires next_pages > 0")
-            elif not ProtocolVersion.has_continuous_paging_next_pages(protocol_version):
+            else:
                 raise UnsupportedOperation(
                     "Continuous paging backpressure may only be used with protocol version "
                     "ProtocolVersion.DSE_V2 or higher. Consider setting Cluster.protocol_version to ProtocolVersion.DSE_V2.")
-            else:
-                write_int(f, self.next_pages)
 
 
 class _ProtocolHandler(object):

tests/integration/__init__.py

@@ -228,15 +228,9 @@ def _get_dse_version_from_cass(cass_version):
 
 def get_default_protocol():
     if CASSANDRA_VERSION >= Version('4.0-a'):
-        if DSE_VERSION:
-            return ProtocolVersion.DSE_V2
-        else:
-            return ProtocolVersion.V5
+        return ProtocolVersion.V5
     if CASSANDRA_VERSION >= Version('3.10'):
-        if DSE_VERSION:
-            return ProtocolVersion.DSE_V1
-        else:
-            return 4
+        return 4
     if CASSANDRA_VERSION >= Version('2.2'):
         return 4
     elif CASSANDRA_VERSION >= Version('2.1'):
@@ -274,15 +268,9 @@ def get_supported_protocol_versions():
         if not DSE_VERSION:
             return (3, 4, 5, 6)
     if CASSANDRA_VERSION >= Version('4.0-a'):
-        if DSE_VERSION:
-            return (3, 4, ProtocolVersion.DSE_V1, ProtocolVersion.DSE_V2)
-        else:
-            return (3, 4, 5)
+        return (3, 4, 5)
     elif CASSANDRA_VERSION >= Version('3.10'):
-        if DSE_VERSION:
-            return (3, 4, ProtocolVersion.DSE_V1)
-        else:
-            return (3, 4)
+        return (3, 4)
     elif CASSANDRA_VERSION >= Version('3.0'):
         return (3, 4)
     elif CASSANDRA_VERSION >= Version('2.2'):
@@ -322,10 +310,7 @@ def get_unsupported_upper_protocol():
         else:
             return ProtocolVersion.DSE_V1
     if CASSANDRA_VERSION >= Version('3.10'):
-        if DSE_VERSION:
-            return ProtocolVersion.DSE_V2
-        else:
-            return 5
+        return 5
     if CASSANDRA_VERSION >= Version('2.2'):
         return 5
     elif CASSANDRA_VERSION >= Version('2.1'):
@@ -357,7 +342,6 @@ def _id_and_mark(f):
 lessthenprotocolv4 = unittest.skipUnless(PROTOCOL_VERSION < 4, 'Protocol versions 4 or greater not supported')
 lessthanprotocolv3 = unittest.skipUnless(PROTOCOL_VERSION < 3, 'Protocol versions 3 or greater not supported')
 greaterthanprotocolv3 = unittest.skipUnless(PROTOCOL_VERSION >= 4, 'Protocol versions less than 4 are not supported')
-protocolv6 = unittest.skipUnless(6 in get_supported_protocol_versions(), 'Protocol versions less than 6 are not supported')
 
 greaterthancass20 = unittest.skipUnless(CASSANDRA_VERSION >= Version('2.1'), 'Cassandra version 2.1 or greater required')
 greaterthancass21 = unittest.skipUnless(CASSANDRA_VERSION >= Version('2.2'), 'Cassandra version 2.2 or greater required')
@@ -372,9 +356,6 @@ def _id_and_mark(f):
 lessthancass40 = unittest.skipUnless(CASSANDRA_VERSION < Version('4.0'), 'Cassandra version less than 4.0 required')
 lessthancass30 = unittest.skipUnless(CASSANDRA_VERSION < Version('3.0'), 'Cassandra version less then 3.0 required')
 
-greaterthanorequaldse68 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('6.8'), "DSE 6.8 or greater required for this test")
-greaterthanorequaldse67 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('6.7'), "DSE 6.7 or greater required for this test")
-greaterthanorequaldse60 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('6.0'), "DSE 6.0 or greater required for this test")
 greaterthanorequaldse51 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('5.1'), "DSE 5.1 or greater required for this test")
 greaterthanorequaldse50 = unittest.skipUnless(DSE_VERSION and DSE_VERSION >= Version('5.0'), "DSE 5.0 or greater required for this test")
 lessthandse51 = unittest.skipUnless(DSE_VERSION and DSE_VERSION < Version('5.1'), "DSE version less than 5.1 required")