Repository: hazelcast/hazelcast-python-client

Issue:
[TRACKING ISSUE] Remove listeners sync with the calling thread
The tracking issue for the Java side PR.

See https://github.com/hazelcast/hazelcast/pull/19289 for details.

---

It turns out that there are some services relying on the removal
of a listener on the member when listener is removed from the client.
Example:
Cache listener count is kept per proxy. When listener is removed,
we decrease the listener count from the cache context.
Since listener removal is async, it could be the case that the cache
is destroyed from the calling thread before we decrement the value.
In such cases, we decrement but the incremented value is lost. Therefore,
we can end up with negative values.

The listener derregistration invoctions are not waited. It was
sync but we have changed the behavior as a side affect during
this fix https://github.com/hazelcast/hazelcast/pull/17646

We may solve this only for CacheListener but there could be more
services relying on sync listener removal. Therefore, as a fix,
we wait for invocation answer from the remote.

fixes https://github.com/hazelcast/hazelcast/issues/19269



Discussion:

Patch:
diff --git a/hazelcast/future.py b/hazelcast/future.py
index 740f934..fcf31f5 100644
--- a/hazelcast/future.py
+++ b/hazelcast/future.py
@@ -254,31 +254,56 @@ class ImmediateExceptionFuture(Future):
 def combine_futures(futures):
     """Combines set of Futures.
 
+    It waits for the completion of the all input Futures regardless
+    of their output.
+
+    The returned Future completes with the list of the results of the input
+    Futures, respecting the input order.
+
+    If one of the input Futures completes exceptionally, the returned
+    Future also completes exceptionally. In case of multiple exceptional
+    completions, the returned Future will be completed with the first
+    exceptional result.
+
     Args:
         futures (list[Future]): List of Futures to be combined.
 
     Returns:
         Future: Result of the combination.
     """
-    expected = len(futures)
-    results = []
-    if expected == 0:
+    count = len(futures)
+    results = [None] * count
+    if count == 0:
         return ImmediateFuture(results)
 
     completed = AtomicInteger()
     combined = Future()
+    errors = []
 
-    def done(f):
-        if not combined.done():
-            if f.is_success():  # TODO: ensure ordering of results as original list
-                results.append(f.result())
-                if completed.get_and_increment() + 1 == expected:
-                    combined.set_result(results)
+    def done(future, index):
+        if future.is_success():
+            results[index] = future.result()
+        else:
+            if not errors:
+                # We are fine with this check-then-act.
+                # At most, we will end up with couple of
+                # errors stored in case of the concurrent calls.
+                # The idea behind this check is to try to minimize
+                # the number of errors we store without
+                # synchronization, as we only need the first error.
+                errors.append((future.exception(), future.traceback()))
+
+        if count == completed.increment_and_get():
+            if errors:
+                first_exception, first_traceback = errors[0]
+                combined.set_exception(first_exception, first_traceback)
             else:
-                combined.set_exception(f.exception(), f.traceback())
+                combined.set_result(results)
 
-    for future in futures:
-        future.add_done_callback(done)
+    for index, future in enumerate(futures):
+        # Capture the index in the closure or else we
+        # will only update the last element.
+        future.add_done_callback(lambda f, captured_index=index: done(f, captured_index))
 
     return combined
 
diff --git a/hazelcast/listener.py b/hazelcast/listener.py
index 01f27e0..40ac327 100644
--- a/hazelcast/listener.py
+++ b/hazelcast/listener.py
@@ -86,6 +86,7 @@ class ListenerService(object):
             if not listener_registration:
                 return ImmediateFuture(False)
 
+            futures = []
             for connection, event_registration in six.iteritems(
                 listener_registration.connection_registrations
             ):
@@ -105,24 +106,28 @@ class ListenerService(object):
                 )
                 self._invocation_service.invoke(invocation)
 
-                def handler(f, connection=connection):
+                def handler(f, captured_connection=connection):
                     e = f.exception()
