IGNITE-27968 Use message serializer for ExchangeFailureMessage

Author: NSAmelchev

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java

@@ -43,6 +43,8 @@
 import org.apache.ignite.internal.processors.cache.ClientCacheChangeDummyDiscoveryMessageMarshallableSerializer;
 import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
 import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatchMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.ExchangeFailureMessage;
+import org.apache.ignite.internal.processors.cache.ExchangeFailureMessageSerializer;
 import org.apache.ignite.internal.processors.cache.TxTimeoutOnPartitionMapExchangeChangeMessage;
 import org.apache.ignite.internal.processors.cache.TxTimeoutOnPartitionMapExchangeChangeMessageSerializer;
 import org.apache.ignite.internal.processors.cache.WalStateFinishMessage;
@@ -55,6 +57,8 @@
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposedMessageSerializer;
 import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessageSerializer;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeIdSerializer;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResult;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResultSerializer;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResult;
@@ -306,6 +310,7 @@ public DiscoveryMessageFactory(Marshaller marsh, ClassLoader clsLdr) {
         factory.register(38, ChangeCacheEncryptionRequest::new, new ChangeCacheEncryptionRequestSerializer());
 
         factory.register(86, GridCacheVersion::new, new GridCacheVersionSerializer());
+        factory.register(87, GridDhtPartitionExchangeId::new, new GridDhtPartitionExchangeIdSerializer());
 
         factory.register(167, ServiceDeploymentProcessId::new, new ServiceDeploymentProcessIdSerializer());
         factory.register(169, ServiceSingleNodeDeploymentResult::new, new ServiceSingleNodeDeploymentResultSerializer());
@@ -361,5 +366,6 @@ public DiscoveryMessageFactory(Marshaller marsh, ClassLoader clsLdr) {
         factory.register(537, ServiceDeploymentRequest::new,
             new ServiceDeploymentRequestMarshallableSerializer(marsh, clsLdr));
         factory.register(538, ServiceUndeploymentRequest::new, new ServiceUndeploymentRequestSerializer());
+        factory.register(539, ExchangeFailureMessage::new, new ExchangeFailureMessageSerializer());
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeFailureMessage.java

@@ -22,6 +22,8 @@
 import java.util.UUID;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
 import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
@@ -32,32 +34,43 @@
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * This class represents discovery message that is used to provide information about dynamic cache start failure.
  */
-public class ExchangeFailureMessage implements DiscoveryCustomMessage {
+public class ExchangeFailureMessage implements DiscoveryCustomMessage, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** Cache names. */
     @GridToStringInclude
-    private final Collection<String> cacheNames;
+    @Order(0)
+    Collection<String> cacheNames;
 
     /** Custom message ID. */
-    private final IgniteUuid id;
+    @Order(1)
+    IgniteUuid id;
 
     /** */
-    private final GridDhtPartitionExchangeId exchId;
+    @Order(2)
+    GridDhtPartitionExchangeId exchId;
 
     /** */
     @GridToStringInclude
-    private final Map<UUID, Throwable> exchangeErrors;
+    @Order(3)
+    Map<UUID, ErrorMessage> exchangeErrors;
 
     /** Actions to be done to rollback changes done before the exchange failure. */
     private transient ExchangeActions exchangeRollbackActions;
 
+    /** Default constructor for {@link MessageFactory}. */
+    public ExchangeFailureMessage() {
+        // No-op.
+    }
+
     /**
      * Creates new DynamicCacheChangeFailureMessage instance.
      *
@@ -78,7 +91,7 @@ public ExchangeFailureMessage(
         this.id = IgniteUuid.fromUuid(locNode.id());
         this.exchId = exchId;
         this.cacheNames = cacheNames;
-        this.exchangeErrors = exchangeErrors;
+        this.exchangeErrors = F.viewReadOnly(exchangeErrors, ErrorMessage::new);
     }
 
     /** {@inheritDoc} */
@@ -95,7 +108,7 @@ public Collection<String> cacheNames() {
 
     /** */
     public Map<UUID, Throwable> exchangeErrors() {
-        return exchangeErrors;
+        return F.viewReadOnly(exchangeErrors, e -> ErrorMessage.error(e));
     }
 
     /**
@@ -123,7 +136,7 @@ public void exchangeRollbackActions(ExchangeActions exchangeRollbackActions) {
     public IgniteCheckedException createFailureCompoundException() {
         IgniteCheckedException ex = new IgniteCheckedException("Failed to complete exchange process.");
 
-        for (Map.Entry<UUID, Throwable> entry : exchangeErrors.entrySet())
+        for (Map.Entry<UUID, Throwable> entry : exchangeErrors().entrySet())
             U.addSuppressed(ex, entry.getValue());
 
         return ex;