diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java index 098e8514f681c..f4953a0ebf471 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java @@ -18,16 +18,18 @@ package org.apache.ignite.internal; import java.io.Serializable; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.Map; import java.util.UUID; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.compute.ComputeJob; -import org.apache.ignite.compute.ComputeJobSibling; import org.apache.ignite.configuration.DeploymentMode; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgnitePredicate; @@ -107,43 +109,43 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage, DeferredUnma @Order(11) String cpSpi; - /** Left unset for a continuous task: such a job requests its siblings from the task node instead. */ - @Marshalled("siblingsBytes") - Collection siblings; - - /** */ + /** Sibling jobs ids. Plain representation of {@link GridJobSiblingImpl#jobId} to reduce the messages number. */ @Order(12) - byte[] siblingsBytes; + @Nullable List sibJobsIds; + + /** Sibling jobs ids. Plain representation of {@link GridJobSiblingImpl#sesId} to reduce the messages number. */ + @Order(13) + @Nullable List sibJobsSesId; /** Transient since needs to hold local creation time. */ private final long createTime = U.currentTimeMillis(); /** */ - @Order(13) + @Order(14) IgniteUuid clsLdrId; /** */ - @Order(14) + @Order(15) DeploymentMode depMode; /** */ - @Order(15) + @Order(16) boolean dynamicSiblings; /** */ - @Order(16) + @Order(17) boolean forceLocDep; /** */ - @Order(17) + @Order(18) boolean sesFullSup; /** */ - @Order(18) + @Order(19) boolean internal; /** */ - @Order(19) + @Order(20) Collection top; /** */ @@ -151,23 +153,23 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage, DeferredUnma IgnitePredicate topPred; /** */ - @Order(20) + @Order(21) byte[] topPredBytes; /** */ - @Order(21) + @Order(22) int[] cacheIds; /** */ - @Order(22) + @Order(23) int part; /** */ - @Order(23) + @Order(24) AffinityTopologyVersion topVer; /** */ - @Order(24) + @Order(25) String execName; /** @@ -215,7 +217,7 @@ public GridJobExecuteRequest( long timeout, @Nullable Collection top, @Nullable IgnitePredicate topPred, - Collection siblings, + Collection siblings, Map sesAttrs, Map jobAttrs, String cpSpi, @@ -253,7 +255,6 @@ public GridJobExecuteRequest( this.top = top; this.topVer = topVer; this.topPred = topPred; - this.siblings = dynamicSiblings ? null : siblings; this.sesAttrs = sesAttrs; this.jobAttrs = jobAttrs; this.clsLdrId = clsLdrId; @@ -269,6 +270,17 @@ public GridJobExecuteRequest( this.execName = execName; this.cpSpi = cpSpi == null || cpSpi.isEmpty() ? null : cpSpi; + + // TODO : Revise after https://issues.apache.org/jira/browse/IGNITE-28964 + if (!dynamicSiblings && !F.isEmpty(siblings)) { + sibJobsIds = new ArrayList<>(siblings.size()); + sibJobsSesId = new ArrayList<>(sibJobsIds.size()); + + siblings.forEach(sibJobImpl -> { + sibJobsIds.add(sibJobImpl.jobId); + sibJobsSesId.add(sibJobImpl.sesId); + }); + } } /** @@ -337,10 +349,17 @@ public long getCreateTime() { } /** - * @return Job siblings. + * @return Sibling job ids. + */ + public @Nullable List siblingJobsIds() { + return sibJobsIds; + } + + /** + * @return Sibling job session ids. */ - public Collection getSiblings() { - return siblings; + public @Nullable List siblingJobsSessionIds() { + return sibJobsSesId; } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridJobSiblingImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridJobSiblingImpl.java index 5004aa946dfa2..3f4cbb73ca1f1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridJobSiblingImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridJobSiblingImpl.java @@ -17,10 +17,6 @@ package org.apache.ignite.internal; -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; import java.util.Collection; import java.util.UUID; import org.apache.ignite.IgniteCheckedException; @@ -39,17 +35,14 @@ /** * This class provides implementation for job sibling. + * TODO : Revise after https://issues.apache.org/jira/browse/IGNITE-28964 */ -public class GridJobSiblingImpl implements ComputeJobSibling, Externalizable { +public class GridJobSiblingImpl implements ComputeJobSibling { /** */ - private static final long serialVersionUID = 0L; + IgniteUuid sesId; /** */ - private IgniteUuid sesId; - - /** */ - @SuppressWarnings({"FieldAccessedSynchronizedAndUnsynchronized"}) - private IgniteUuid jobId; + final IgniteUuid jobId; /** */ private Object taskTopic; @@ -64,12 +57,7 @@ public class GridJobSiblingImpl implements ComputeJobSibling, Externalizable { private boolean isJobDone; /** */ - private transient GridKernalContext ctx; - - /** */ - public GridJobSiblingImpl() { - // No-op. - } + private GridKernalContext ctx; /** * @param sesId Task session ID. @@ -173,20 +161,6 @@ public synchronized Object jobTopic() { ctx.job().cancelJob(sesId, jobId, false); } - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - // Don't serialize node ID. - U.writeIgniteUuid(out, sesId); - U.writeIgniteUuid(out, jobId); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - // Don't serialize node ID. - sesId = U.readIgniteUuid(in); - jobId = U.readIgniteUuid(in); - } - /** {@inheritDoc} */ @Override public String toString() { return S.toString(GridJobSiblingImpl.class, this); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java index 27d07e67b10e4..5ccf55e209d50 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java @@ -166,7 +166,7 @@ public GridTaskSessionImpl( @Nullable IgnitePredicate topPred, long startTime, long endTime, - Collection siblings, + @Nullable Collection siblings, @Nullable Map attrs, GridKernalContext ctx, boolean fullSup, 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 b81a8b43bf21e..b2fbacd19c290 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 @@ -18,9 +18,11 @@ package org.apache.ignite.internal.processors.job; import java.util.AbstractCollection; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.NoSuchElementException; import java.util.Objects; @@ -54,6 +56,7 @@ import org.apache.ignite.internal.GridJobExecuteRequest; import org.apache.ignite.internal.GridJobExecuteResponse; import org.apache.ignite.internal.GridJobSessionImpl; +import org.apache.ignite.internal.GridJobSiblingImpl; import org.apache.ignite.internal.GridJobSiblingsRequest; import org.apache.ignite.internal.GridJobSiblingsResponse; import org.apache.ignite.internal.GridKernalContext; @@ -1256,6 +1259,23 @@ public void processJobExecuteRequest(ClusterNode node, final GridJobExecuteReque U.resolveClassLoader(dep.classLoader(), ctx.config())); } + // TODO : Revise after https://issues.apache.org/jira/browse/IGNITE-28964 + List siblJobsIds = req.siblingJobsIds(); + List siblJobsSesIds = req.siblingJobsSessionIds(); + + assert F.isEmpty(siblJobsIds) == F.isEmpty(siblJobsSesIds); + + Collection siblings = null; + + if (!F.isEmpty(siblJobsIds)) { + assert siblJobsSesIds.size() == siblJobsIds.size(); + + siblings = new ArrayList<>(siblJobsIds.size()); + + for (int i = 0; i < siblJobsIds.size(); ++i) + siblings.add(new GridJobSiblingImpl(siblJobsSesIds.get(i), siblJobsIds.get(i), node.id(), ctx)); + } + GridTaskSessionImpl taskSes = ctx.session().createTaskSession( req.sessionId(), node.id(), @@ -1266,7 +1286,7 @@ public void processJobExecuteRequest(ClusterNode node, final GridJobExecuteReque req.getTopologyPredicate(), req.startTaskTime(), endTime, - req.getSiblings(), + siblings, req.getSessionAttributes(), req.sessionFullSupport(), req.internal(), diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java index f0877df777b1f..955865226e20c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java @@ -95,7 +95,7 @@ public GridTaskSessionImpl createTaskSession( @Nullable IgnitePredicate topPred, long startTime, long endTime, - Collection siblings, + @Nullable Collection siblings, Map attrs, boolean fullSup, boolean internal, 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 0f262d928f5a9..369023bbe7620 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 @@ -1392,7 +1392,7 @@ private void sendRequest(ComputeJobResult res) { timeout, ses.getTopology(), ses.getTopologyPredicate(), - ses.getJobSiblings(), + downcast(ses.getJobSiblings()), sesAttrs, jobAttrs, ses.getCheckpointSpi(), @@ -1473,6 +1473,19 @@ else if (log.isDebugEnabled()) } } + /** + * TODO : Revise after https://issues.apache.org/jira/browse/IGNITE-28964 + * Downcasts collection type. + * + * @param

Parent type. + * @param Child type. + * @param p Initial collection. + * @return Resulting collection.downcast + */ + private static Collection downcast(Collection

p) { + return (Collection)p; + } + /** * @param nodeId Node ID. */ diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties index fc1966fef4b4e..d1282ad57f523 100644 --- a/modules/core/src/main/resources/META-INF/classnames.properties +++ b/modules/core/src/main/resources/META-INF/classnames.properties @@ -220,7 +220,6 @@ org.apache.ignite.internal.GridJobCancelRequest org.apache.ignite.internal.GridJobContextImpl org.apache.ignite.internal.GridJobExecuteRequest org.apache.ignite.internal.GridJobExecuteResponse -org.apache.ignite.internal.GridJobSiblingImpl org.apache.ignite.internal.GridJobSiblingsRequest org.apache.ignite.internal.GridJobSiblingsResponse org.apache.ignite.internal.GridKernalContextImpl