Use MessageSerializer to transfer continuous query DTO's in DataBag optimized

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -179,7 +179,9 @@
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse;
 import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
 import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryBatchAck;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryDeployableObject;
 import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryEntry;
+import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
 import org.apache.ignite.internal.processors.cache.transactions.TxEntryValueHolder;
@@ -201,7 +203,13 @@
 import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
 import org.apache.ignite.internal.processors.cluster.NodeFullMetricsMessage;
 import org.apache.ignite.internal.processors.cluster.NodeMetricsMessage;
+import org.apache.ignite.internal.processors.continuous.ContinousRoutineDiscoveryData;
+import org.apache.ignite.internal.processors.continuous.ContinousRoutineDiscoveryDataItem;
+import org.apache.ignite.internal.processors.continuous.ContinousRoutineLocalInfo;
+import org.apache.ignite.internal.processors.continuous.ContinuousRoutineInfo;
 import org.apache.ignite.internal.processors.continuous.ContinuousRoutineStartResultMessage;
+import org.apache.ignite.internal.processors.continuous.ContinuousRoutinesCommonDiscoveryData;
+import org.apache.ignite.internal.processors.continuous.ContinuousRoutinesJoiningNodeDiscoveryData;
 import org.apache.ignite.internal.processors.continuous.GridContinuousMessage;
 import org.apache.ignite.internal.processors.continuous.StartRequestData;
 import org.apache.ignite.internal.processors.continuous.StartRoutineAckDiscoveryMessage;
