IGNITE-27900 Use message serializer for CacheStatisticsClearMessage, ChangeGlobalStateMessage, DynamicCacheChangeBatch, ClientCacheChangeDummyDiscoveryMessage

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java

@@ -33,10 +33,16 @@
 import org.apache.ignite.internal.processors.authentication.UserSerializer;
 import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
 import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessageSerializer;
+import org.apache.ignite.internal.processors.cache.CacheStatisticsClearMessage;
+import org.apache.ignite.internal.processors.cache.CacheStatisticsClearMessageSerializer;
 import org.apache.ignite.internal.processors.cache.CacheStatisticsModeChangeMessage;
 import org.apache.ignite.internal.processors.cache.CacheStatisticsModeChangeMessageSerializer;
 import org.apache.ignite.internal.processors.cache.ClientCacheChangeDiscoveryMessage;
 import org.apache.ignite.internal.processors.cache.ClientCacheChangeDiscoveryMessageSerializer;
+import org.apache.ignite.internal.processors.cache.ClientCacheChangeDummyDiscoveryMessage;
+import org.apache.ignite.internal.processors.cache.ClientCacheChangeDummyDiscoveryMessageMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
+import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatchMarshallableSerializer;
 import org.apache.ignite.internal.processors.cache.TxTimeoutOnPartitionMapExchangeChangeMessage;
 import org.apache.ignite.internal.processors.cache.TxTimeoutOnPartitionMapExchangeChangeMessageSerializer;
 import org.apache.ignite.internal.processors.cache.WalStateFinishMessage;
@@ -85,6 +91,8 @@
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersionSerializer;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessageSerializer;
+import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
+import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessageMarshallableSerializer;
 import org.apache.ignite.internal.processors.continuous.StopRoutineAckDiscoveryMessage;
 import org.apache.ignite.internal.processors.continuous.StopRoutineAckDiscoveryMessageSerializer;
 import org.apache.ignite.internal.processors.continuous.StopRoutineDiscoveryMessage;
