IGNITE-26736 Use MessageSerializer for Calcite's ErrorMessage (#12457)

Author: Vladsz83

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

@@ -32,7 +32,7 @@
 import org.apache.ignite.internal.processors.query.calcite.exec.rel.Outbox;
 import org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpIoTracker;
 import org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpMemoryTracker;
-import org.apache.ignite.internal.processors.query.calcite.message.ErrorMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.CalciteErrorMessage;
 import org.apache.ignite.internal.processors.query.calcite.message.InboxCloseMessage;
 import org.apache.ignite.internal.processors.query.calcite.message.MessageService;
 import org.apache.ignite.internal.processors.query.calcite.message.MessageType;
@@ -171,7 +171,7 @@ public void queryRegistry(QueryRegistry qryRegistry) {
 
     /** {@inheritDoc} */
     @Override public void sendError(UUID nodeId, UUID qryId, long fragmentId, Throwable err) throws IgniteCheckedException {
-        messageService().send(nodeId, new ErrorMessage(qryId, fragmentId, err));
+        messageService().send(nodeId, new CalciteErrorMessage(qryId, fragmentId, err));
     }
 
     /** {@inheritDoc} */

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

@@ -74,7 +74,7 @@
 import org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpMemoryTracker;
 import org.apache.ignite.internal.processors.query.calcite.exec.tracker.PerformanceStatisticsIoTracker;
 import org.apache.ignite.internal.processors.query.calcite.exec.tracker.QueryMemoryTracker;
-import org.apache.ignite.internal.processors.query.calcite.message.ErrorMessage;
+import org.apache.ignite.internal.processors.query.calcite.message.CalciteErrorMessage;
 import org.apache.ignite.internal.processors.query.calcite.message.MessageService;
 import org.apache.ignite.internal.processors.query.calcite.message.MessageType;
 import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
@@ -472,7 +472,7 @@ public void injectService(InjectResourcesService injectSvc) {
     @Override public void init() {
         messageService().register((n, m) -> onMessage(n, (QueryStartRequest)m), MessageType.QUERY_START_REQUEST);
         messageService().register((n, m) -> onMessage(n, (QueryStartResponse)m), MessageType.QUERY_START_RESPONSE);
-        messageService().register((n, m) -> onMessage(n, (ErrorMessage)m), MessageType.QUERY_ERROR_MESSAGE);
+        messageService().register((n, m) -> onMessage(n, (CalciteErrorMessage)m), MessageType.QUERY_ERROR_MESSAGE);
 
         eventManager().addDiscoveryEventListener(discoLsnr, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
 
@@ -945,7 +945,7 @@ private void onMessage(UUID nodeId, QueryStartResponse msg) {
     }
 
     /** */
-    private void onMessage(UUID nodeId, ErrorMessage msg) {
+    private void onMessage(UUID nodeId, CalciteErrorMessage msg) {
         assert nodeId != null && msg != null;
 
         Query<?> qry = qryReg.query(msg.queryId());

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

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query.calcite.message;
+
+import java.util.UUID;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
+
+/**
+ *
+ */
+public class CalciteErrorMessage extends ErrorMessage implements CalciteMessage {
+    /** */
+    @Order(value = 1, method = "queryId")
+    private UUID qryId;
+
+    /** */
+    @Order(2)
+    private long fragmentId;
+
+    /** */
+    public CalciteErrorMessage() {
+        // No-op.
+    }
+
+    /** */
+    public CalciteErrorMessage(UUID qryId, long fragmentId, Throwable err) {
+        super(err);
+
+        assert err != null;
+
+        this.qryId = qryId;
+        this.fragmentId = fragmentId;
+    }
+
+    /**
+     * @return Query ID.
+     */
+    public UUID queryId() {
+        return qryId;
+    }
+
+    /** */
+    public void queryId(UUID qryId) {
+        this.qryId = qryId;
+    }
+
+    /**
+     * @return Fragment ID.
+     */
+    public long fragmentId() {
+        return fragmentId;
+    }
+
+    /** */
+    public void fragmentId(long fragmentId) {
+        this.fragmentId = fragmentId;
+    }
+
+    /** {@inheritDoc} */
+    @Override public MessageType type() {
+        return MessageType.QUERY_ERROR_MESSAGE;
+    }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return MessageType.QUERY_ERROR_MESSAGE.directType();
+    }
+}

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

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.query.calcite.message;
 
 import java.util.function.Supplier;
+import org.apache.ignite.internal.codegen.CalciteErrorMessageSerializer;
 import org.apache.ignite.internal.codegen.ColocationGroupSerializer;
 import org.apache.ignite.internal.codegen.FragmentDescriptionSerializer;
 import org.apache.ignite.internal.codegen.FragmentMappingSerializer;
@@ -42,7 +43,7 @@ public enum MessageType {
     QUERY_START_RESPONSE(301, QueryStartResponse::new, new QueryStartResponseSerializer()),
 
     /** */
-    QUERY_ERROR_MESSAGE(302, ErrorMessage::new),
+    QUERY_ERROR_MESSAGE(302, CalciteErrorMessage::new, new CalciteErrorMessageSerializer()),
 
     /** */
     QUERY_BATCH_MESSAGE(303, QueryBatchMessage::new),

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

@@ -22,6 +22,7 @@
 import org.apache.ignite.internal.MessageProcessor;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+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.marshaller.jdk.JdkMarshaller;
@@ -101,7 +102,7 @@ public ErrorMessage(@Nullable Throwable err) {
      * @see MessageWriter
      */
     public void errorBytes(@Nullable byte[] errBytes) {
-        if (errBytes == null)
+        if (F.isEmpty(errBytes))
             err = null;
         else {
             try {