-                    if e:
-                        if isinstance(
-                            e, (HazelcastClientNotActiveError, IOError, TargetDisconnectedError)
-                        ):
-                            return
-
-                        _logger.warning(
-                            "Deregistration of listener with ID %s has failed for address %s",
-                            user_registration_id,
-                            connection.remote_address,
-                        )
+                    if not e:
+                        return
+
+                    if isinstance(
+                        e, (HazelcastClientNotActiveError, IOError, TargetDisconnectedError)
+                    ):
+                        return
+
+                    _logger.warning(
+                        "Deregistration of listener with ID %s has failed for address %s: %s",
+                        user_registration_id,
+                        captured_connection.remote_address,
+                        e,
+                    )
 
                 invocation.future.add_done_callback(handler)
+                futures.append(invocation.future)
 
             listener_registration.connection_registrations.clear()
-            return ImmediateFuture(True)
+            return combine_futures(futures).continue_with(lambda _: True)
 
     def handle_client_message(self, message, correlation_id):
         handler = self._event_handlers.get(correlation_id, None)
diff --git a/hazelcast/protocol/codec/custom/sql_error_codec.py b/hazelcast/protocol/codec/custom/sql_error_codec.py
index 37bea69..01dd277 100644
--- a/hazelcast/protocol/codec/custom/sql_error_codec.py
+++ b/hazelcast/protocol/codec/custom/sql_error_codec.py
@@ -19,6 +19,7 @@ class SqlErrorCodec(object):
         FixSizedTypesCodec.encode_uuid(initial_frame_buf, _ORIGINATING_MEMBER_ID_ENCODE_OFFSET, sql_error.originating_member_id)
         buf.extend(initial_frame_buf)
         CodecUtil.encode_nullable(buf, sql_error.message, StringCodec.encode)
+        CodecUtil.encode_nullable(buf, sql_error.suggestion, StringCodec.encode)
         if is_final:
             buf.extend(END_FINAL_FRAME_BUF)
         else:
@@ -31,5 +32,10 @@ class SqlErrorCodec(object):
         code = FixSizedTypesCodec.decode_int(initial_frame.buf, _CODE_DECODE_OFFSET)
         originating_member_id = FixSizedTypesCodec.decode_uuid(initial_frame.buf, _ORIGINATING_MEMBER_ID_DECODE_OFFSET)
         message = CodecUtil.decode_nullable(msg, StringCodec.decode)
+        is_suggestion_exists = False
+        suggestion = None
+        if not msg.peek_next_frame().is_end_frame():
+            suggestion = CodecUtil.decode_nullable(msg, StringCodec.decode)
+            is_suggestion_exists = True
         CodecUtil.fast_forward_to_end_frame(msg)
-        return _SqlError(code, message, originating_member_id)
+        return _SqlError(code, message, originating_member_id, is_suggestion_exists, suggestion)
diff --git a/hazelcast/sql.py b/hazelcast/sql.py
index 2e6ba1f..9a11c12 100644
--- a/hazelcast/sql.py
+++ b/hazelcast/sql.py
@@ -250,9 +250,9 @@ class SqlColumnMetadata(object):
 class _SqlError(object):
     """Server-side error that is propagated to the client."""
 
-    __slots__ = ("code", "message", "originating_member_uuid")
+    __slots__ = ("code", "message", "originating_member_uuid", "suggestion")
 
-    def __init__(self, code, message, originating_member_uuid):
+    def __init__(self, code, message, originating_member_uuid, _, suggestion):
         self.code = code
         """_SqlErrorCode: The error code."""
 
@@ -262,6 +262,9 @@ class _SqlError(object):
         self.originating_member_uuid = originating_member_uuid
         """uuid.UUID: UUID of the member that caused or initiated an error condition."""
 
+        self.suggestion = suggestion
+        """str: Suggested SQL statement to remediate experienced error."""
+
 
 class _SqlPage(object):
     """A finite set of rows returned to the user."""