@@ -330,5 +338,12 @@ public DiscoveryMessageFactory(Marshaller marsh, ClassLoader clsLdr) {
         factory.register((short)529, SnapshotCheckHandlersNodeResponse::new, new SnapshotCheckHandlersNodeResponseSerializer());
         factory.register((short)530, SnapshotPartitionsVerifyHandlerResponse::new,
             new SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer(marsh, clsLdr));
+        factory.register((short)531, CacheStatisticsClearMessage::new, new CacheStatisticsClearMessageSerializer());
+        factory.register((short)532, ChangeGlobalStateMessage::new,
+            new ChangeGlobalStateMessageMarshallableSerializer(marsh, clsLdr));
+        factory.register((short)533, ClientCacheChangeDummyDiscoveryMessage::new,
+            new ClientCacheChangeDummyDiscoveryMessageMarshallableSerializer(marsh, clsLdr));
+        factory.register((short)534, DynamicCacheChangeBatch::new,
+            new DynamicCacheChangeBatchMarshallableSerializer(marsh, clsLdr));
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheStatisticsClearMessage.java

@@ -19,32 +19,45 @@
 
 import java.util.Collection;
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Cache statistics clear discovery message.
  */
-public class CacheStatisticsClearMessage implements DiscoveryCustomMessage {
+public class CacheStatisticsClearMessage implements DiscoveryCustomMessage, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** Initial message flag mask. */
     private static final byte INITIAL_MSG_MASK = 0x01;
 
     /** Custom message ID. */
-    private final IgniteUuid id = IgniteUuid.randomUuid();
+    @Order(0)
+    IgniteUuid id;
 
     /** Request id. */
-    private final UUID reqId;
+    @Order(1)
+    UUID reqId;
 
     /** Cache names. */
-    private final Collection<String> caches;
+    @Order(2)
+    Collection<String> caches;
 
     /** Flags. */
-    private final byte flags;
+    @Order(3)
+    byte flags;
+
+    /**
+     * Default constructor.
+     */
+    public CacheStatisticsClearMessage() {
+        // No-op.
+    }
 
     /**
      * Constructor for request.
@@ -54,7 +67,9 @@
     public CacheStatisticsClearMessage(UUID reqId, Collection<String> caches) {
         this.reqId = reqId;
         this.caches = caches;
-        this.flags = INITIAL_MSG_MASK;
+
+        id = IgniteUuid.randomUuid();
+        flags = INITIAL_MSG_MASK;
     }
 
     /**
@@ -63,26 +78,32 @@
      * @param msg Request message.
      */
     private CacheStatisticsClearMessage(CacheStatisticsClearMessage msg) {
-        this.reqId = msg.reqId;
-        this.caches = null;
-        this.flags = 0;
+        id = IgniteUuid.randomUuid();
+        reqId = msg.reqId;
+        caches = null;
+        flags = 0;
     }
 
     /** {@inheritDoc} */
     @Override public IgniteUuid id() {
-        return this.id;
+        return id;
     }
 
     /** {@inheritDoc} */
     @Nullable @Override public DiscoveryCustomMessage ackMessage() {
         return initial() ? new CacheStatisticsClearMessage(this) : null;
     }
 
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 531;
+    }
+
     /**
      * @return Cache names.
      */
     public Collection<String> caches() {
-        return this.caches;
+        return caches;
     }
 
     /**
@@ -96,7 +117,7 @@
      * @return Request id.
      */
     public UUID requestId() {
-        return this.reqId;
+        return reqId;
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDummyDiscoveryMessage.java

@@ -20,30 +20,46 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.processors.security.SecurityContext;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Dummy discovery message which is not really sent via ring, it is just added in local discovery worker queue.
  */
 public class ClientCacheChangeDummyDiscoveryMessage extends AbstractCachePartitionExchangeWorkerTask
-    implements DiscoveryCustomMessage {
+    implements DiscoveryCustomMessage, MarshallableMessage {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private final UUID reqId;
+    @Order(0)
+    UUID reqId;
 
     /** */
-    private final Map<String, DynamicCacheChangeRequest> startReqs;
+    Map<String, DynamicCacheChangeRequest> startReqs;
+
+    /** JDK Serialized version of startReqs. */
+    @Order(1)
+    byte[] startRequestsBytes;
 
     /** */
     @GridToStringInclude
-    private final Set<String> cachesToClose;
+    @Order(2)
+    Set<String> cachesToClose;
+
+    /** */
+    public ClientCacheChangeDummyDiscoveryMessage() {
+        super(null);
+    }
 
     /**
      * @param secCtx Security context in which current task must be executed.
@@ -103,6 +119,23 @@
         throw new UnsupportedOperationException();
     }
 
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (startReqs != null)
+            startRequestsBytes = U.marshal(marsh, startReqs);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        if (startRequestsBytes != null)
+            startReqs = U.unmarshal(marsh, startRequestsBytes, clsLdr);
+    }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 533;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(ClientCacheChangeDummyDiscoveryMessage.class, this,

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeBatch.java

@@ -19,6 +19,8 @@
 
 import java.util.Collection;
 import java.util.Set;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
@@ -28,42 +30,58 @@
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Cache change batch.
  */
-public class DynamicCacheChangeBatch implements DiscoveryCustomMessage {
+public class DynamicCacheChangeBatch implements DiscoveryCustomMessage, MarshallableMessage {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** Discovery custom message ID. */
-    private IgniteUuid id = IgniteUuid.randomUuid();
+    @Order(0)
+    IgniteUuid id;
 
     /** Change requests. */
     @GridToStringInclude
-    private Collection<DynamicCacheChangeRequest> reqs;
+    Collection<DynamicCacheChangeRequest> reqs;
+
+    /** JDK Serialized version of reqs. */
+    @Order(1)
+    byte[] requestsBytes;
 
     /** Cache updates to be executed on exchange. */
-    private transient ExchangeActions exchangeActions;
+    private ExchangeActions exchangeActions;
 
     /** */
-    private boolean startCaches;
+    @Order(2)
+    boolean startCaches;
 
     /** Restarting caches. */
-    private Set<String> restartingCaches;
+    @Order(3)
+    Set<String> restartingCaches;
 
     /** Affinity (cache related) services updates to be processed on services deployment process. */
     @GridToStringExclude
-    @Nullable private transient ServiceDeploymentActions serviceDeploymentActions;
+    @Nullable private ServiceDeploymentActions serviceDeploymentActions;
+
+    /** */
+    public DynamicCacheChangeBatch() {
+        // No-op.
+    }
 
     /**
      * @param reqs Requests.
      */
     public DynamicCacheChangeBatch(Collection<DynamicCacheChangeRequest> reqs) {
         assert !F.isEmpty(reqs) : reqs;
 
+        id = IgniteUuid.randomUuid();
         this.reqs = reqs;
     }
 
@@ -157,6 +175,23 @@
         this.startCaches = startCaches;
     }
 
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (reqs != null)
+            requestsBytes = U.marshal(marsh, reqs);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        if (requestsBytes != null)
+            reqs = U.unmarshal(marsh, requestsBytes, clsLdr);
+    }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 534;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(DynamicCacheChangeBatch.class, this);