IGNITE-28541 Use MarshallableMessage for the compute messages (#13024)

Author: Vladsz83

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

@@ -23,13 +23,12 @@
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
-import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Job siblings response.
  */
-public class GridJobSiblingsResponse implements Message {
+public class GridJobSiblingsResponse implements MarshallableMessage {
     /** */
     private @Nullable Collection<ComputeJobSibling> siblings;
 
@@ -58,23 +57,14 @@ public GridJobSiblingsResponse(@Nullable Collection<ComputeJobSibling> siblings)
         return siblings;
     }
 
-    /**
-     * Marshals siblings to byte array.
-     *
-     * @param marsh Marshaller.
-     * @throws IgniteCheckedException In case of error.
-     */
-    public void marshalSiblings(Marshaller marsh) throws IgniteCheckedException {
-        siblingsBytes = U.marshal(marsh, siblings);
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (siblings != null)
+            siblingsBytes = U.marshal(marsh, siblings);
     }
 
-    /**
-     * Unmarshals siblings from byte array.
-     *
-     * @param marsh Marshaller.
-     * @throws IgniteCheckedException In case of error.
-     */
-    public void unmarshalSiblings(Marshaller marsh) throws IgniteCheckedException {
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         assert marsh != null;
 
         if (siblingsBytes != null) {
@@ -84,7 +74,6 @@ public void unmarshalSiblings(Marshaller marsh) throws IgniteCheckedException {
         }
     }
 
-
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridJobSiblingsResponse.class, this);

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

@@ -655,17 +655,6 @@ else if (!nodeId.equals(taskNodeId))
                 else {
                     // Sender and message type are fine.
                     res = (GridJobSiblingsResponse)msg;
-
-                    if (res.jobSiblings() == null) {
-                        try {
-                            res.unmarshalSiblings(marsh);
-                        }
-                        catch (IgniteCheckedException e) {
-                            U.error(log, "Failed to unmarshal job siblings.", e);
-
-                            err = e.getMessage();
-                        }
-                    }
                 }
 
                 lock.lock();

modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskCommandHandler.java

@@ -133,7 +133,7 @@ public GridTaskCommandHandler(final GridKernalContext ctx) {
                         if (err != null)
                             res.error(err.getMessage());
                         else
-                            res.marshalResult(ctx, desc.result());
+                            res.res = desc.result();
                     }
                     else
                         res.found(false);
@@ -430,13 +430,6 @@ else if (!nodeId.equals(resHolderId))
                     // Sender and message type are fine.
                     res = (GridTaskResultResponse)msg;
 
-                try {
-                    res.unmarshalResult(ctx);
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Failed to unmarshal task result: " + res, e);
-                }
-
                 lock.lock();
 
                 try {

modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/task/GridTaskResultResponse.java

@@ -18,22 +18,22 @@
 package org.apache.ignite.internal.processors.rest.handlers.task;
 
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.marshaller.Marshaller;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Task result response.
  */
-public class GridTaskResultResponse implements Message {
+public class GridTaskResultResponse implements MarshallableMessage {
     /** Result. */
-    private @Nullable Object res;
+    public @Nullable Object res;
 
     /** Serialized result. */
     @Order(0)
-    byte[] resBytes;
+    @Nullable byte[] resBytes;
 
     /** Finished flag. */
     @Order(1)
@@ -96,24 +96,16 @@ public void error(String err) {
         this.err = err;
     }
 
-    /**
-     * Marshals task result to byte array.
-     *
-     * @param ctx Context.
-     * @param res Task result.
-     */
-    public void marshalResult(GridKernalContext ctx, @Nullable Object res) throws IgniteCheckedException {
-        resBytes = U.marshal(ctx, res);
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (res != null)
+            resBytes = U.marshal(marsh, res);
     }
 
-    /**
-     * Unmarshals task result from byte array.
-     *
-     * @param ctx Context.
-     */
-    public void unmarshalResult(GridKernalContext ctx) throws IgniteCheckedException {
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         if (resBytes != null) {
-            res = U.unmarshal(ctx, resBytes, U.resolveClassLoader(ctx.config()));
+            res = U.unmarshal(marsh, resBytes, clsLdr);
 
             // It is not required anymore.
             resBytes = null;

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

@@ -1418,13 +1418,8 @@ private class JobSiblingsMessageListener implements GridMessageListener {
                 try {
                     Object topic = TOPIC_JOB_SIBLINGS.topic(req.sessionId(), req.topicId());
 
-                    boolean loc = ctx.localNodeId().equals(nodeId);
-
                     GridJobSiblingsResponse resp = new GridJobSiblingsResponse(siblings);
 
-                    if (!loc)
-                        resp.marshalSiblings(marsh);
-
                     ctx.io().sendToCustomTopic(nodeId, topic, resp, SYSTEM_POOL);
                 }
                 catch (IgniteCheckedException e) {