@@ -409,11 +412,6 @@ class _SqlErrorCode(object):
     A problem with partition distribution.
     """
 
-    MAP_DESTROYED = 1006
-    """
-    An error caused by a concurrent destroy of a map.
-    """
-
     MAP_LOADING_IN_PROGRESS = 1007
     """
     Map loading is not finished yet.
@@ -429,6 +427,11 @@ class _SqlErrorCode(object):
     An error caused by an attempt to query an index that is not valid.
     """
 
+    OBJECT_NOT_FOUND = 1010
+    """
+    Object (mapping/table) not found.
+    """
+
     DATA_EXCEPTION = 2000
     """
     An error with data conversion or transformation.
@@ -438,9 +441,10 @@ class _SqlErrorCode(object):
 class HazelcastSqlError(HazelcastError):
     """Represents an error occurred during the SQL query execution."""
 
-    def __init__(self, originating_member_uuid, code, message, cause):
+    def __init__(self, originating_member_uuid, code, message, cause, suggestion=None):
         super(HazelcastSqlError, self).__init__(message, cause)
         self._originating_member_uuid = originating_member_uuid
+        self._suggestion = suggestion
 
         # TODO: This is private API, might be good to make it public or
         #  remove this information altogether.
@@ -451,6 +455,11 @@ class HazelcastSqlError(HazelcastError):
         """uuid.UUID: UUID of the member that caused or initiated an error condition."""
         return self._originating_member_uuid
 
+    @property
+    def suggestion(self):
+        """str: Suggested SQL statement to remediate experienced error."""
+        return self._suggestion
+
 
 class SqlRowMetadata(object):
     """Metadata for the returned rows."""
@@ -514,11 +523,12 @@ class SqlRowMetadata(object):
 class SqlRow(object):
     """One of the rows of an SQL query result."""
 
-    __slots__ = ("_row_metadata", "_row")
+    __slots__ = ("_row_metadata", "_row", "_deserialize_fn")
 
-    def __init__(self, row_metadata, row):
+    def __init__(self, row_metadata, row, deserialize_fn):
         self._row_metadata = row_metadata
         self._row = row
+        self._deserialize_fn = deserialize_fn
 
     def get_object(self, column_name):
         """Gets the value in the column indicated by the column name.
@@ -530,6 +540,13 @@ class SqlRow(object):
         The type of the returned value depends on the SQL type of the column.
         No implicit conversions are performed on the value.
 
+        Warnings:
+
+            Each call to this method might result in a deserialization if the
+            column type for this object is :const:`SqlColumnType.OBJECT`.
+            It is advised to assign the result of this method call to some
+            variable and reuse it.
+
         Args:
             column_name (str):
 
@@ -539,6 +556,7 @@ class SqlRow(object):
         Raises:
             ValueError: If a column with the given name does not exist.
             AssertionError: If the column name is not a string.
+            HazelcastSqlError: If the object cannot be deserialized.
 
         See Also:
             :attr:`metadata`
@@ -552,7 +570,7 @@ class SqlRow(object):
         index = self._row_metadata.find_column(column_name)
         if index == SqlRowMetadata.COLUMN_NOT_FOUND:
             raise ValueError("Column '%s' doesn't exist" % column_name)
-        return self._row[index]
+        return self._deserialize_fn(self._row[index])
 
     def get_object_with_index(self, column_index):
         """Gets the value of the column by index.
@@ -560,6 +578,13 @@ class SqlRow(object):
         The class of the returned value depends on the SQL type of the column.
         No implicit conversions are performed on the value.
 
+        Warnings:
+
+            Each call to this method might result in a deserialization if the
+            column type for this object is :const:`SqlColumnType.OBJECT`.
+            It is advised to assign the result of this method call to some
+            variable and reuse it.
+
         Args:
             column_index (int): Zero-based column index.
 
@@ -569,6 +594,7 @@ class SqlRow(object):
         Raises:
             IndexError: If the column index is out of bounds.
             AssertionError: If the column index is not an integer.
+            HazelcastSqlError: If the object cannot be deserialized.
 
         See Also:
             :attr:`metadata`
