IGNITE-28964 Investigate usage of GridJobSiblingImpl

modules/core/src/main/java/org/apache/ignite/internal/GridJobExecuteRequest.java

@@ -23,7 +23,6 @@
 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;
@@ -109,7 +108,7 @@ public class GridJobExecuteRequest implements ExecutorAwareMessage, DeferredUnma
 
     /** Left unset for a continuous task: such a job requests its siblings from the task node instead. */
     @Marshalled("siblingsBytes")
-    Collection<ComputeJobSibling> siblings;
+    Collection<IgniteUuid> siblingJobsIds;
 
     /** */
     @Order(12)
@@ -188,7 +187,7 @@ public GridJobExecuteRequest() {
      * @param timeout Task execution timeout.
      * @param top Topology.
      * @param topPred Topology predicate.
-     * @param siblings Collection of split siblings.
+     * @param siblingJobsIds Collection sibling jobs ids.
      * @param sesAttrs Session attributes.
      * @param jobAttrs Job attributes.
      * @param cpSpi Collision SPI.
@@ -215,7 +214,7 @@ public GridJobExecuteRequest(
             long timeout,
             @Nullable Collection<UUID> top,
             @Nullable IgnitePredicate<ClusterNode> topPred,
-            Collection<ComputeJobSibling> siblings,
+            @Nullable Collection<IgniteUuid> siblingJobsIds,
             Map<Object, Object> sesAttrs,
             Map<? extends Serializable, ? extends Serializable> jobAttrs,
             String cpSpi,
@@ -253,7 +252,7 @@ public GridJobExecuteRequest(
         this.top = top;
         this.topVer = topVer;
         this.topPred = topPred;
-        this.siblings = dynamicSiblings ? null : siblings;
+        this.siblingJobsIds = dynamicSiblings ? null : siblingJobsIds;
         this.sesAttrs = sesAttrs;
         this.jobAttrs = jobAttrs;
         this.clsLdrId = clsLdrId;
@@ -337,10 +336,10 @@ public long getCreateTime() {
     }
 
     /**
-     * @return Job siblings.
+     * @return Siblings jobs ids.
      */
-    public Collection<ComputeJobSibling> getSiblings() {
-        return siblings;
+    public Collection<IgniteUuid> siblingJobsIds() {
+        return siblingJobsIds;
     }
 
     /**

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;
@@ -40,10 +36,7 @@
 /**
  * This class provides implementation for job sibling.
  */
-public class GridJobSiblingImpl implements ComputeJobSibling, Externalizable {
-    /** */
-    private static final long serialVersionUID = 0L;
-
+public class GridJobSiblingImpl implements ComputeJobSibling {
     /** */
     private IgniteUuid sesId;
 
@@ -173,20 +166,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);

modules/core/src/main/java/org/apache/ignite/internal/GridTaskSessionImpl.java

@@ -25,6 +25,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.cluster.ClusterNode;
@@ -148,7 +149,7 @@ public class GridTaskSessionImpl implements GridTaskSessionInternal {
      * @param topPred Topology predicate.
      * @param startTime Task execution start time.
      * @param endTime Task execution end time.
-     * @param siblings Collection of siblings.
+     * @param siblingJobsIds Collection of sibling jobs ids.
      * @param attrs Session attributes.
      * @param ctx Grid Kernal Context.
      * @param fullSup Session full support enabled flag.
@@ -166,7 +167,7 @@ public GridTaskSessionImpl(
         @Nullable IgnitePredicate<ClusterNode> topPred,
         long startTime,
         long endTime,
-        Collection<ComputeJobSibling> siblings,
+        @Nullable Collection<IgniteUuid> siblingJobsIds,
         @Nullable Map<Object, Object> attrs,
         GridKernalContext ctx,
         boolean fullSup,
@@ -191,7 +192,7 @@ public GridTaskSessionImpl(
         this.sesId = sesId;
         this.startTime = startTime;
         this.endTime = endTime;
-        this.siblings = siblings != null ? unmodifiableCollection(siblings) : null;
+        this.siblings = localSiblingsWrap(siblingJobsIds);
         this.ctx = ctx;
 
         if (attrs != null && !attrs.isEmpty()) {
@@ -209,6 +210,16 @@ public GridTaskSessionImpl(
         this.secCtx = secCtx;
     }
 
+    /**
+     * Creates local representation of {@link ComputeJobSibling}s.
+     *
+     * @see LocalComputeJobSiblingWrap
+     */
+    private static Collection<ComputeJobSibling> localSiblingsWrap(@Nullable Collection<IgniteUuid> siblingJobsIds) {
+        return F.isEmpty(siblingJobsIds) ? Collections.emptyList() : siblingJobsIds.stream().map(LocalComputeJobSiblingWrap::new)
+            .collect(Collectors.toList());
+    }
+
     /** {@inheritDoc} */
     @Override public boolean isFullSupport() {
         return fullSup;
@@ -989,4 +1000,26 @@ public Object login() {
     @Override public String toString() {
         return S.toString(GridTaskSessionImpl.class, this);
     }
+
+    /** A {@link ComputeJobSibling} providing {@link ComputeJobSibling#getJobId()} and unable to {@link ComputeJobSibling#cancel()}. */
+    private static class LocalComputeJobSiblingWrap implements ComputeJobSibling {
+        /** */
+        private final IgniteUuid jobId;
+
+        /** */
+        private LocalComputeJobSiblingWrap(IgniteUuid id) {
+            jobId = id;
+        }
+
+        /** {@inheritDoc} */
+        @Override public IgniteUuid getJobId() {
+            return jobId;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void cancel() throws IgniteException {
+            throw new IgniteException(new UnsupportedOperationException("Cancelation of sibling compute jobs is allowed " +
+                "only where it started."));
+        }
+    }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java

@@ -1266,7 +1266,7 @@ public void processJobExecuteRequest(ClusterNode node, final GridJobExecuteReque
                             req.getTopologyPredicate(),
                             req.startTaskTime(),
                             endTime,
-                            req.getSiblings(),
+                            req.siblingJobsIds(),
                             req.getSessionAttributes(),
                             req.sessionFullSupport(),
                             req.internal(),

modules/core/src/main/java/org/apache/ignite/internal/processors/session/GridTaskSessionProcessor.java

@@ -24,7 +24,6 @@
 import java.util.concurrent.ConcurrentMap;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.compute.ComputeJobSibling;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.GridTaskSessionImpl;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
@@ -77,7 +76,7 @@ public GridTaskSessionProcessor(GridKernalContext ctx) {
      * @param topPred Topology predicate.
      * @param startTime Execution start time.
      * @param endTime Execution end time.
-     * @param siblings Collection of siblings.
+     * @param siblingJobsIds Collection of sibling jons ids.
      * @param attrs Map of attributes.
      * @param fullSup {@code True} to enable distributed session attributes and checkpoints.
      * @param internal {@code True} in case of internal task.
@@ -95,7 +94,7 @@ public GridTaskSessionImpl createTaskSession(
         @Nullable IgnitePredicate<ClusterNode> topPred,
         long startTime,
         long endTime,
-        Collection<ComputeJobSibling> siblings,
+        @Nullable Collection<IgniteUuid> siblingJobsIds,
         Map<Object, Object> attrs,
         boolean fullSup,
         boolean internal,
@@ -113,7 +112,7 @@ public GridTaskSessionImpl createTaskSession(
                 topPred,
                 startTime,
                 endTime,
-                siblings,
+                siblingJobsIds,
                 attrs,
                 ctx,
                 false,
@@ -139,7 +138,7 @@ public GridTaskSessionImpl createTaskSession(
                         topPred,
                         startTime,
                         endTime,
-                        siblings,
+                        siblingJobsIds,
                         attrs,
                         ctx,
                         true,

modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java

@@ -1381,6 +1381,10 @@ private void sendRequest(ComputeJobResult res) {
 
                     boolean forceLocDep = internal || !ctx.deploy().enabled();
 
+                    Collection<IgniteUuid> siblingJobsIds = F.isEmpty(ses.getJobSiblings())
+                        ? null
+                        : ses.getJobSiblings().stream().map(ComputeJobSibling::getJobId).toList();
+
                     req = new GridJobExecuteRequest(
                         ses.getId(),
                         res.getJobContext().getJobId(),
@@ -1392,7 +1396,7 @@ private void sendRequest(ComputeJobResult res) {
                         timeout,
                         ses.getTopology(),
                         ses.getTopologyPredicate(),
-                        ses.getJobSiblings(),
+                        siblingJobsIds,
                         sesAttrs,
                         jobAttrs,
                         ses.getCheckpointSpi(),