Schema distribution on cluster-restart is implemented [API-1328] (#1067)

* Schema distribution on cluster-restart is implemented.

Author: OzanCansel

hazelcast/generated-sources/src/hazelcast/client/protocol/codec/codecs.cpp

@@ -5804,6 +5804,23 @@ client_sendschema_encode(const serialization::pimpl::schema& schema)
     return msg;
 }
 
+ClientMessage
+client_sendallschemas_encode(
+  const std::vector<serialization::pimpl::schema>& schemas)
+{
+    size_t initial_frame_size = ClientMessage::REQUEST_HEADER_LEN;
+    ClientMessage msg(initial_frame_size);
+    msg.set_retryable(true);
+    msg.set_operation_name("client.sendallschemas");
+
+    msg.set_message_type(static_cast<int32_t>(5376));
+    msg.set_partition_id(-1);
+
+    msg.set(schemas, true);
+
+    return msg;
+}
+
 } // namespace codec
 } // namespace protocol
 } // namespace client

hazelcast/generated-sources/src/hazelcast/client/protocol/codec/codecs.h

@@ -3085,6 +3085,13 @@ sql_fetch_encode(const sql::impl::query_id& query_id,
 ClientMessage HAZELCAST_API
 client_sendschema_encode(const serialization::pimpl::schema& schema);
 
+/**
+ * Sends all the schemas to the cluster
+ */
+ClientMessage HAZELCAST_API
+client_sendallschemas_encode(
+  const std::vector<serialization::pimpl::schema>& schemas);
+
 } // namespace codec
 } // namespace protocol
 } // namespace client

hazelcast/include/hazelcast/client/connection/ClientConnectionManagerImpl.h

@@ -21,6 +21,7 @@
 #include <thread>
 #include <future>
 #include <vector>
+#include <mutex>
 #include <boost/asio.hpp>
 #include <boost/smart_ptr/atomic_shared_ptr.hpp>
 
@@ -137,8 +138,24 @@ class HAZELCAST_API ClientConnectionManagerImpl
 
     boost::uuids::uuid get_client_uuid() const;
 
+    /**
+     * Check the connected state and user connection strategy configuration to
+     * see if an invocation is allowed at the moment returns without throwing
+     * exception only when is the client is Connected to cluster
+     *
+     * @throws io_exception                    if client is disconnected and
+     * ReconnectMode is ON or if client is starting and async start is false
+     * @throws hazelcast_client_offline        if client is disconnected and
+     * ReconnectMode is ASYNC or if client is starting and async start is true
+     */
     void check_invocation_allowed();
 
+    /**
+     * Returns {@code true} if the client is initialized on the cluster, by
+     * sending its local state, if necessary.
+     */
+    bool client_initialized_on_cluster() const;
+
     void connect_to_all_cluster_members();
 
     void notify_backup(int64_t call_id);
@@ -160,6 +177,42 @@ class HAZELCAST_API ClientConnectionManagerImpl
         std::string server_version;
     };
 
+    enum class client_state
+    {
+        /**
+         * Clients start with this state. Once a client connects to a cluster,
+         * it directly switches to {@link #INITIALIZED_ON_CLUSTER} instead of
+         * {@link #CONNECTED_TO_CLUSTER} because on startup a client has no
+         * local state to send to the cluster.
+         */
+        INITIAL,
+
+        /**
+         * When a client switches to a new cluster, it moves to this state.
+         * It means that the client has connected to a new cluster but not sent
+         * its local state to the new cluster yet.
+         */
+        CONNECTED_TO_CLUSTER,
+
+        /**
+         * When a client sends its local state to the cluster it has connected,
+         * it switches to this state.
+         * <p>
+         * Invocations are allowed in this state.
+         */
+        INITIALIZED_ON_CLUSTER,
+
+        /**
+         * When the client closes the last connection to the cluster it
+         * currently connected to, it switches to this state.
+         * <p>
+         */
+        DISCONNECTED_FROM_CLUSTER,
+    };
+
+    friend std::ostream HAZELCAST_API& operator<<(std::ostream& os,
+                                                  client_state);
+
     auth_response authenticate_on_cluster(
       std::shared_ptr<Connection>& connection);
 
@@ -178,7 +231,7 @@ class HAZELCAST_API ClientConnectionManagerImpl
     static void shutdown_with_external_thread(
       std::weak_ptr<client::impl::hazelcast_client_instance_impl> client_impl);
 
-    bool do_connect_to_cluster();
+    void do_connect_to_cluster();
 
     std::vector<address> get_possible_member_addresses();
 
