This is an automated email from the ASF dual-hosted git repository.
anton-vinogradov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 5cbb7fc2c1c IGNITE-28271 Use the generated marshalling for
GridJobExecuteResponse (#13435)
5cbb7fc2c1c is described below
commit 5cbb7fc2c1c4ad3c4a3d943d2bdf283586310eb5
Author: Anton Vinogradov <[email protected]>
AuthorDate: Wed Aug 5 17:08:41 2026 +0300
IGNITE-28271 Use the generated marshalling for GridJobExecuteResponse
(#13435)
---
.../ignite/internal/GridJobExecuteResponse.java | 140 +++------------------
.../internal/processors/job/GridJobProcessor.java | 18 ++-
.../internal/processors/job/GridJobWorker.java | 39 ++++--
.../internal/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 0f24d1effc5..6e1c14fe240 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 @@ package org.apache.ignite.internal;
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<Object, Object> jobAttrs;
/** */
- /** Job attributes serialization call holder. */
@Order(5)
byte[] jobAttrsBytes;
- /** */
- private Map<Object, Object> jobAttrs;
-
/** */
@Order(6)
boolean isCancelled;
@@ -145,23 +143,11 @@ public class GridJobExecuteResponse implements Message {
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 class GridJobExecuteResponse implements Message {
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 4a26d329f52..b81a8b43bf2 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 @@ public class GridJobProcessor extends
GridProcessorAdapter {
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 1ab5b7f1e4f..167874ceff5 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.GridKernalContext;
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.typedef.internal.U;
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 @@ public class GridJobWorker extends GridWorker implements
GridTimeoutObject {
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 31e7b65bd55..0f262d928f5 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.ClusterGroupEmptyCheckedException;
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.typedef.internal.U;
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 @@ public class GridTaskWorker<T, R> extends GridWorker
implements GridTimeoutObjec
/** */
private final IgniteLogger log;
- /** */
- private final Marshaller marsh;
-
/** */
private final GridTaskSessionImpl ses;
@@ -325,8 +322,6 @@ public class GridTaskWorker<T, R> extends GridWorker
implements GridTimeoutObjec
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 @@ public class GridTaskWorker<T, R> extends GridWorker
implements GridTimeoutObjec
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());