@@ -627,6 +635,16 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) {
         register(QueryProposalsDataBagItem.class);
         register(QueryEntityMessage.class);
         register(QueryEntityExMessage.class);
+        register(ContinuousRoutineInfo.class);
+        register(ContinuousRoutinesJoiningNodeDiscoveryData.class);
+        register(CacheContinuousQueryDeployableObject.class);
+        register(CacheContinuousQueryHandler.class);
+        register(GridEventConsumeHandler.class);
+        register(GridMessageListenHandler.class);
+        register(ContinousRoutineLocalInfo.class);
+        register(ContinousRoutineDiscoveryDataItem.class);
+        register(ContinousRoutineDiscoveryData.class);
+        register(ContinuousRoutinesCommonDiscoveryData.class);
 
         // [11200 - 11300]: Compute, distributed process messages.
         msgIdx = 11200;

modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java

@@ -61,12 +61,9 @@
 /**
  * Continuous routine handler for remote event listening.
  */
-class GridEventConsumeHandler implements GridContinuousHandler {
-    /** */
-    private static final long serialVersionUID = 0L;
-
+public final class GridEventConsumeHandler implements GridContinuousHandler, MarshallableMessage {
     /** Default callback. */
-    private static final IgniteBiPredicate<UUID, Event> DFLT_CALLBACK = new P2<UUID, Event>() {
+    private static final IgniteBiPredicate<UUID, Event> DFLT_CALLBACK = new P2<>() {
         @Override public boolean apply(UUID uuid, Event e) {
             return true;
         }
@@ -76,28 +73,32 @@ class GridEventConsumeHandler implements GridContinuousHandler {
     private IgniteBiPredicate<UUID, Event> cb;
 
     /** Filter. */
-    private IgnitePredicate<Event> filter;
+    @Nullable volatile IgnitePredicate<Event> filter;
 
-    /** Serialized filter. */
-    private byte[] filterBytes;
+    /** Marshaled {@link #filter}. */
+    @Order(0)
+    @Nullable volatile byte[] filterBytes;
 
-    /** Deployment class name. */
-    private String clsName;
+    /** Deployment class name. Is {@code null} if P2P deployment is disabled. */
+    @Order(1)
+    @Nullable volatile String clsName;
 
-    /** Deployment info. */
-    private GridDeploymentInfo depInfo;
+    /** Deployment info. Is {@code null} if P2P deployment is disabled. */
+    @Order(2)
+    @Nullable volatile GridDeploymentInfoBean depInfo;
 
     /** Types. */
-    private int[] types;
+    @Order(3)
+    int[] types;
 
     /** Listener. */
     private GridLocalEventListener lsnr;
 
     /** P2P unmarshalling future. */
-    private IgniteInternalFuture<Void> p2pUnmarshalFut = new GridFinishedFuture<>();
+    private volatile IgniteInternalFuture<Void> p2pUnmarshalFut = new GridFinishedFuture<>();
 
     /**
-     * Required by {@link Externalizable}.
+     * Empty constructor for serialization purposes.
      */
     public GridEventConsumeHandler() {
         // No-op.
@@ -225,8 +226,6 @@ private void initFilter(IgnitePredicate<Event> filter, GridKernalContext ctx) th
                                                 EventWrapper wrapper = new EventWrapper(evt);
 
                                                 if (evt instanceof CacheEvent) {
-                                                    String cacheName = ((CacheEvent)evt).cacheName();
-
                                                     ClusterNode node = ctx.discovery().node(t3.get1());
 
                                                     if (node == null)
@@ -393,6 +392,10 @@ private boolean filterDropsEvent(Event evt) {
         assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
 
+        // TODO : Remove this check after https://issues.apache.org/jira/browse/IGNITE-28945
+        if (filterBytes != null)
+            return;
+
         if (filter != null) {
             Class cls = U.detectClass(filter);
 
@@ -415,6 +418,10 @@ private boolean filterDropsEvent(Event evt) {
         assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
 
+        // TODO : Remove this check after https://issues.apache.org/jira/browse/IGNITE-28945
+        if (filter != null)
+            return;
+
         if (filterBytes != null) {
             try {
                 GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(), clsName, clsName,
@@ -471,36 +478,27 @@ private boolean filterDropsEvent(Event evt) {
     }
 
     /** {@inheritDoc} */
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        boolean b = filterBytes != null;
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
+        assert (clsName == null) == (depInfo == null);
 
-        out.writeBoolean(b);
-
-        if (b) {
-            U.writeByteArray(out, filterBytes);
-            U.writeString(out, clsName);
-            out.writeObject(depInfo);
-        }
-        else
-            out.writeObject(filter);
-
-        out.writeObject(types);
+        /** Are marshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */
+        if (filter != null && depInfo == null)
+            filterBytes = marsh.marshal(filter);
     }
 
     /** {@inheritDoc} */
-    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        boolean b = in.readBoolean();
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        assert (clsName == null) == (depInfo == null);
 
-        if (b) {
+        /** Are unmarshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */
+        if (depInfo != null) {
             p2pUnmarshalFut = new GridFutureAdapter<>();
-            filterBytes = U.readByteArray(in);
-            clsName = U.readString(in);
-            depInfo = (GridDeploymentInfo)in.readObject();
+
+            return;
         }
-        else
-            filter = (IgnitePredicate<Event>)in.readObject();
 
-        types = (int[])in.readObject();
+        if (filterBytes != null)
+            filter = marsh.unmarshal(filterBytes, clsLdr);
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/GridMessageListenHandler.java

@@ -17,10 +17,6 @@
 
 package org.apache.ignite.internal;
 
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
@@ -39,41 +35,40 @@
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.marshaller.Marshaller;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Continuous handler for message subscription.
  */
-public class GridMessageListenHandler implements GridContinuousHandler {
+public final class GridMessageListenHandler implements GridContinuousHandler, MarshallableMessage {
     /** */
-    private static final long serialVersionUID = 0L;
+    private volatile @Nullable Object topic;
 
-    /** */
-    private Object topic;
+    /** Marshalled {@link #topic}. */
+    @Order(0)
+    @Nullable volatile byte[] topicBytes;
 
     /** */
-    private IgniteBiPredicate<UUID, Object> pred;
+    private volatile IgniteBiPredicate<UUID, Object> pred;
 
-    /** */
-    private byte[] topicBytes;
+    /** Marshalled {@link #pred}. */
+    @Order(1)
+    volatile byte[] predBytes;
 
-    /** */
-    private byte[] predBytes;
+    /** Class name of {@link #pred}. Is {@code null} if the P2P deployment is disabled. */
+    @Order(2)
+    @Nullable volatile String clsName;
 
-    /** */
-    private String clsName;
-
-    /** */
-    private GridDeploymentInfoBean depInfo;
-
-    /** */
-    private boolean depEnabled;
+    /** P2P deploy info of {@link #pred}. Is {@code null} if the P2P deployment is disabled. */
+    @Order(3)
+    @Nullable volatile GridDeploymentInfoBean predDepInfo;
 
     /** P2P unmarshalling future. */
-    private IgniteInternalFuture<Void> p2pUnmarshalFut = new GridFinishedFuture<>();
+    private volatile IgniteInternalFuture<Void> p2pUnmarshalFut = new GridFinishedFuture<>();
 
     /**
-     * Required by {@link Externalizable}.
+     * Empty constructor for serialization purposes
      */
     public GridMessageListenHandler() {
         // No-op.
@@ -151,6 +146,10 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
 
+        // TODO : Remove this check after https://issues.apache.org/jira/browse/IGNITE-28945
+        if (predDepInfo != null)
+            return;
+
         if (topic != null)
             topicBytes = U.marshal(ctx.marshaller(), topic);
 
@@ -166,9 +165,19 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         if (dep == null)
             throw new IgniteDeploymentCheckedException("Failed to deploy message listener.");
 
-        depInfo = new GridDeploymentInfoBean(dep);
+        predDepInfo = new GridDeploymentInfoBean(dep);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
+        /** Are marshaled in {@link #p2pMarshal(GridKernalContext)}. */
+        if (predDepInfo != null)
+            return;
+
+        if (topic != null)
+            topicBytes = marsh.marshal(topic);
 
-        depEnabled = true;
+        predBytes = marsh.marshal(pred);
     }
 
     /** {@inheritDoc} */
@@ -177,9 +186,13 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
 
+        // TODO : Remove this check after https://issues.apache.org/jira/browse/IGNITE-28945
+        if (pred != null)
+            return;
+
         try {
-            GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(), clsName, clsName,
-                depInfo.userVersion(), nodeId, depInfo.classLoaderId(), depInfo.participants(), null);
+            GridDeployment dep = ctx.deploy().getGlobalDeployment(predDepInfo.deployMode(), clsName, clsName,
+                predDepInfo.userVersion(), nodeId, predDepInfo.classLoaderId(), predDepInfo.participants(), null);
 
             if (dep == null)
                 throw new IgniteDeploymentCheckedException("Failed to obtain deployment for class: " + clsName);
@@ -205,6 +218,21 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         ((GridFutureAdapter)p2pUnmarshalFut).onDone();
     }
 
+    /** {@inheritDoc} */
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        /** Are unmarshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */
+        if (predDepInfo != null) {
+            p2pUnmarshalFut = new GridFutureAdapter<>();
+
+            return;
+        }
+
+        if (topicBytes != null)
+            topic = marsh.unmarshal(topicBytes, clsLdr);
+
+        pred = marsh.unmarshal(predBytes, clsLdr);
+    }
+
     /** {@inheritDoc} */
     @Override public GridContinuousBatch createBatch() {
         return new GridContinuousBatchAdapter();
@@ -235,39 +263,6 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         }
     }
 
-    /** {@inheritDoc} */
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        out.writeBoolean(depEnabled);
-
-        if (depEnabled) {
-            U.writeByteArray(out, topicBytes);
-            U.writeByteArray(out, predBytes);
-            U.writeString(out, clsName);
-            out.writeObject(depInfo);
-        }
-        else {
-            out.writeObject(topic);
-            out.writeObject(pred);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        depEnabled = in.readBoolean();
-
-        if (depEnabled) {
-            p2pUnmarshalFut = new GridFutureAdapter<>();
-            topicBytes = U.readByteArray(in);
-            predBytes = U.readByteArray(in);
-            clsName = U.readString(in);
-            depInfo = (GridDeploymentInfoBean)in.readObject();
-        }
-        else {
-            topic = in.readObject();
-            pred = (IgniteBiPredicate<UUID, Object>)in.readObject();
-        }
-    }
-
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridMessageListenHandler.class, this);

modules/core/src/main/java/org/apache/ignite/internal/managers/deployment/GridDeploymentInfoBean.java

@@ -30,7 +30,7 @@
 /**
  * Deployment info bean.
  */
-public class GridDeploymentInfoBean implements Message, GridDeploymentInfo, Serializable {
+public final class GridDeploymentInfoBean implements Message, GridDeploymentInfo, Serializable {
     /** */
     private static final long serialVersionUID = 0L;
 

modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityUtils.java

@@ -86,7 +86,7 @@ private static GridAffinityMessage affinityMessage(GridKernalContext ctx, Object
     }
 
     /**
-     * Unmarshalls transfer object from remote node within a given context.
+     * Unmarshals transfer object from remote node within a given context.
      *
      * @param ctx Grid kernal context that provides deployment and marshalling services.
      * @param sndNodeId {@link UUID} of the sender node.