From c11ea39433d0d4dacf151c724a36288b3e73a4de Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Wed, 5 Aug 2026 00:08:44 +0300 Subject: [PATCH] IGNITE-28271 Use the generated marshalling for GridJobExecuteResponse Co-Authored-By: Claude Opus 5 --- .../internal/GridJobExecuteResponse.java | 140 +++--------------- .../processors/job/GridJobProcessor.java | 18 ++- .../processors/job/GridJobWorker.java | 39 ++++- .../processors/task/GridTaskWorker.java | 9 +- 4 files changed, 67 insertions(+), 139 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteResponse.java index 0f24d1effc596..6e1c14fe240f9 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteResponse.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteResponse.java @@ -19,25 +19,18 @@ import java.util.Map; import java.util.UUID; -import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; -import org.apache.ignite.IgniteLogger; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.util.tostring.GridToStringExclude; -import org.apache.ignite.internal.util.typedef.F; -import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.S; -import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteUuid; -import org.apache.ignite.marshaller.Marshaller; -import org.apache.ignite.plugin.extensions.communication.Message; import org.jetbrains.annotations.Nullable; /** * Job execution response. */ @UseBinaryMarshaller -public class GridJobExecuteResponse implements Message { +public class GridJobExecuteResponse implements DeferredUnmarshalMessage { /** */ @Order(0) UUID nodeId; @@ -50,28 +43,33 @@ public class GridJobExecuteResponse implements Message { @Order(2) IgniteUuid jobId; - /** Job result exception call holder. */ + /** */ + @GridToStringExclude + @Marshalled("gridExBytes") + @Nullable IgniteException gridEx; + + /** */ @Order(3) @Nullable byte[] gridExBytes; /** */ - private IgniteException gridEx; + @GridToStringExclude + @Marshalled("resBytes") + @Nullable Object res; - /** Job result serialization call holder. */ + /** */ @Order(4) @Nullable byte[] resBytes; /** */ - private @Nullable Object res; + @GridToStringExclude + @Marshalled("jobAttrsBytes") + Map jobAttrs; /** */ - /** Job attributes serialization call holder. */ @Order(5) byte[] jobAttrsBytes; - /** */ - private Map jobAttrs; - /** */ @Order(6) boolean isCancelled; @@ -145,23 +143,11 @@ public IgniteUuid jobId() { return res; } - /** - * @return Job exception. - */ + /** @return Job exception. */ @Nullable public IgniteException exception() { return gridEx; } - /** */ - public void exceptionBytes(@Nullable byte[] gridExBytes) { - this.gridExBytes = gridExBytes; - } - - /** */ - public @Nullable byte[] exceptionBytes() { - return gridExBytes; - } - /** * @return Job attributes. */ @@ -212,101 +198,11 @@ public boolean retry() { return retry; } - /** - * Serializes user data to byte[] with provided marshaller. - * Erases non-marshalled data like {@link #getJobAttributes()} or {@link #getJobResult()}. - */ - public void marshallUserData(Marshaller marsh, @Nullable IgniteLogger log) throws IgniteCheckedException { - if (res != null) { - try { - resBytes = U.marshal(marsh, res); - } - catch (IgniteCheckedException e) { - resBytes = null; - - String msg = "Failed to serialize job response [nodeId=" + nodeId + - ", ses=" + sesId + ", jobId=" + jobId + - ", resCls=" + (res == null ? null : res.getClass()) + ']'; - - wrapSerializationError(e, msg, log); - } - - res = null; - } - - if (!F.isEmpty(jobAttrs)) { - try { - jobAttrsBytes = U.marshal(marsh, jobAttrs); - } - catch (IgniteCheckedException e) { - jobAttrsBytes = null; - - String msg = "Failed to serialize job attributes [nodeId=" + nodeId + - ", ses=" + sesId + ", jobId=" + jobId + - ", attrs=" + jobAttrs + ']'; - - wrapSerializationError(e, msg, log); - } - - jobAttrs = null; - } - - if (gridEx != null) { - try { - gridExBytes = U.marshal(marsh, gridEx); - } - catch (IgniteCheckedException e) { - String msg = "Failed to serialize job exception [nodeId=" + nodeId + - ", ses=" + sesId + ", jobId=" + jobId + - ", msg=\"" + e.getMessage() + "\"]"; - - gridEx = new IgniteException(msg); - - U.error(log, msg, e); - - gridExBytes = U.marshal(marsh, gridEx); - } - - gridEx = null; - } - } - - /** - * Deserializes user data from byte[] with provided marshaller and class loader. - * Erases marshalled data like {@link #jobAttrubutesBytes()} or {@link #jobResultBytes()}. - */ - public void unmarshallUserData(Marshaller marshaller, ClassLoader clsLdr) throws IgniteCheckedException { - if (jobAttrsBytes != null) { - jobAttrs = U.unmarshal(marshaller, jobAttrsBytes, clsLdr); - - jobAttrsBytes = null; - } - - if (resBytes != null) { - res = U.unmarshal(marshaller, resBytes, clsLdr); - - resBytes = null; - } - - if (gridExBytes != null) { - gridEx = U.unmarshal(marshaller, gridExBytes, clsLdr); - - gridExBytes = null; - } + /** @return A copy carrying {@code err} and no payload. */ + public GridJobExecuteResponse withError(IgniteException err) { + return new GridJobExecuteResponse(nodeId, sesId, jobId, err, null, null, isCancelled, retry); } - /** */ - private void wrapSerializationError(IgniteCheckedException e, String msg, @Nullable IgniteLogger log) { - if (gridEx != null) - gridEx.addSuppressed(e); - else - gridEx = U.convertException(e); - - if (log != null && (log.isDebugEnabled() || !X.hasCause(e, NodeStoppingException.class))) - U.error(log, msg, e); - } - - /** {@inheritDoc} */ @Override public String toString() { return S.toString(GridJobExecuteResponse.class, this); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java index 4a26d329f5282..b81a8b43bf21e 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java @@ -1617,8 +1617,22 @@ private void handleException(ClusterNode node, GridJobExecuteRequest req, Ignite false, null); - if (!loc) - jobRes.marshallUserData(marsh, log); + if (!loc) { + try { + MessageMarshalling.marshal(jobRes, ctx, null); + } + catch (IgniteCheckedException e) { + // The exception is the only payload of this response, so it is what could not be written. + String errMsg = "Failed to serialize job exception [nodeId=" + sndNode.id() + + ", ses=" + req.sessionId() + ", jobId=" + req.jobId() + ']'; + + U.error(log, errMsg, e); + + jobRes = jobRes.withError(new IgniteException(errMsg)); + + MessageMarshalling.marshal(jobRes, ctx, null); + } + } if (req.sessionFullSupport()) { // Send response to designated job topic. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java index 1ab5b7f1e4f31..167874ceff5d4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java @@ -46,6 +46,7 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.internal.NodeStoppingException; import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; +import org.apache.ignite.internal.managers.communication.MessageMarshalling; import org.apache.ignite.internal.managers.deployment.GridDeployment; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.distributed.dht.GridReservable; @@ -64,7 +65,6 @@ import org.apache.ignite.internal.util.worker.GridWorker; import org.apache.ignite.lang.IgniteBiTuple; import org.apache.ignite.lang.IgniteUuid; -import org.apache.ignite.marshaller.Marshaller; import org.jetbrains.annotations.Nullable; import static org.apache.ignite.events.EventType.EVT_JOB_CANCELLED; @@ -122,9 +122,6 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject { /** */ private final IgniteLogger log; - /** */ - private final Marshaller marsh; - /** */ private final GridJobSessionImpl ses; @@ -245,8 +242,6 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject { log = U.logger(ctx, logRef, this); - marsh = ctx.marshaller(); - UUID locNodeId = ctx.discovery().localNode().id(); jobTopic = TOPIC_JOB.topic(ses.getJobId(), locNodeId); @@ -887,8 +882,36 @@ else if (!internal && ctx.event().isRecordable(EVT_JOB_REJECTED)) isCancelled(), retry ? ctx.cache().context().exchange().readyAffinityVersion() : null); - if (!loc) - jobRes.marshallUserData(marsh, log); + if (!loc) { + try { + MessageMarshalling.marshal(jobRes, ctx, null); + } + catch (IgniteCheckedException e) { + String ids = "[nodeId=" + sndNode.id() + ", ses=" + ses.getId() + + ", jobId=" + ses.getJobId() + ']'; + + logError("Failed to serialize job response " + ids, e); + + // Drop the payload, keeping the job exception when there is one. + jobRes = jobRes.withError(jobRes.exception() != null + ? jobRes.exception() + : U.convertException(e)); + + try { + MessageMarshalling.marshal(jobRes, ctx, null); + } + catch (IgniteCheckedException e0) { + // Then the exception itself is what could not be written. + String errMsg = "Failed to serialize job exception " + ids; + + logError(errMsg, e0); + + jobRes = jobRes.withError(new IgniteException(errMsg)); + + MessageMarshalling.marshal(jobRes, ctx, null); + } + } + } long timeout = ses.getEndTime() - U.currentTimeMillis(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java index 31e7b65bd55b8..0f262d928f5a9 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java @@ -68,6 +68,7 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; import org.apache.ignite.internal.compute.ComputeTaskCancelledCheckedException; import org.apache.ignite.internal.compute.ComputeTaskTimeoutCheckedException; +import org.apache.ignite.internal.managers.communication.MessageMarshalling; import org.apache.ignite.internal.managers.deployment.GridDeployment; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.closure.AffinityTask; @@ -87,7 +88,6 @@ import org.apache.ignite.internal.util.worker.GridWorker; import org.apache.ignite.lang.IgniteInClosure; import org.apache.ignite.lang.IgniteUuid; -import org.apache.ignite.marshaller.Marshaller; import org.apache.ignite.plugin.security.SecurityException; import org.apache.ignite.resources.TaskContinuousMapperResource; import org.jetbrains.annotations.Nullable; @@ -156,9 +156,6 @@ private enum State { /** */ private final IgniteLogger log; - /** */ - private final Marshaller marsh; - /** */ private final GridTaskSessionImpl ses; @@ -325,8 +322,6 @@ private enum State { log = U.logger(ctx, logRef, this); - marsh = ctx.marshaller(); - boolean noResCacheAnnotation = dep.annotation(taskCls, ComputeTaskNoResultCache.class) != null; resCache = !(noResCacheAnnotation || opts.isResultCacheDisabled()); @@ -829,7 +824,7 @@ void onResponse(GridJobExecuteResponse msg) { boolean loc = ctx.localNodeId().equals(res.nodeId()) && !ctx.config().isMarshalLocalJobs(); if (!loc) - res.unmarshallUserData(marsh, U.resolveClassLoader(dep.classLoader(), ctx.config())); + MessageMarshalling.unmarshal(res, ctx, null, U.resolveClassLoader(dep.classLoader(), ctx.config())); jobRes.onResponse(res.getJobResult(), res.exception(), res.getJobAttributes(), res.cancelled());