IGNITE-28790 Use MessageSerializer to transfer continuous query DTO's in DataBag

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

@@ -175,7 +175,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;
@@ -193,7 +195,12 @@
 import org.apache.ignite.internal.processors.cluster.ClusterUpdateNotifierDataBagItem;
 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.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;
@@ -613,6 +620,15 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) {
         withNoSchema(QueryProposalsDataBagItem.class);
         withNoSchema(QueryEntityMessage.class);
         withNoSchema(QueryEntityExMessage.class);
+        withNoSchema(ContinuousRoutineInfo.class);
+        withNoSchema(ContinuousRoutinesJoiningNodeDiscoveryData.class);
+        withNoSchema(CacheContinuousQueryDeployableObject.class);
+        withSchema(CacheContinuousQueryHandler.class);
+        withSchema(GridEventConsumeHandler.class);
+        withSchema(GridMessageListenHandler.class);
+        withSchema(ContinousRoutineLocalInfo.class);
+        withSchema(ContinousRoutineDiscoveryDataItem.class);
+        withNoSchema(ContinousRoutineDiscoveryData.class);
 
         // [11200 - 11300]: Compute, distributed process messages.
         msgIdx = 11200;

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

@@ -61,12 +61,10 @@
 /**
  * Continuous routine handler for remote event listening.
  */
-class GridEventConsumeHandler implements GridContinuousHandler {
-    /** */
-    private static final long serialVersionUID = 0L;
-
+@UseBinaryMarshaller
+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 +74,41 @@ 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;
+
+    /**
+     * Lever of own marshaling.
+     *
+     * @see #p2pMarshal(GridKernalContext)
+     * @see #marshal(Marshaller)
+     */
+    @Order(4)
+    volatile boolean externalMarshal;
 
     /** 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 +236,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)
@@ -390,11 +399,12 @@ private boolean filterDropsEvent(Event evt) {
 
     /** {@inheritDoc} */
     @Override public void p2pMarshal(GridKernalContext ctx) throws IgniteCheckedException {
-        assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
 
         if (filter != null) {
-            Class cls = U.detectClass(filter);
+            assert filterBytes == null : "Duplicated p2p-marshalling, " + getClass().getSimpleName();
+
+            Class<?> cls = U.detectClass(filter);
 
             clsName = cls.getName();
 
@@ -406,16 +416,22 @@ private boolean filterDropsEvent(Event evt) {
             depInfo = new GridDeploymentInfoBean(dep);
 
             filterBytes = U.marshal(ctx.marshaller(), filter);
+
+            externalMarshal = true;
         }
     }
 
     /** {@inheritDoc} */
     @Override public void p2pUnmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException {
         assert nodeId != null;
-        assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
+        assert externalMarshal : "Is not p2p-marshaled " + getClass().getSimpleName();
+        assert !p2pUnmarshalFut.isDone() && p2pUnmarshalFut instanceof GridFutureAdapter :
+            "Can't p2p-unmarshal, the p2p-umarshalling future seems to be already done, " + getClass().getSimpleName();
 
         if (filterBytes != null) {
+            assert filter == null : "Duplicated p2p-unmarshalling, " + getClass().getSimpleName();
+
             try {
                 GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(), clsName, clsName,
                     depInfo.userVersion(), nodeId, depInfo.classLoaderId(), depInfo.participants(), null);
@@ -471,36 +487,28 @@ private boolean filterDropsEvent(Event evt) {
     }
 
     /** {@inheritDoc} */
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        boolean b = filterBytes != null;
-
-        out.writeBoolean(b);
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
+        assert (clsName == null) == (depInfo == null);
+        assert (depInfo == null) == !externalMarshal;
 
-        if (b) {
-            U.writeByteArray(out, filterBytes);
-            U.writeString(out, clsName);
-            out.writeObject(depInfo);
-        }
-        else
-            out.writeObject(filter);
-
-        out.writeObject(types);
+        if (filter != null && !externalMarshal)
+            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);
+        assert (depInfo == null) == !externalMarshal;
 
-        if (b) {
+        /** Are unmarshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */
+        if (externalMarshal) {
             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,49 @@
 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 {
+@UseBinaryMarshaller
+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;
+    /** P2P deploy info of {@link #pred}. Is {@code null} if the P2P deployment is disabled. */
+    @Order(3)
+    @Nullable volatile GridDeploymentInfoBean predDepInfo;
 
-    /** */
-    private GridDeploymentInfoBean depInfo;
-
-    /** */
-    private boolean depEnabled;
+    /**
+     * Lever of the own marshaling.
+     *
+     * @see #p2pMarshal(GridKernalContext)
+     */
+    @Order(4)
+    volatile boolean externalMarshal;
 
     /** 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.
@@ -166,20 +170,36 @@ 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);
 
-        depEnabled = true;
+        externalMarshal = true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
+        assert externalMarshal == (clsName != null);
+        assert externalMarshal == (predDepInfo != null);
+
+        /** Are marshaled in {@link #p2pMarshal(GridKernalContext)}. */
+        if (externalMarshal)
+            return;
+
+        assert predBytes == null;
+
+        topicBytes = marsh.marshal(topic);
+        predBytes = marsh.marshal(pred);
     }
 
     /** {@inheritDoc} */
     @Override public void p2pUnmarshal(UUID nodeId, GridKernalContext ctx) throws IgniteCheckedException {
         assert nodeId != null;
         assert ctx != null;
         assert ctx.config().isPeerClassLoadingEnabled();
+        assert externalMarshal : "Is not p2p-marshaled " + getClass().getSimpleName();
 
         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 +225,23 @@ public GridMessageListenHandler(@Nullable Object topic, IgniteBiPredicate<UUID,
         ((GridFutureAdapter)p2pUnmarshalFut).onDone();
     }
 
+    /** {@inheritDoc} */
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+        assert externalMarshal == (clsName != null);
+        assert externalMarshal == (predDepInfo != null);
+        assert predBytes != null;
+
+        /** Are unmarshaled in {@link #p2pUnmarshal(UUID, GridKernalContext)}. */
+        if (externalMarshal) {
+            p2pUnmarshalFut = new GridFutureAdapter<>();
+
+            return;
+        }
+
+        topic = marsh.unmarshal(topicBytes, clsLdr);
+        pred = marsh.unmarshal(predBytes, clsLdr);
+    }
+
     /** {@inheritDoc} */
     @Override public GridContinuousBatch createBatch() {
         return new GridContinuousBatchAdapter();
@@ -235,39 +272,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;