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

@@ -170,9 +170,9 @@ class ProtocolVersion(object):
     DSE private protocol v2, supported in DSE 6.0+
     """
 
-    SUPPORTED_VERSIONS = (DSE_V2, DSE_V1, V6, V5, V4, V3)
+    SUPPORTED_VERSIONS = (V5, V4, V3)
     """
-    A tuple of all supported protocol versions
+    A tuple of all supported protocol versions for ScyllaDB, including future v5 support.
     """
 
     BETA_VERSIONS = (V6,)
@@ -227,10 +227,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.
 
@@ -2692,7 +2692,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:
@@ -2889,26 +2888,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
@@ -3053,13 +3036,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

@@ -444,32 +444,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

@@ -615,8 +615,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):
@@ -1050,12 +1048,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

@@ -192,7 +192,7 @@ def get_supported_protocol_versions():
     4.0(C*) -> 6(beta),5,4,3
 `   """
     if CASSANDRA_VERSION >= Version('4.0-beta5'):
-        return (3, 4, 5, 6)
+        return (3, 4, 5)
     if CASSANDRA_VERSION >= Version('4.0-a'):
        return (3, 4, 5)
     elif CASSANDRA_VERSION >= Version('3.10'):
@@ -261,7 +261,6 @@ def _id_and_mark(f):
 local = local_decorator_creator()
 notprotocolv1 = unittest.skipUnless(PROTOCOL_VERSION > 1, 'Protocol v1 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')

tests/integration/advanced/graph/fluent/test_graph.py

@@ -0,0 +1,40 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from cassandra import cluster
+from cassandra.cluster import ContinuousPagingOptions
+from cassandra.datastax.graph.fluent import DseGraph
+from cassandra.graph import VertexProperty
+
+from tests.integration.advanced.graph import (
+    GraphUnitTestCase, ClassicGraphSchema, CoreGraphSchema,
+    VertexLabel, GraphTestConfiguration
+)
+from tests.integration.advanced.graph.fluent import (
+    BaseExplicitExecutionTest, create_traversal_profiles, check_equality_base)
+
+import unittest
+
+
+class ContinuousPagingOptionsForTests(ContinuousPagingOptions):
+    def __init__(self,
+                 page_unit=ContinuousPagingOptions.PagingUnit.ROWS, max_pages=1,  # max_pages=1
+                 max_pages_per_second=0, max_queue_size=4):
+        super(ContinuousPagingOptionsForTests, self).__init__(page_unit, max_pages, max_pages_per_second,
+                                                              max_queue_size)
+
+
+def reset_paging_options():
+    cluster.ContinuousPagingOptions = ContinuousPagingOptions
+