Do not invoke urgent invocations that contain serialized data immediately

Although we believe it was a mistake, the urgent invocations on the
client-side are used for some user invocations like listener registrations,
apart from the actual urgent invocations like heartbeats, authentication, etc.

When the client reconnects to some cluster, it sends the local state
in some executor. So, there might be some time between the local state is
sent. However, during that time, urgent invocations are allowed to
go through from the client. That might violate our assumption that
the schema is received by the cluster before the data.

To solve this, we will not invoke urgent invocations that contain
serialized data if the client is not initialized on the cluster, and there
were some compact schemas sent to the previous clusters. Such
invocations are only related to user invocations so we will not delay
invoking actual urgent invocations like heartbeats.

Author: mdumandag

hazelcast/compact.py

@@ -41,6 +41,7 @@ def __init__(
         self._cluster_service = cluster_service
         self._reactor = reactor
         self._invocation_retry_pause = config.invocation_retry_pause
+        self._has_replicated_schemas = False
 
     def fetch_schema(self, schema_id: int) -> Future:
         _logger.debug(
@@ -68,6 +69,7 @@ def send_schema_and_retry(
         request = client_send_schema_codec.encode_request(schema)
 
         def callback():
+            self._has_replicated_schemas = True
             self._compact_serializer.register_schema_to_type(schema, clazz)
             return func(*args, **kwargs)
 
@@ -150,3 +152,10 @@ def register_fetched_schema(self, schema_id: int, schema: typing.Optional["Schem
             )
 
         self._compact_serializer.register_schema_to_id(schema)
+
+    def has_replicated_schemas(self):
+        """
+        Returns ``True`` is the client has replicated
+        any Compact schemas to the cluster.
+        """
+        return self._has_replicated_schemas

hazelcast/connection.py

@@ -366,6 +366,13 @@ def check_invocation_allowed(self):
         else:
             raise IOError("No connection found to cluster")
 
+    def initialized_on_cluster(self) -> bool:
+        """
+        Returns ``True`` if the client is initialized on the cluster, by
+        sending its local state, if necessary.
+        """
+        return self._client_state == _ClientState.INITIALIZED_ON_CLUSTER
+
     def _get_or_connect_to_address(self, address):
         for connection in list(self.active_connections.values()):
             if connection.remote_address == address:

hazelcast/errors.py

@@ -436,6 +436,22 @@ class NotLeaderError(HazelcastError):
     pass
 
 
+class InvocationMightContainCompactDataError(HazelcastError):
+    """
+    Signals that the invocation might contain Compact serialized data,
+    and it would not be safe to send that invocation now to make sure
+    that the invariant regarding not sending the data before the schemas
+    are hold while the client reconnects or retries urgent invocations.
+    """
+
+    def __init__(self):
+        super(InvocationMightContainCompactDataError, self).__init__(
+            "The invocation might contain Compact serialized "
+            "data and it is not safe to invoke it when the client "
+            "is not yet initialized on the cluster."
+        )
+
+
 # Error Codes
 _UNDEFINED = 0
 _ARRAY_INDEX_OUT_OF_BOUNDS = 1

hazelcast/invocation.py

@@ -12,6 +12,7 @@
     EXCEPTION_MESSAGE_TYPE,
     IndeterminateOperationStateError,
     OperationTimeoutError,
+    InvocationMightContainCompactDataError,
 )
 from hazelcast.future import Future
 from hazelcast.protocol.client_message import InboundMessage
@@ -175,7 +176,9 @@ def _invoke_on_random_connection(self, invocation):
 
     def _invoke_smart(self, invocation):
         try:
-            if not invocation.urgent:
+            if invocation.urgent:
+                self._check_urgent_invocation_allowed(invocation)
+            else:
                 self._check_invocation_allowed_fn()
 
             connection = invocation.connection
@@ -204,7 +207,9 @@ def _invoke_smart(self, invocation):
 
     def _invoke_non_smart(self, invocation):
         try:
-            if not invocation.urgent:
+            if invocation.urgent:
+                self._check_urgent_invocation_allowed(invocation)
+            else:
                 self._check_invocation_allowed_fn()
 
             connection = invocation.connection
@@ -309,6 +314,9 @@ def _retry_if_not_done(self, invocation):
             self._do_invoke(invocation)
 
     def _should_retry(self, invocation, error):
+        if isinstance(error, InvocationMightContainCompactDataError):
+            return True
+
         if invocation.connection and isinstance(error, (IOError, TargetDisconnectedError)):
             return False
 
@@ -325,6 +333,32 @@ def _should_retry(self, invocation, error):
 
         return False
 
+    def _check_urgent_invocation_allowed(self, invocation: Invocation):
+        if self._connection_manager.initialized_on_cluster():
+            # If the client is initialized on the cluster, that means we
+            # have sent all the schemas to the cluster, even if we are
+            # reconnected to it
+            return
+
+        if not self._compact_schema_service.has_replicated_schemas():
+            # If there were no Compact schemas to begin with, we don't need
+            # to perform the check below. If the client didn't send a Compact
+            # schema up until this point, the retries or listener registrations
+            # could not send a schema, because if they were, we wouldn't hit
+            # this line.
+            return
+
+        # We are not yet initialized on cluster, so the Compact schemas might
+        # not be sent yet. This message contains some serialized data,
+        # and it is possible that it can also contain Compact serialized data.
+        # In that case, allowing this invocation to go through now could
+        # violate the invariant that the schema must come to cluster before
+        # the data. We will retry this invocation and wait until the client
+        # is initialized on the cluster, which means schemas are replicated
+        # in the cluster.
+        if invocation.request.contains_data:
+            raise InvocationMightContainCompactDataError()
+
     def _register_backup_listener(self):
         codec = client_local_backup_listener_codec
         request = codec.encode_request()

hazelcast/protocol/client_message.py

@@ -64,11 +64,12 @@ def create_initial_buffer_custom(size, is_begin_frame=False):
 
 
 class OutboundMessage:
-    __slots__ = ("buf", "retryable")
+    __slots__ = ("buf", "retryable", "contains_data")
 
-    def __init__(self, buf, retryable):
+    def __init__(self, buf, retryable, contains_data=False):
         self.buf = buf
         self.retryable = retryable
+        self.contains_data = contains_data
 
     def set_correlation_id(self, correlation_id):
         LE_LONG.pack_into(self.buf, _OUTBOUND_MESSAGE_CORRELATION_ID_OFFSET, correlation_id)
@@ -80,7 +81,7 @@ def set_partition_id(self, partition_id):
         LE_INT.pack_into(self.buf, _OUTBOUND_MESSAGE_PARTITION_ID_OFFSET, partition_id)
 
     def copy(self):
-        return OutboundMessage(bytearray(self.buf), self.retryable)
+        return OutboundMessage(bytearray(self.buf), self.retryable, self.contains_data)
 
     def set_backup_aware_flag(self):
         flags = LE_UINT16.unpack_from(self.buf, INT_SIZE_IN_BYTES)[0]

hazelcast/protocol/codec/atomic_long_alter_codec.py

@@ -21,7 +21,7 @@ def encode_request(group_id, name, function, return_value_type):
     RaftGroupIdCodec.encode(buf, group_id)
     StringCodec.encode(buf, name)
     DataCodec.encode(buf, function, True)
-    return OutboundMessage(buf, False)
+    return OutboundMessage(buf, False, True)
 
 
 def decode_response(msg):

hazelcast/protocol/codec/atomic_long_apply_codec.py

@@ -17,7 +17,7 @@ def encode_request(group_id, name, function):
     RaftGroupIdCodec.encode(buf, group_id)
     StringCodec.encode(buf, name)
     DataCodec.encode(buf, function, True)
-    return OutboundMessage(buf, False)
+    return OutboundMessage(buf, False, True)
 
 
 def decode_response(msg):

hazelcast/protocol/codec/atomic_ref_apply_codec.py

@@ -23,7 +23,7 @@ def encode_request(group_id, name, function, return_value_type, alter):
     RaftGroupIdCodec.encode(buf, group_id)
     StringCodec.encode(buf, name)
     DataCodec.encode(buf, function, True)
-    return OutboundMessage(buf, False)
+    return OutboundMessage(buf, False, True)
 
 
 def decode_response(msg):

hazelcast/protocol/codec/atomic_ref_compare_and_set_codec.py

@@ -20,7 +20,7 @@ def encode_request(group_id, name, old_value, new_value):
     StringCodec.encode(buf, name)
     CodecUtil.encode_nullable(buf, old_value, DataCodec.encode)
     CodecUtil.encode_nullable(buf, new_value, DataCodec.encode, True)
-    return OutboundMessage(buf, False)
+    return OutboundMessage(buf, False, True)
 
 
 def decode_response(msg):

hazelcast/protocol/codec/atomic_ref_contains_codec.py

@@ -19,7 +19,7 @@ def encode_request(group_id, name, value):
     RaftGroupIdCodec.encode(buf, group_id)
     StringCodec.encode(buf, name)
     CodecUtil.encode_nullable(buf, value, DataCodec.encode, True)
-    return OutboundMessage(buf, True)
+    return OutboundMessage(buf, True, True)
 
 
 def decode_response(msg):