IGNITE-28520 Auto-generate message serialization / marshalling / deployment

modules/binary/api/src/main/java/org/apache/ignite/internal/processors/cache/CacheObject.java

@@ -120,13 +120,13 @@ public interface CacheObject {
      * @param ldr Class loader.
      * @throws IgniteCheckedException If failed.
      */
-    public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException;
+    public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException;
 
     /**
      * @param ctx Context.
      * @throws IgniteCheckedException If failed.
      */
-    public void prepareMarshal(CacheObjectValueContext ctx) throws IgniteCheckedException;
+    public void marshal(CacheObjectValueContext ctx) throws IgniteCheckedException;
 
     /**
      * @return {@code True} if postWriteHashCode call required, {@code false} otherwise.

modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java

@@ -367,12 +367,12 @@ private <T> T uncachedValue(Class<?> cls) throws BinaryObjectException {
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+    @Override public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         this.ctx = ctx.binaryContext();
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
+    @Override public void marshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
         // No-op.
     }
 

modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java

@@ -220,13 +220,13 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) {
     @Override public CacheObject prepareForCache(CacheObjectValueContext ctx) {
         BinaryObjectImpl res = detached() ? this : detach(false);
 
-        res.prepareMarshal(ctx);
+        res.marshal(ctx);
 
         return res;
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+    @Override public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         assert arr != null || valBytes != null;
 
         if (arr == null)
@@ -238,7 +238,7 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(CacheObjectValueContext ctx) {
+    @Override public void marshal(CacheObjectValueContext ctx) {
         assert arr != null || valBytes != null;
 
         if (valBytes == null)

modules/binary/impl/src/main/java/org/apache/ignite/internal/binary/BinaryObjectOffheapImpl.java

@@ -453,12 +453,12 @@ else if (fieldOffLen == BinaryUtils.OFFSET_2)
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
+    @Override public void unmarshal(CacheObjectValueContext ctx, ClassLoader ldr) throws IgniteCheckedException {
         throw new UnsupportedOperationException();
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
+    @Override public void marshal(CacheObjectValueContext ctx) throws IgniteCheckedException {
         throw new UnsupportedOperationException();
     }
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteMessageFactory.java

@@ -17,12 +17,12 @@
 
 package org.apache.ignite.internal.processors.query.calcite.message;
 
+import org.apache.ignite.internal.managers.communication.IgniteMessageFactory;
 import org.apache.ignite.internal.plugin.AbstractMarshallableMessageFactoryProvider;
 import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
 import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 
 /**
  * Message factory.
@@ -35,19 +35,19 @@ public class CalciteMessageFactory extends AbstractMarshallableMessageFactoryPro
     public static final short MAX_MESSAGE_TYPE = 311;
 
     /** {@inheritDoc} */
-    @Override public void registerAll(MessageFactory factory) {
-        register(factory, QueryStartRequest.class, (short)300, schemaAwareMarsh, resolvedClsLdr);
-        register(factory, QueryStartResponse.class, (short)301, dfltMarsh, dftlClsLdr);
-        register(factory, CalciteErrorMessage.class, (short)302, dfltMarsh, resolvedClsLdr);
-        register(factory, QueryBatchMessage.class, (short)303, dfltMarsh, dftlClsLdr);
-        register(factory, QueryBatchAcknowledgeMessage.class, (short)304, dfltMarsh, dftlClsLdr);
-        register(factory, QueryInboxCloseMessage.class, (short)305, dfltMarsh, dftlClsLdr);
-        register(factory, QueryCloseMessage.class, (short)306, dfltMarsh, dftlClsLdr);
-        register(factory, GenericValueMessage.class, (short)307, schemaAwareMarsh, resolvedClsLdr);
-        register(factory, FragmentMapping.class, (short)308, dfltMarsh, dftlClsLdr);
-        register(factory, ColocationGroup.class, (short)309, dfltMarsh, dftlClsLdr);
-        register(factory, FragmentDescription.class, (short)310, dfltMarsh, dftlClsLdr);
-        register(factory, QueryTxEntry.class, (short)311, dfltMarsh, dftlClsLdr);
+    @Override public void registerAll(IgniteMessageFactory factory) {
+        register(factory, QueryStartRequest.class, (short)300, schemaAwareMarsh);
+        register(factory, QueryStartResponse.class, (short)301, dfltMarsh);
+        register(factory, CalciteErrorMessage.class, (short)302, dfltMarsh);
+        register(factory, QueryBatchMessage.class, (short)303, dfltMarsh);
+        register(factory, QueryBatchAcknowledgeMessage.class, (short)304, dfltMarsh);
+        register(factory, QueryInboxCloseMessage.class, (short)305, dfltMarsh);
+        register(factory, QueryCloseMessage.class, (short)306, dfltMarsh);
+        register(factory, GenericValueMessage.class, (short)307, schemaAwareMarsh);
+        register(factory, FragmentMapping.class, (short)308, dfltMarsh);
+        register(factory, ColocationGroup.class, (short)309, dfltMarsh);
+        register(factory, FragmentDescription.class, (short)310, dfltMarsh);
+        register(factory, QueryTxEntry.class, (short)311, dfltMarsh);
     }
 
     /** */

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/GenericValueMessage.java

@@ -18,10 +18,10 @@
 package org.apache.ignite.internal.processors.query.calcite.message;
 
 import org.apache.ignite.IgniteCheckedException;
-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.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 
 /** */
 public final class GenericValueMessage implements MarshallableMessage {
@@ -48,13 +48,13 @@ public Object value() {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
         if (val != null && serialized == null)
             serialized = U.marshal(marsh, val);
     }
 
     /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         if (serialized != null && val == null)
             val = U.unmarshal(marsh, serialized, clsLdr);
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/MessageServiceImpl.java

@@ -23,20 +23,15 @@
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.failure.FailureContext;
-import org.apache.ignite.failure.FailureType;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.GridTopic;
 import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
 import org.apache.ignite.internal.managers.communication.GridIoManager;
 import org.apache.ignite.internal.managers.communication.GridIoPolicy;
 import org.apache.ignite.internal.managers.communication.GridMessageListener;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.processors.failure.FailureProcessor;
 import org.apache.ignite.internal.processors.query.calcite.CalciteQueryProcessor;
 import org.apache.ignite.internal.processors.query.calcite.exec.QueryTaskExecutor;
 import org.apache.ignite.internal.processors.query.calcite.util.AbstractService;
-import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.plugin.extensions.communication.Message;
 
@@ -47,12 +42,6 @@ public class MessageServiceImpl extends AbstractService implements MessageServic
     /** */
     private final GridMessageListener msgLsnr;
 
-    /** */
-    private final GridCacheSharedContext<?, ?> ctx;
-
-    /** */
-    private final ClassLoader clsLdr;
-
     /** */
     private UUID locNodeId;
 
@@ -62,18 +51,13 @@ public class MessageServiceImpl extends AbstractService implements MessageServic
     /** */
     private QueryTaskExecutor taskExecutor;
 
-    /** */
-    private FailureProcessor failureProc;
-
     /** */
     private Map<Class<? extends Message>, MessageListener> lsnrs;
 
     /** */
     public MessageServiceImpl(GridKernalContext ctx) {
         super(ctx);
 
-        this.ctx = ctx.cache().context();
-        clsLdr = U.resolveClassLoader(ctx.config());
         ioMgr = ctx.io();
         msgLsnr = this::onMessage;
     }
@@ -113,28 +97,13 @@ public QueryTaskExecutor taskExecutor() {
         return taskExecutor;
     }
 
-    /**
-     * @param failureProc Failure processor.
-     */
-    public void failureProcessor(FailureProcessor failureProc) {
-        this.failureProc = failureProc;
-    }
-
-    /**
-     * @return Failure processor.
-     */
-    public FailureProcessor failureProcessor() {
-        return failureProc;
-    }
-
     /** {@inheritDoc} */
     @Override public void onStart(GridKernalContext ctx) {
         localNodeId(ctx.localNodeId());
 
         CalciteQueryProcessor proc = queryProcessor(ctx);
 
         taskExecutor(proc.taskExecutor());
-        failureProcessor(proc.failureProcessor());
 
         init();
     }
@@ -154,11 +123,8 @@ public FailureProcessor failureProcessor() {
     @Override public void send(UUID nodeId, Message msg) throws IgniteCheckedException {
         if (localNodeId().equals(nodeId))
             onMessage(nodeId, msg);
-        else {
-            prepareMarshal(msg);
-
+        else
             ioManager().sendToGridTopic(nodeId, GridTopic.TOPIC_QUERY, msg, GridIoPolicy.CALLER_THREAD);
-        }
     }
 
     /** {@inheritDoc} */
@@ -181,32 +147,6 @@ public FailureProcessor failureProcessor() {
         }
     }
 
-    /** */
-    protected void prepareMarshal(Message msg) throws IgniteCheckedException {
-        try {
-            if (msg instanceof CalciteContextMarshallableMessage)
-                ((CalciteContextMarshallableMessage)msg).prepareMarshal(ctx);
-        }
-        catch (Exception e) {
-            failureProcessor().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
-
-            throw e;
-        }
-    }
-
-    /** */
-    protected void prepareUnmarshal(Message msg) throws IgniteCheckedException {
-        try {
-            if (msg instanceof CalciteContextMarshallableMessage)
-                ((CalciteContextMarshallableMessage)msg).finishUnmarshal(ctx, clsLdr);
-        }
-        catch (Exception e) {
-            failureProcessor().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
-
-            throw e;
-        }
-    }
-
     /** */
     protected void onMessage(UUID nodeId, Message msg) {
         if (msg instanceof ExecutionContextAware) {
@@ -221,22 +161,16 @@ protected void onMessage(UUID nodeId, Message msg) {
             );
     }
 
-    /** */
+    /** Listener for messages arriving from remote nodes; unmarshalled by the generic {@code GridIoManager} payload pass. */
     private void onMessage(UUID nodeId, Object msg, byte plc) {
         if (msg instanceof Message && CalciteMessageFactory.isCalciteMessage((Message)msg))
             onMessage(nodeId, (Message)msg);
     }
 
     /** */
     private void onMessageInternal(UUID nodeId, Message msg) {
-        try {
-            prepareUnmarshal(msg);
+        MessageListener lsnr = Objects.requireNonNull(lsnrs.get(msg.getClass()));
 
-            MessageListener lsnr = Objects.requireNonNull(lsnrs.get(msg.getClass()));
-            lsnr.onMessage(nodeId, msg);
-        }
-        catch (IgniteCheckedException e) {
-            throw U.convertException(e);
-        }
+        lsnr.onMessage(nodeId, msg);
     }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java

@@ -21,19 +21,16 @@
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 import org.jetbrains.annotations.Nullable;
 
-/**
- *
- */
-public class QueryStartRequest implements MarshallableMessage, CalciteContextMarshallableMessage, ExecutionContextAware {
+/** Message sent to remote nodes to start a query fragment execution. */
+public class QueryStartRequest implements MarshallableMessage, ExecutionContextAware {
     /** */
     @Order(0)
     String schema;
@@ -215,32 +212,16 @@ public boolean keepBinaryMode() {
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+    @Override public void marshal(Marshaller marsh) throws IgniteCheckedException {
         if (paramsBytes == null && params != null)
             paramsBytes = U.marshal(marsh, params);
     }
 
     /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        if (qryTxEntries != null) {
-            for (QueryTxEntry e : qryTxEntries)
-                e.prepareMarshal(ctx);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+    @Override public void unmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
         if (params == null && paramsBytes != null)
             params = U.unmarshal(marsh, paramsBytes, clsLdr);
 
         paramsBytes = null;
     }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(GridCacheSharedContext<?, ?> ctx, ClassLoader clsLdr) throws IgniteCheckedException {
-        if (qryTxEntries != null) {
-            for (QueryTxEntry e : qryTxEntries)
-                e.finishUnmarshal(ctx, clsLdr);
-        }
-    }
 }