From 04d113fc8776162b7447a3d31ace64d96ec92972 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Wed, 5 Aug 2026 20:42:19 +0300 Subject: [PATCH] IGNITE-28528 Move the start request conversion out of the message StartRequestData no longer marshals itself: GridContinuousProcessor does it, where the sender id, the deployment and the right thread are already at hand. The message keeps only its fields, and the marshaller is now picked where we marshal. Co-Authored-By: Claude Opus 5 --- .../continuous/GridContinuousProcessor.java | 74 ++++++++++++++++++- .../continuous/StartRequestData.java | 65 +--------------- 2 files changed, 73 insertions(+), 66 deletions(-) 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); - } - } - } }