IGNITE-28270 Investigate possibity to use MessageSerializer for ComputeJobSibling

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,67 +109,67 @@ 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<ComputeJobSibling> siblings;
-
-    /** */
+    /** Sibling jobs ids. Plain representation of {@link GridJobSiblingImpl#jobId} to reduce the messages number. */
     @Order(12)
-    byte[] siblingsBytes;
+    @Nullable List<IgniteUuid> sibJobsIds;
+
+    /** Sibling jobs ids. Plain representation of {@link GridJobSiblingImpl#sesId} to reduce the messages number. */
+    @Order(13)
+    @Nullable List<IgniteUuid> 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<UUID> top;
 
     /** */
     @Marshalled("topPredBytes")
     IgnitePredicate<ClusterNode> 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<UUID> top,
             @Nullable IgnitePredicate<ClusterNode> topPred,
-            Collection<ComputeJobSibling> siblings,
+            Collection<GridJobSiblingImpl> siblings,
             Map<Object, Object> sesAttrs,
             Map<? extends Serializable, ? extends Serializable> 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<IgniteUuid> siblingJobsIds() {
+        return sibJobsIds;
+    }
+
+    /**
+     * @return Sibling job session ids.
      */
-    public Collection<ComputeJobSibling> getSiblings() {
-        return siblings;
+    public @Nullable List<IgniteUuid> siblingJobsSessionIds() {
+        return sibJobsSesId;
     }
 
     /**

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);

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

@@ -166,7 +166,7 @@ public GridTaskSessionImpl(
         @Nullable IgnitePredicate<ClusterNode> topPred,
         long startTime,
         long endTime,
-        Collection<ComputeJobSibling> siblings,
+        @Nullable Collection<ComputeJobSibling> siblings,
         @Nullable Map<Object, Object> attrs,
         GridKernalContext ctx,
         boolean fullSup,

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<IgniteUuid> siblJobsIds = req.siblingJobsIds();
+                        List<IgniteUuid> siblJobsSesIds = req.siblingJobsSessionIds();
+
+                        assert F.isEmpty(siblJobsIds) == F.isEmpty(siblJobsSesIds);
+
+                        Collection<ComputeJobSibling> 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(),

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

@@ -95,7 +95,7 @@ public GridTaskSessionImpl createTaskSession(
         @Nullable IgnitePredicate<ClusterNode> topPred,
         long startTime,
         long endTime,
-        Collection<ComputeJobSibling> siblings,
+        @Nullable Collection<ComputeJobSibling> siblings,
         Map<Object, Object> attrs,
         boolean fullSup,
         boolean internal,

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 <P> Parent type.
+     * @param <C> Child type.
+     * @param p Initial collection.
+     * @return Resulting collection.downcast
+     */
+    private static <P, C extends P> Collection<C> downcast(Collection<P> p) {
+        return (Collection<C>)p;
+    }
+
     /**
      * @param nodeId Node ID.
      */

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