IGNITE-28528 Move the start request conversion out of the message

modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java

@@ -981,7 +981,7 @@ private AbstractContinuousMessage createStartMessage(UUID routineId,
             reqData.deploymentInfo(dep);
         }
 
-        reqData.marshal(ctx);
+        marshalStartRequestData(reqData);
 
         if (!immutableDiscoCustomMsg) {
             StartRoutineDiscoveryMessage msg = new StartRoutineDiscoveryMessage(routineId, reqData, Mode.MUTABLE);
@@ -1353,7 +1353,7 @@ private void processStartRequestMutable(ClusterNode node, StartRoutineDiscoveryM
         IgniteCheckedException err = null;
 
         try {
-            data.unmarshal(ctx, node.id());
+            unmarshalStartRequestData(data, node.id());
         }
         catch (IgniteCheckedException e) {
             U.error(log, "Failed to unmarshal start request data [nodeId=" + node.id() +
@@ -1495,7 +1495,7 @@ private void processStartRequestImmutable(final AffinityTopologyVersion topVer,
                 Exception err = null;
 
                 try {
-                    reqData.unmarshal(ctx, snd.id());
+                    unmarshalStartRequestData(reqData, snd.id());
                 }
                 catch (IgniteCheckedException e) {
                     err = e;
@@ -2286,6 +2286,73 @@ IgniteBiTuple<GridContinuousBatch, Long> checkInterval() {
         }
     }
 
+    /**
+     * Serializes the user objects of {@code reqData} before it goes into a discovery message. Must not run on the
+     * discovery thread: marshalling a user class registers its name, and that registration waits for a discovery
+     * message of its own.
+     *
+     * @param reqData Start request data.
+     * @throws IgniteCheckedException If failed.
+     */
+    private void marshalStartRequestData(StartRequestData reqData) throws IgniteCheckedException {
+        if (reqData.hnd != null) {
+            if (ctx.config().isPeerClassLoadingEnabled()) {
+                // Handle peer deployment for the objects the handler carries.
+                reqData.hnd.p2pMarshal(ctx);
+            }
+
+            reqData.hndBytes = U.marshal(ctx.marshaller(), reqData.hnd);
+        }
+
+        if (reqData.nodeFilter != null)
+            reqData.nodeFilterBytes = U.marshal(ctx.marshaller(), reqData.nodeFilter);
+    }
+
+    /**
+     * Restores the user objects of {@code reqData} received from {@code sndId}. With peer class loading on, the node
+     * filter needs the deployment class loader of the sender, so the deployment is resolved first. Each object is put
+     * back into the message as soon as it is read: the caller reports a failure of the steps that follow and needs
+     * what was restored before it.
+     *
+     * @param reqData Start request data.
+     * @param sndId Sender node ID.
+     * @throws IgniteCheckedException If failed.
+     */
+    private void unmarshalStartRequestData(StartRequestData reqData, UUID sndId) throws IgniteCheckedException {
+        ClassLoader clsLdr = U.resolveClassLoader(ctx.config());
+
+        if (ctx.config().isPeerClassLoadingEnabled() && reqData.clsName != null) {
+            GridDeployment dep = ctx.deploy().getGlobalDeployment(reqData.depInfo.deployMode(),
+                reqData.clsName,
+                reqData.clsName,
+                reqData.depInfo.userVersion(),
+                sndId,
+                reqData.depInfo.classLoaderId(),
+                reqData.depInfo.participants(),
+                null);
+
+            if (dep == null)
+                throw new IgniteDeploymentCheckedException("Failed to obtain deployment for class: " + reqData.clsName);
+
+            clsLdr = U.resolveClassLoader(dep.classLoader(), ctx.config());
+        }
+
+        reqData.nodeFilter = U.unmarshal(ctx.marshaller(), reqData.nodeFilterBytes, clsLdr);
+
+        if (reqData.hndBytes != null) {
+            reqData.hnd = U.unmarshal(ctx.marshaller(), reqData.hndBytes, U.resolveClassLoader(ctx.config()));
+
+            if (ctx.config().isPeerClassLoadingEnabled())
+                reqData.hnd.p2pUnmarshal(sndId, ctx);
+
+            if (reqData.keepBinary) {
+                assert reqData.hnd instanceof CacheContinuousQueryHandler : reqData.hnd;
+
+                ((CacheContinuousQueryHandler<?, ?>)reqData.hnd).keepBinary(true);
+            }
+        }
+    }
+
     /**
      * Discovery data.
      */
@@ -2695,4 +2762,5 @@ UUID nodeId() {
             return S.toString(SyncMessageAckFuture.class, this);
         }
     }
+
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java

@@ -17,17 +17,10 @@
 
 package org.apache.ignite.internal.processors.continuous;
 
-import java.util.UUID;
-import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.internal.IgniteDeploymentCheckedException;
 import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.managers.deployment.GridDeployment;
 import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
-import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
 import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgnitePredicate;
 import org.apache.ignite.plugin.extensions.communication.Message;
 
@@ -36,7 +29,7 @@
  */
 public class StartRequestData implements Message {
     /** Node filter. */
-    private IgnitePredicate<ClusterNode> nodeFilter;
+    IgnitePredicate<ClusterNode> nodeFilter;
 
     /** Serialized node filter. */
     @Order(0)
@@ -51,7 +44,7 @@ public class StartRequestData implements Message {
     GridDeploymentInfoBean depInfo;
 
     /** Handler. */
-    private GridContinuousHandler hnd;
+    GridContinuousHandler hnd;
 
     /** Serialized handler. */
     @Order(3)
@@ -169,58 +162,4 @@ public void autoUnsubscribe(boolean autoUnsubscribe) {
     @Override public String toString() {
         return S.toString(StartRequestData.class, this);
     }
-
-    /** */
-    public void marshal(GridKernalContext ctx) throws IgniteCheckedException {
-        if (hnd != null) {
-            if (ctx.config().isPeerClassLoadingEnabled()) {
-                // Handle peer deployment for other handler-specific objects.
-                hnd.p2pMarshal(ctx);
-            }
-
-            hndBytes = U.marshal(ctx.marshaller(), hnd);
-        }
-
-        if (nodeFilter != null)
-            nodeFilterBytes = U.marshal(ctx.marshaller(), nodeFilter);
-    }
-
-    /** */
-    public void unmarshal(GridKernalContext ctx, UUID sndId) throws IgniteCheckedException {
-        if (ctx.config().isPeerClassLoadingEnabled() && clsName != null) {
-            GridDeployment dep = ctx.deploy().getGlobalDeployment(depInfo.deployMode(),
-                clsName,
-                clsName,
-                depInfo.userVersion(),
-                sndId,
-                depInfo.classLoaderId(),
-                depInfo.participants(),
-                null);
-
-            if (dep == null)
-                throw new IgniteDeploymentCheckedException("Failed to obtain deployment for class: " + clsName);
-
-            nodeFilter = U.unmarshal(ctx.marshaller(),
-                nodeFilterBytes,
-                U.resolveClassLoader(dep.classLoader(), ctx.config()));
-        }
-        else {
-            nodeFilter = U.unmarshal(ctx.marshaller(),
-                nodeFilterBytes,
-                U.resolveClassLoader(ctx.config()));
-        }
-
-        if (hndBytes != null) {
-            hnd = U.unmarshal(ctx.marshaller(), hndBytes, U.resolveClassLoader(ctx.config()));
-
-            if (ctx.config().isPeerClassLoadingEnabled())
-                hnd.p2pUnmarshal(sndId, ctx);
-
-            if (keepBinary) {
-                assert hnd instanceof CacheContinuousQueryHandler : hnd;
-
-                ((CacheContinuousQueryHandler<?, ?>)hnd).keepBinary(true);
-            }
-        }
-    }
 }