diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java index fecd065ff7551..4b3d66653341a 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java +++ b/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 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); } } + } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java index 9571c231bbf9a..acb2fbb6a8dfe 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/StartRequestData.java +++ b/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 nodeFilter; + IgnitePredicate 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); - } - } - } }