[API-701] v5 SQL: Use different logic for getting query connections (#457)

* v5 SQL: Use different logic for getting query connections

In v5, the logic behind selecting a query connection is changed
a little bit.

Formerly, we were relying on the load balancers to fetch the next
data member. Now, we are selecting a random connection from the
largest same-version (ignoring the patch version) member group.
This way, SQL will be available on the client-side during rolling
upgrade scenarios.

* remove unused method

* correctly override _mark_minimum_server_version

* address review comments

* json -> json-flat

* make the name of version markers more clear

* update black version

Author: mdumandag

hazelcast/connection.py

@@ -32,11 +32,17 @@
     client_authentication_custom_codec,
     client_ping_codec,
 )
-from hazelcast.util import AtomicInteger, calculate_version, UNKNOWN_VERSION
+from hazelcast.util import (
+    AtomicInteger,
+    calculate_version,
+    UNKNOWN_VERSION,
+    member_of_larger_same_version_group,
+)
 
 _logger = logging.getLogger(__name__)
 
 _INF = float("inf")
+_SQL_CONNECTION_RANDOM_ATTEMPTS = 10
 
 
 class _WaitStrategy(object):
@@ -156,26 +162,66 @@ def add_listener(self, on_connection_opened=None, on_connection_closed=None):
     def get_connection(self, member_uuid):
         return self.active_connections.get(member_uuid, None)
 
-    def get_random_connection(self, should_get_data_member=False):
+    def get_random_connection(self):
+        # Try getting the connection from the load balancer, if smart routing is enabled
         if self._smart_routing_enabled:
-            connection = self._get_connection_from_load_balancer(should_get_data_member)
-            if connection:
-                return connection
-
-        # We should not get to this point under normal circumstances
-        # for the smart client. For uni-socket client, there would be
-        # a single connection in the dict. Therefore, copying the list
-        # should be acceptable.
-        for member_uuid, connection in list(six.iteritems(self.active_connections)):
-            if should_get_data_member:
-                member = self._cluster_service.get_member(member_uuid)
-                if not member or member.lite_member:
-                    continue
+            member = self._load_balancer.next()
+            if member:
+                connection = self.get_connection(member.uuid)
+                if connection:
+                    return connection
 
+        # Otherwise iterate over connections and return the first one
+        for connection in list(six.itervalues(self.active_connections)):
             return connection
 
+        # Failed to get a connection
         return None
 
+    def get_random_connection_for_sql(self):
+        """Returns a random connection for SQL.
+
+        The connection is tried to be selected in the following order.
+
+            - Random connection to a data member from the larger same-version
+              group.
+            - Random connection to a data member.
+            - Any random connection
+            - ``None``, if there is no connection.
+
+        Returns:
+            Connection: A random connection for SQL.
+        """
+        if self._smart_routing_enabled:
+            # There might be a race - the chosen member might be just connected or disconnected.
+            # Try a couple of times, the member_of_larger_same_version_group returns a random
+            # connection, we might be lucky...
+            for _ in range(_SQL_CONNECTION_RANDOM_ATTEMPTS):
+                members = self._cluster_service.get_members()
+                member = member_of_larger_same_version_group(members)
+                if not member:
+                    break
+
+                connection = self.get_connection(member.uuid)
+                if connection:
+                    return connection
+
+        # Otherwise iterate over connections and return the first one
+        # that's not to a lite member.
+        first_connection = None
+        for member_uuid, connection in list(six.iteritems(self.active_connections)):
+            if not first_connection:
+                first_connection = connection
+
+            member = self._cluster_service.get_member(member_uuid)
+            if not member or member.lite_member:
+                continue
+
+            return connection
+
+        # Failed to get a connection to a data member.
+        return first_connection
+
     def start(self, load_balancer):
         if self.live:
             return
@@ -271,20 +317,6 @@ def check_invocation_allowed(self):
         else:
             raise IOError("No connection found to cluster")
 
-    def _get_connection_from_load_balancer(self, should_get_data_member):
-        load_balancer = self._load_balancer
-        member = None
-        if should_get_data_member:
-            if load_balancer.can_get_next_data_member():
-                member = load_balancer.next_data_member()
-        else:
-            member = load_balancer.next()
-
-        if not member:
-            return None
-
-        return self.get_connection(member.uuid)
-
     def _get_or_connect_to_address(self, address):
         for connection in list(six.itervalues(self.active_connections)):
             if connection.remote_address == address:

hazelcast/sql.py

