Skip to content

IGNITE-28271 Use the generated marshalling for GridJobExecuteResponse #13435

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
anton-vinogradov merged 1 commit into from
Aug 5, 2026
Merged

IGNITE-28271 Use the generated marshalling for GridJobExecuteResponse #13435

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
140 changes: 18 additions & 122 deletions modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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.
*/
Expand Down Expand Up @@ -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);
Expand Down
18 changes: 16 additions & 2 deletions modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
39 changes: 31 additions & 8 deletions modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -122,9 +122,6 @@ public class GridJobWorker extends GridWorker implements GridTimeoutObject {
/** */
private final IgniteLogger log;

/** */
private final Marshaller marsh;

/** */
private final GridJobSessionImpl ses;

Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();

Expand Down
9 changes: 2 additions & 7 deletions modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -156,9 +156,6 @@ private enum State {
/** */
private final IgniteLogger log;

/** */
private final Marshaller marsh;

/** */
private final GridTaskSessionImpl ses;

Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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());

Expand Down
Loading