@@ -197,6 +250,8 @@ class HAZELCAST_API ClientConnectionManagerImpl
 
     void check_client_active();
 
+    void initialize_client_on_cluster(boost::uuids::uuid);
+
     template<typename T>
     std::shared_ptr<Connection> try_connect(const T& target)
     {
@@ -255,7 +310,7 @@ class HAZELCAST_API ClientConnectionManagerImpl
     wait_strategy wait_strategy_;
 
     // following fields are updated inside synchronized(clientStateMutex)
-    std::recursive_mutex client_state_mutex_;
+    mutable std::recursive_mutex client_state_mutex_;
     util::SynchronizedMap<boost::uuids::uuid,
                           Connection,
                           boost::hash<boost::uuids::uuid>>
@@ -267,7 +322,9 @@ class HAZELCAST_API ClientConnectionManagerImpl
 #else
     std::atomic<boost::uuids::uuid> cluster_id_;
 #endif
+    client_state client_state_;
     std::atomic_bool connect_to_cluster_task_submitted_;
+    bool established_initial_cluster_connection;
 
     bool use_public_address_{ false };
 
@@ -283,6 +340,7 @@ class HAZELCAST_API ClientConnectionManagerImpl
 
     address translate(const member& m);
 };
+
 } // namespace connection
 } // namespace client
 } // namespace hazelcast

hazelcast/include/hazelcast/client/exception/protocol_exceptions.h

