Merge branch 'master' into IGNITE-26835

# Conflicts:
#	modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

Author: wernerdv

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java

@@ -854,7 +854,7 @@ private void executeFragment(Query<Row> qry, FragmentPlan plan, ExecutionContext
 
         if (!qry.isExchangeWithInitNodeStarted(ectx.fragmentId())) {
             try {
-                messageService().send(origNodeId, new QueryStartResponse(qry.id(), ectx.fragmentId()));
+                messageService().send(origNodeId, new QueryStartResponse(qry.id(), ectx.fragmentId(), null));
             }
             catch (IgniteCheckedException e) {
                 IgniteException wrpEx = new IgniteException("Failed to send reply. [nodeId=" + origNodeId + ']', e);

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

@@ -18,10 +18,13 @@
 package org.apache.ignite.internal.processors.query.calcite.message;
 
 import java.util.function.Supplier;
+import org.apache.ignite.internal.codegen.ColocationGroupSerializer;
 import org.apache.ignite.internal.codegen.FragmentDescriptionSerializer;
+import org.apache.ignite.internal.codegen.FragmentMappingSerializer;
 import org.apache.ignite.internal.codegen.InboxCloseMessageSerializer;
 import org.apache.ignite.internal.codegen.QueryBatchAcknowledgeMessageSerializer;
 import org.apache.ignite.internal.codegen.QueryCloseMessageSerializer;
+import org.apache.ignite.internal.codegen.QueryStartResponseSerializer;
 import org.apache.ignite.internal.codegen.QueryTxEntrySerializer;
 import org.apache.ignite.internal.processors.query.calcite.metadata.ColocationGroup;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
@@ -36,7 +39,7 @@ public enum MessageType {
     QUERY_START_REQUEST(300, QueryStartRequest::new),
 
     /** */
-    QUERY_START_RESPONSE(301, QueryStartResponse::new),
+    QUERY_START_RESPONSE(301, QueryStartResponse::new, new QueryStartResponseSerializer()),
 
     /** */
     QUERY_ERROR_MESSAGE(302, ErrorMessage::new),
@@ -57,10 +60,10 @@ public enum MessageType {
     GENERIC_VALUE_MESSAGE(307, GenericValueMessage::new),
 
     /** */
-    FRAGMENT_MAPPING(350, FragmentMapping::new),
+    FRAGMENT_MAPPING(350, FragmentMapping::new, new FragmentMappingSerializer()),
 
     /** */
-    COLOCATION_GROUP(351, ColocationGroup::new),
+    COLOCATION_GROUP(351, ColocationGroup::new, new ColocationGroupSerializer()),
 
     /** */
     FRAGMENT_DESCRIPTION(352, FragmentDescription::new, new FragmentDescriptionSerializer()),

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

@@ -17,52 +17,49 @@
 
 package org.apache.ignite.internal.processors.query.calcite.message;
 
-import java.nio.ByteBuffer;
 import java.util.UUID;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.GridDirectTransient;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
+import org.jetbrains.annotations.Nullable;
 
 /**
  *
  */
-public class QueryStartResponse implements MarshalableMessage {
+public class QueryStartResponse implements CalciteMessage {
     /** */
-    private UUID queryId;
+    @Order(value = 0, method = "queryId")
+    private UUID qryId;
 
     /** */
+    @Order(1)
     private long fragmentId;
 
     /** */
-    @GridDirectTransient
-    private Throwable error;
-
-    /** */
-    private byte[] errBytes;
+    @Order(value = 2, method = "errorMessage")
+    private @Nullable ErrorMessage errMsg;
 
     /** */
     public QueryStartResponse() {}
 
     /** */
-    public QueryStartResponse(UUID queryId, long fragmentId) {
-        this(queryId, fragmentId, null);
-    }
-
-    /** */
-    public QueryStartResponse(UUID queryId, long fragmentId, Throwable error) {
-        this.queryId = queryId;
+    public QueryStartResponse(UUID qryId, long fragmentId, @Nullable Throwable error) {
+        this.qryId = qryId;
         this.fragmentId = fragmentId;
-        this.error = error;
+
+        if (error != null)
+            errMsg = new ErrorMessage(error);
     }
 
     /**
      * @return Query ID.
      */
     public UUID queryId() {
-        return queryId;
+        return qryId;
+    }
+
+    /** */
+    public void queryId(UUID qryId) {
+        this.qryId = qryId;
     }
 
     /**
@@ -72,92 +69,26 @@ public long fragmentId() {
         return fragmentId;
     }
 
+    /** */
+    public void fragmentId(long fragmentId) {
+        this.fragmentId = fragmentId;
+    }
+
     /**
      * @return Error.
      */
-    public Throwable error() {
-        return error;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        if (error != null)
-            errBytes = U.marshal(ctx, error);
+    public @Nullable Throwable error() {
+        return ErrorMessage.error(errMsg);
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareUnmarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
-        if (errBytes != null)
-            error = U.unmarshal(ctx, errBytes, U.resolveClassLoader(ctx.gridConfig()));
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
-        writer.setBuffer(buf);
-
-        if (!writer.isHeaderWritten()) {
-            if (!writer.writeHeader(directType()))
-                return false;
-
-            writer.onHeaderWritten();
-        }
-
-        switch (writer.state()) {
-            case 0:
-                if (!writer.writeByteArray(errBytes))
-                    return false;
-
-                writer.incrementState();
-
-            case 1:
-                if (!writer.writeLong(fragmentId))
-                    return false;
-
-                writer.incrementState();
-
-            case 2:
-                if (!writer.writeUuid(queryId))
-                    return false;
-
-                writer.incrementState();
-
-        }
-
-        return true;
+    /** */
+    public @Nullable ErrorMessage errorMessage() {
+        return errMsg;
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
-        reader.setBuffer(buf);
-
-        switch (reader.state()) {
-            case 0:
-                errBytes = reader.readByteArray();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 1:
-                fragmentId = reader.readLong();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 2:
-                queryId = reader.readUuid();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-        }
-
-        return true;
+    /** */
+    public void errorMessage(@Nullable ErrorMessage errMsg) {
+        this.errMsg = errMsg;
     }
 
     /** {@inheritDoc} */