@@ -1313,23 +1313,11 @@ def execute_statement(self, statement):
         Returns:
             SqlResult: The execution result.
         """
-
-        # Get a random Data member (non-lite member)
-        connection = self._connection_manager.get_random_connection(True)
-        if not connection:
-            # Either the client is not connected to the cluster, or
-            # there are no data members in the cluster.
-            raise HazelcastSqlError(
-                self.get_client_id(),
-                _SqlErrorCode.CONNECTION_PROBLEM,
-                "Client is not currently connected to the cluster.",
-                None,
-            )
+        connection = self._get_query_connection()
+        # Create a new, unique query id.
+        query_id = _SqlQueryId.from_uuid(connection.remote_uuid)
 
         try:
-            # Create a new, unique query id.
-            query_id = _SqlQueryId.from_uuid(connection.remote_uuid)
-
             # Serialize the passed parameters.
             serialized_params = [
                 self._serialization_service.to_data(param) for param in statement.parameters
@@ -1397,15 +1385,15 @@ def re_raise(self, error, connection):
 
         Args:
             error (Exception): The error to reraise.
-            connection (hazelcast.connection.Connection): Connection
+            connection (hazelcast.connection.Connection|None): Connection
                 that the query requests are routed to. If it is not
                 live, we will inform the user about the possible
                 cluster topology change.
 
         Returns:
             HazelcastSqlError: The reraised error.
         """