@@ -28,6 +28,11 @@
 
 namespace hazelcast {
 namespace client {
+namespace spi {
+namespace impl {
+class ClientInvocation;
+}
+} // namespace spi
 namespace exception {
 #define DEFINE_EXCEPTION_CLASS(ClassName, errorNo, isRuntime)                  \
     class HAZELCAST_API ClassName : public iexception                          \
@@ -351,6 +356,14 @@ class HAZELCAST_API query : public hazelcast_
     boost::uuids::uuid originating_member_uuid_;
 };
 
+class HAZELCAST_API invocation_might_contain_compact_data : public hazelcast_
+{
+public:
+    explicit invocation_might_contain_compact_data(
+      std::string source,
+      const spi::impl::ClientInvocation& invocation);
+};
+
 class HAZELCAST_API member_left : public execution
 {
 public:

hazelcast/include/hazelcast/client/impl/hazelcast_client_instance_impl.h

@@ -222,6 +222,14 @@ class HAZELCAST_API hazelcast_client_instance_impl
 
     sql::sql_service& get_sql();
 
+    void send_state_to_cluster();
+
+    /**
+     * Returns {@code true} if we need to check the urgent invocations, by
+     * examining the local registry of the schema service.
+     */
+    bool should_check_urgent_invocations() const;
+
 private:
     client_config client_config_;
     client_properties client_properties_;

hazelcast/include/hazelcast/client/protocol/ClientMessage.h

@@ -1122,6 +1122,10 @@ class HAZELCAST_API ClientMessage
             if (is_final) {
                 h->flags |= IS_FINAL_FLAG;
             }
+
+            if (std::is_same<T, serialization::pimpl::data>::value) {
+                contains_serialized_data_in_request_ = true;
+            }
         } else {
             set(*value, is_final);
         }
@@ -1241,6 +1245,8 @@ class HAZELCAST_API ClientMessage
         copy(begin(replicated_schemas),
              end(replicated_schemas),
              back_inserter(schemas_will_be_replicated_));
+
+        contains_serialized_data_in_request_ = true;
     }
 
     inline void set(const serialization::pimpl::data* value,
@@ -1423,6 +1429,8 @@ class HAZELCAST_API ClientMessage
 
     void drop_fragmentation_frame();
 
+    bool contains_serialized_data_in_request() const;
+
     friend std::ostream HAZELCAST_API& operator<<(std::ostream& os,
                                                   const ClientMessage& message);
 
@@ -1520,6 +1528,7 @@ class HAZELCAST_API ClientMessage
     std::vector<std::vector<byte>> data_buffer_;
     size_t buffer_index_{ 0 };
     size_t offset_{ 0 };
+    bool contains_serialized_data_in_request_;
     std::vector<serialization::pimpl::schema> schemas_will_be_replicated_;
 };
 

hazelcast/include/hazelcast/client/serialization/pimpl/compact/default_schema_service.h

@@ -21,6 +21,7 @@
 #include "hazelcast/util/export.h"
 #include "hazelcast/util/SynchronizedMap.h"
 #include "hazelcast/client/serialization/pimpl/compact/schema.h"
+#include "hazelcast/logger.h"
 
 namespace hazelcast {
 namespace client {
@@ -32,7 +33,6 @@ class ClientMessage;
 namespace spi {
 class ClientContext;
 }
-
 namespace serialization {
 namespace pimpl {
 
@@ -51,6 +51,8 @@ class HAZELCAST_API default_schema_service
     static constexpr const char* MAX_PUT_RETRY_COUNT_DEFAULT = "100";
 
     default_schema_service(spi::ClientContext&);
+    default_schema_service(const default_schema_service&) = delete;
+    default_schema_service& operator=(const default_schema_service&) = delete;
 
     /**
      * Gets the schema with the given id either by
@@ -68,6 +70,16 @@ class HAZELCAST_API default_schema_service
 
     bool is_schema_replicated(const schema&);
 
+    /**
+     * Replicates all schemas on cluster
+     */
+    void replicate_all_schemas();
+
+    /**
+     * Check whether any schemas exist in cache
+     */
+    bool has_any_schemas() const;
+
 private:
     void put_if_absent(schema);
 

hazelcast/include/hazelcast/client/spi/impl/ClientInvocationServiceImpl.h

@@ -59,6 +59,8 @@ class HAZELCAST_API ClientInvocationServiceImpl
 
     void check_invocation_allowed();
 
+    void check_urgent_invocation_allowed(const ClientInvocation&);
+
     bool is_smart_routing() const;
 
     std::chrono::milliseconds get_invocation_timeout() const;

hazelcast/src/hazelcast/client/client_impl.cpp

@@ -477,6 +477,18 @@ hazelcast_client_instance_impl::get_sql()
     return sql_service_;
 }
 
+void
+hazelcast_client_instance_impl::send_state_to_cluster()
+{
+    schema_service_.replicate_all_schemas();
+}
+
+bool
+hazelcast_client_instance_impl::should_check_urgent_invocations() const
+{
+    return schema_service_.has_any_schemas();
+}
+
 void
 hazelcast_client_instance_impl::check_discovery_configuration_consistency(
   bool address_list_provided,
@@ -1363,6 +1375,21 @@ query::originating_member_uuid() const
     return originating_member_uuid_;
 }
 
+invocation_might_contain_compact_data::invocation_might_contain_compact_data(
+  std::string source,
+  const spi::impl::ClientInvocation& invocation)
+  : hazelcast_{
+      move(source),
+      boost::str(
+        boost::format(
+          "The invocation %1% might contain Compact serialized "
+          "data and it is not safe to invoke it when the client is not "
+          "yet initialized on the cluster") %
+        invocation)
+  }
+{
+}
+
 } // namespace exception
 } // namespace client
 

hazelcast/src/hazelcast/client/compact.cpp

@@ -1995,6 +1995,66 @@ default_schema_service::send_schema_response_decode(
                                           boost::hash<boost::uuids::uuid>>>();
 }
 
+bool
+default_schema_service::has_any_schemas() const
+{
+    return replicateds_.size();
+}
+
+std::ostream&
+operator<<(std::ostream& os, const std::vector<schema>& schemas)
+{
+    os << "Schemas {";
+
+    for (const auto& s : schemas)
+        os << s << " , ";
+
+    os << "}";
+
+    return os;
+}
+
+void
+default_schema_service::replicate_all_schemas()
+{
+    using level = hazelcast::logger::level;
+    using namespace protocol::codec;
+
+    auto logger = context_.get_logger();
+    if (replicateds_.empty()) {
+        if (logger.enabled(level::finest)) {
+            logger.log(level::finest,
+                       "There is no schema to send to the cluster.");
+        }
+
+        return;
+    }
+
+    std::vector<std::shared_ptr<schema>> schemas_sptr = replicateds_.values();
+    std::vector<schema> all_schemas;
+
+    all_schemas.reserve(schemas_sptr.size());
+
+    transform(begin(schemas_sptr),
+              end(schemas_sptr),
+              back_inserter(all_schemas),
+              [](const std::shared_ptr<schema>& s) { return *s; });
+
+    if (logger.enabled(level::finest)) {
+        logger.log(
+          level::finest,
+          (boost::format("Sending schemas to the cluster %1%") % all_schemas)
+            .str());
+    }
+
+    auto message = client_sendallschemas_encode(all_schemas);
+
+    auto invocation =
+      spi::impl::ClientInvocation::create(context_, message, SERVICE_NAME);
+
+    invocation->invoke_urgent().get();
+}
+
 compact_stream_serializer::compact_stream_serializer(
   default_schema_service& service)
   : schema_service{ service }