@@ -576,7 +602,7 @@ class SqlRow(object):
             :attr:`SqlColumnMetadata.type`
         """
         check_is_int(column_index, "Column index must be an integer")
-        return self._row[column_index]
+        return self._deserialize_fn(self._row[column_index])
 
     @property
     def metadata(self):
@@ -589,7 +615,7 @@ class SqlRow(object):
             % (
                 self._row_metadata.get_column(i).name,
                 get_attr_name(SqlColumnType, self._row_metadata.get_column(i).type),
-                self._row[i],
+                self.get_object_with_index(i),
             )
             for i in range(self._row_metadata.column_count)
         )
@@ -670,12 +696,8 @@ class _IteratorBase(object):
             list: The row pointed by the current position.
         """
 
-        # The column might contain user objects so we have to deserialize it.
-        # Deserialization is no-op if the value is not Data.
-        return [
-            self.deserialize_fn(self.page.get_value(i, self.position))
-            for i in range(self.page.column_count)
-        ]
+        # Deserialization happens lazily while getting the object.
+        return [self.page.get_value(i, self.position) for i in range(self.page.column_count)]
 
 
 class _FutureProducingIterator(_IteratorBase):
@@ -715,7 +737,7 @@ class _FutureProducingIterator(_IteratorBase):
 
         row = self._get_current_row()
         self.position += 1
-        return SqlRow(self.row_metadata, row)
+        return SqlRow(self.row_metadata, row, self.deserialize_fn)
 
     def _has_next(self):
         """Returns a Future indicating whether there are more rows
@@ -779,7 +801,7 @@ class _BlockingIterator(_IteratorBase):
 
         row = self._get_current_row()
         self.position += 1
-        return SqlRow(self.row_metadata, row)
+        return SqlRow(self.row_metadata, row, self.deserialize_fn)
 
     def _has_next(self):
         while self.position == self.row_count:
@@ -1216,7 +1238,13 @@ class SqlResult(object):
             ``None`` otherwise.
         """
         if error:
-            return HazelcastSqlError(error.originating_member_uuid, error.code, error.message, None)
+            return HazelcastSqlError(
+                error.originating_member_uuid,
+                error.code,
+                error.message,
+                None,
+                error.suggestion,
+            )
         return None
 
     def _on_execute_error(self, error):
@@ -1349,7 +1377,15 @@ class _InternalSqlService(object):
             raise self.re_raise(e, connection)
 
     def deserialize_object(self, obj):
-        return self._serialization_service.to_object(obj)
+        try:
+            return self._serialization_service.to_object(obj)
+        except Exception as e:
+            raise HazelcastSqlError(
+                self.get_client_id(),
+                _SqlErrorCode.GENERIC,
+                "Failed to deserialize query result value: %s" % try_to_get_error_message(e),
+                e,
+            )
 
     def fetch(self, connection, query_id, cursor_buffer_size):
         """Fetches the next page of the query execution.
diff --git a/hazelcast/util.py b/hazelcast/util.py
index c6c7020..1654bc9 100644
--- a/hazelcast/util.py
+++ b/hazelcast/util.py
@@ -101,6 +101,16 @@ class AtomicInteger(object):
             self._counter += 1
             return res
 
+    def increment_and_get(self):
+        """Increments the current value and returns it.
+
+        Returns:
+            int: Incremented value of AtomicInteger.
+        """
+        with self._mux:
+            self._counter += 1
+            return self._counter
+
     def get(self):
         """Returns the current value.
 


Test patch:
diff --git a/tests/integration/backward_compatible/sql_test.py b/tests/integration/backward_compatible/sql_test.py
index 9fe05c9..ae5309a 100644
--- a/tests/integration/backward_compatible/sql_test.py
+++ b/tests/integration/backward_compatible/sql_test.py
@@ -17,6 +17,7 @@ from tests.util import (
     compare_client_version,
     skip_if_server_version_older_than,
     skip_if_server_version_newer_than_or_equal,
+    skip_if_client_version_older_than,
 )
 
 try:
@@ -99,7 +100,7 @@ class SqlTestBase(HazelcastTestCase):
             return
 
         create_mapping_query = """
-        CREATE MAPPING %s (
+        CREATE MAPPING "%s" (
             __key INT,
             this %s
         )
@@ -123,8 +124,8 @@ class SqlTestBase(HazelcastTestCase):
             return
 
         create_mapping_query = """
-        CREATE MAPPING %s (
-            __key INT,
+        CREATE MAPPING "%s" (
+            __key INT%s
             %s
         )
         TYPE IMaP 
@@ -136,6 +137,7 @@ class SqlTestBase(HazelcastTestCase):
         )
         """ % (
             self.map_name,
+            "," if len(columns) > 0 else "",
             ",\n".join(["%s %s" % (c_name, c_type) for c_name, c_type in columns.items()]),
             factory_id,
             class_id,
@@ -315,6 +317,26 @@ class SqlServiceTest(SqlTestBase):
     # Can't test the schema, because the IMDG SQL engine does not support
     # specifying a schema yet.
 
+    def test_provided_suggestions(self):
+        skip_if_client_version_older_than(self, "5.0")
+        skip_if_server_version_older_than(self, self.client, "5.0")
+
+        # We don't create a mapping intentionally to get suggestions
+        self.map.put(1, "value-1")
+        select_all_query = 'SELECT * FROM "%s"' % self.map_name
+        with self.assertRaises(HazelcastSqlError) as cm:
+            with self.client.sql.execute(select_all_query) as result:
+                result.update_count().result()
+
+        with self.client.sql.execute(cm.exception.suggestion) as result:
+            result.update_count().result()
+
+        with self.client.sql.execute(select_all_query) as result:
+            self.assertEqual(
+                [(1, "value-1")],
+                [(r.get_object("__key"), r.get_object("this")) for r in result],
+            )
+
 
 @unittest.skipIf(
     compare_client_version("4.2") < 0, "Tests the features added in 4.2 version of the client"
@@ -521,6 +543,35 @@ class SqlResultTest(SqlTestBase):
 
         self.assertIsInstance(result.close(), ImmediateFuture)
 
+    def test_lazy_deserialization(self):
+        skip_if_client_version_older_than(self, "5.0")
+
+        # Using a Portable that is not defined on the client-side.
+        self._create_mapping_for_portable(666, 1, {})
+
+        script = (
+            """
+        var m = instance_0.getMap("%s");
+        m.put(1, new com.hazelcast.client.test.Employee(1, "Joe"));
+        """
+            % self.map_name
+        )
+
+        res = self.rc.executeOnController(self.cluster.id, script, Lang.JAVASCRIPT)
+        self.assertTrue(res.success)
+
+        with self.client.sql.execute('SELECT __key, this FROM "%s"' % self.map_name) as result:
+            rows = list(result)
+            self.assertEqual(1, len(rows))
+            row = rows[0]
+            # We should be able to deserialize parts of the response
+            self.assertEqual(1, row.get_object("__key"))
+
+            # We should throw lazily when we try to access the columns
+            # that are not deserializable
+            with self.assertRaises(HazelcastSqlError):
+                row.get_object("this")
+
 
 @unittest.skipIf(
     compare_client_version("4.2") < 0, "Tests the features added in 4.2 version of the client"
diff --git a/tests/unit/future_test.py b/tests/unit/future_test.py
index 64b0bf5..61c3c91 100644
--- a/tests/unit/future_test.py
+++ b/tests/unit/future_test.py
@@ -394,7 +394,7 @@ class ImmediateExceptionFutureTest(unittest.TestCase):
 
 
 class CombineFutureTest(unittest.TestCase):
-    def test_combine_futures(self):
+    def test_combining(self):
         f1, f2, f3 = Future(), Future(), Future()
 
         combined = combine_futures([f1, f2, f3])
@@ -408,7 +408,7 @@ class CombineFutureTest(unittest.TestCase):
         f3.set_result("done3")
         self.assertEqual(combined.result(), ["done1", "done2", "done3"])
 
-    def test_combine_futures_exception(self):
+    def test_exceptional_result(self):
         f1, f2, f3 = Future(), Future(), Future()
 
         combined = combine_futures([f1, f2, f3])
@@ -420,11 +420,59 @@ class CombineFutureTest(unittest.TestCase):
 
         self.assertEqual(e, combined.exception())
 
-    def test_combine_futures_with_empty_list(self):
+    def test_empty_input(self):
         combined = combine_futures([])
         self.assertTrue(combined.done())
         self.assertEqual([], combined.result())
 
+    def test_completion_order(self):
+        # Verifies that the returned list order is
+        # consistent with the order of input futures
+        # regardless of the completion order of futures.
+        f1, f2, f3 = Future(), Future(), Future()
+        combined = combine_futures([f1, f2, f3])
+
+        f3.set_result(3)
+        f1.set_result(1)
+        f2.set_result(2)
+
+        self.assertEqual([1, 2, 3], combined.result())
+
+    def test_waiting_for_all_inputs(self):
+        # Verifies that we wait for all futures, even if some
+        # of them fails.
+        f1, f2, f3 = Future(), Future(), Future()
+        combined = combine_futures([f1, f2, f3])
+
+        e = RuntimeError("expected")
+        f1.set_exception(e)
+        self.assertFalse(combined.done())
+        f2.set_result(1)
+        self.assertFalse(combined.done())
+        f3.set_result(2)
+        self.assertTrue(combined.done())
+        self.assertFalse(combined.is_success())
+        self.assertEqual(e, combined.exception())
+
+    def test_returning_first_error(self):
+        # Verifies that we return the first error occurred, even
+        # if more than one of the input futures completed
+        # exceptionally.
+        f1, f2, f3 = Future(), Future(), Future()
+        combined = combine_futures([f1, f2, f3])
+
+        e1 = RuntimeError("expected")
+        f1.set_exception(e1)
+        self.assertFalse(combined.done())
+        e2 = RuntimeError("expected2")
+        f2.set_exception(e2)
+        self.assertFalse(combined.done())
+        e3 = RuntimeError("expected3")
+        f3.set_exception(e3)
+        self.assertTrue(combined.done())
+        self.assertFalse(combined.is_success())
+        self.assertEqual(e1, combined.exception())
+
 
 class MakeBlockingTest(unittest.TestCase):
     class Calculator(object):
diff --git a/tests/unit/sql_test.py b/tests/unit/sql_test.py
index 9e25c91..c0aa33e 100644
--- a/tests/unit/sql_test.py
+++ b/tests/unit/sql_test.py
@@ -116,12 +116,12 @@ class SqlMockTest(unittest.TestCase):
 
     def test_close_when_close_request_fails(self):
         future = self.result.close()
-        self.set_close_error(HazelcastSqlError(None, _SqlErrorCode.MAP_DESTROYED, "expected", None))
+        self.set_close_error(HazelcastSqlError(None, _SqlErrorCode.PARSING, "expected", None))
 
         with self.assertRaises(HazelcastSqlError) as cm:
             future.result()
 
-        self.assertEqual(_SqlErrorCode.MAP_DESTROYED, cm.exception._code)
+        self.assertEqual(_SqlErrorCode.PARSING, cm.exception._code)
 
     def test_fetch_error(self):
         self.set_execute_response_with_rows(is_last=False)
@@ -184,7 +184,10 @@ class SqlMockTest(unittest.TestCase):
         self.assertEqual(_SqlErrorCode.CANCELLED_BY_USER, cm.exception._code)
 
     def set_fetch_response_with_error(self):
-        response = {"row_page": None, "error": _SqlError(_SqlErrorCode.PARSING, "expected", None)}
+        response = {
+            "row_page": None,
+            "error": _SqlError(_SqlErrorCode.PARSING, "expected", None, None, ""),
+        }
         self.set_future_result_or_exception(response, sql_fetch_codec._REQUEST_MESSAGE_TYPE)
 
     def set_fetch_error(self, error):