-        if not connection.live:
+        if connection and not connection.live:
             return HazelcastSqlError(
                 self.get_client_id(),
                 _SqlErrorCode.CONNECTION_PROBLEM,
@@ -1438,6 +1426,22 @@ def close(self, connection, query_id):
         self._invocation_service.invoke(invocation)
         return invocation.future
 
+    def _get_query_connection(self):
+        try:
+            connection = self._connection_manager.get_random_connection_for_sql()
+        except Exception as e:
+            raise self.re_raise(e, None)
+
+        if not connection:
+            raise HazelcastSqlError(
+                self.get_client_id(),
+                _SqlErrorCode.CONNECTION_PROBLEM,
+                "Client is not connected",
+                None,
+            )
+
+        return connection
+
 
 class SqlExpectedResultType(object):
     """The expected statement result type."""

hazelcast/util.py

@@ -454,3 +454,71 @@ def try_to_get_error_message(error):
     elif len(error.args) > 0:
         return error.args[0]
     return None
+
+
+def _is_same_version(v1, v2):
+    # Ignores the patch version
+    return v1.major == v2.major and v1.minor == v2.minor
+
+
+def _is_newer_version(v1, v2):
+    # Ignores the patch version
+    return v1.major > v2.major or (v1.major == v2.major and v1.minor > v2.minor)
+
+
+def member_of_larger_same_version_group(members):
+    """Finds a larger same-version group of data members from a collection of
+    members and return a random member from the group.
+
+    If the same-version groups have the same size, return a member from the
+    newer group.
+
+    Args:
+        members (list[hazelcast.core.MemberInfo]): List of all members.
+
+    Returns:
+        hazelcast.core.MemberInfo: The chosen member or ``None``, if no data
+        member is found.
+    """
+    # The members should have at most 2 different version (ignoring the patch
+    # version).
+
+    version0 = None
+    version1 = None
+    count0 = 0
+    count1 = 0
+
+    for member in members:
+        if member.lite_member:
+            continue
+
+        v = member.version
+        if not version0 or _is_same_version(version0, v):
+            version0 = v
+            count0 += 1
+        elif not version1 or _is_same_version(version1, v):
+            version1 = v
+            count1 += 1
+        else:
+            raise ValueError(
+                "More than 2 distinct member versions found: %s, %s, %s" % (version0, version1, v)
+            )
+
+    # no data members
+    if count0 == 0:
+        return None
+
+    if count0 > count1 or count0 == count1 and _is_newer_version(version0, version1):
+        count = count0
+        version = version0
+    else:
+        count = count0
+        version = version1
+
+    # return a random member from the larger group
+    random_member_idx = random.randrange(0, count)
+    for member in members:
+        if not member.lite_member and _is_same_version(version, member.version):
+            random_member_idx -= 1
+            if random_member_idx < 0:
+                return member

requirements-dev.txt

@@ -1,4 +1,4 @@
 -r requirements-test.txt
-black==20.8b1; python_version >= "3.6"
+black==21.7b0; python_version >= "3.6"
 Sphinx==3.4.2; python_version >= "3.5"
 sphinx-rtd-theme==0.5.1; python_version >= "3.5"

tests/integration/backward_compatible/authentication_tests/authentication_test.py

@@ -3,7 +3,7 @@
 
 from hazelcast.errors import HazelcastError
 from tests.base import HazelcastTestCase
-from tests.util import get_abs_path, set_attr, is_client_version_older_than
+from tests.util import get_abs_path, set_attr, compare_client_version
 from hazelcast.client import HazelcastClient
 
 try:
@@ -14,7 +14,7 @@
 
 @set_attr(enterprise=True)
 @unittest.skipIf(
-    is_client_version_older_than("4.2.1"), "Tests the features added in 4.2.1 version of the client"
+    compare_client_version("4.2.1") < 0, "Tests the features added in 4.2.1 version of the client"
 )
 class AuthenticationTest(HazelcastTestCase):
     current_directory = os.path.dirname(__file__)

tests/integration/backward_compatible/cluster_test.py

@@ -4,7 +4,7 @@
 from hazelcast import HazelcastClient, six
 from hazelcast.util import RandomLB, RoundRobinLB
 from tests.base import HazelcastTestCase
-from tests.util import set_attr, random_string, event_collector, mark_client_version_at_least
+from tests.util import set_attr, random_string, event_collector, skip_if_client_version_older_than
 
 
 class ClusterTest(HazelcastTestCase):
@@ -228,7 +228,7 @@ def assertion():
         self.assertEqual(member.uuid, str(members[0].uuid))
 
     def test_when_member_started_with_the_same_address(self):
-        mark_client_version_at_least(self, "4.2")
+        skip_if_client_version_older_than(self, "4.2")
 
         old_member = self.cluster.start_member()
         self.client = HazelcastClient(cluster_name=self.cluster.id)

tests/integration/backward_compatible/proxy/cp/atomic_long_test.py

@@ -1,7 +1,7 @@
 from hazelcast.errors import DistributedObjectDestroyedError
 from hazelcast.serialization.api import IdentifiedDataSerializable
 from tests.integration.backward_compatible.proxy.cp import CPTestCase
-from tests.util import mark_server_version_at_least
+from tests.util import skip_if_server_version_older_than
 
 
 class Multiplication(IdentifiedDataSerializable):
@@ -108,28 +108,28 @@ def test_set(self):
 
     def test_alter(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.atomic_long.set(2)
         self.assertIsNone(self.atomic_long.alter(Multiplication(5)))
         self.assertEqual(10, self.atomic_long.get())
 
     def test_alter_and_get(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.atomic_long.set(-3)
         self.assertEqual(-9, self.atomic_long.alter_and_get(Multiplication(3)))
         self.assertEqual(-9, self.atomic_long.get())
 
     def test_get_and_alter(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.atomic_long.set(123)
         self.assertEqual(123, self.atomic_long.get_and_alter(Multiplication(-1)))
         self.assertEqual(-123, self.atomic_long.get())
 
     def test_apply(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.atomic_long.set(42)
         self.assertEqual(84, self.atomic_long.apply(Multiplication(2)))
         self.assertEqual(42, self.atomic_long.get())

tests/integration/backward_compatible/proxy/cp/atomic_reference_test.py

@@ -3,7 +3,7 @@
 
 from tests.integration.backward_compatible.util import write_string_to_output
 from tests.integration.backward_compatible.proxy.cp import CPTestCase
-from tests.util import mark_server_version_at_least
+from tests.util import skip_if_server_version_older_than
 
 
 class AppendString(IdentifiedDataSerializable):
@@ -110,35 +110,35 @@ def test_contains(self):
 
     def test_alter(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.ref.set("hey")
         self.assertIsNone(self.ref.alter(AppendString("123")))
         self.assertEqual("hey123", self.ref.get())
 
     def test_alter_with_incompatible_types(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.ref.set(42)
         with self.assertRaises(ClassCastError):
             self.ref.alter(AppendString("."))
 
     def test_alter_and_get(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.ref.set("123")
         self.assertEqual("123...", self.ref.alter_and_get(AppendString("...")))
         self.assertEqual("123...", self.ref.get())
 
     def test_get_and_alter(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.ref.set("hell")
         self.assertEqual("hell", self.ref.get_and_alter(AppendString("o")))
         self.assertEqual("hello", self.ref.get())
 
     def test_apply(self):
         # the class is defined in the 4.1 JAR
-        mark_server_version_at_least(self, self.client, "4.1")
+        skip_if_server_version_older_than(self, self.client, "4.1")
         self.ref.set("hell")
         self.assertEqual("hello", self.ref.apply(AppendString("o")))
         self.assertEqual("hell", self.ref.get())