IGNITE-28046 Use MessageSerializer for MetadataUpdateProposedMessage

Author: NSAmelchev

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

@@ -57,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.binary.MetadataUpdateProposedMessage;
+import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessageMarshallableSerializer;
 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;
@@ -379,5 +381,6 @@ public DiscoveryMessageFactory(Marshaller marsh, ClassLoader clsLdr) {
         factory.register(540, StartRoutineDiscoveryMessage::new, new StartRoutineDiscoveryMessageSerializer());
         factory.register(541, StartRoutineAckDiscoveryMessage::new, new StartRoutineAckDiscoveryMessageSerializer());
         factory.register(542, StartRoutineDiscoveryMessageV2::new, new StartRoutineDiscoveryMessageV2Serializer());
+        factory.register(543, MetadataUpdateProposedMessage::new, new MetadataUpdateProposedMessageMarshallableSerializer(marsh, clsLdr));
     }
 }

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

@@ -550,7 +550,7 @@ private final class MetadataUpdateProposedListener implements CustomEventListene
                 //coordinator receives update request
                 if (metaVerInfo != null) {
                     if (metaVerInfo.removing()) {
-                        msg.markRejected(new BinaryObjectException("The type is removing now [typeId=" + typeId + ']'));
+                        msg.markRejected("The type is removing now [typeId=" + typeId + ']');
 
                         pendingVer = REMOVED_VERSION;
                         acceptedVer = REMOVED_VERSION;
@@ -589,7 +589,7 @@ private final class MetadataUpdateProposedListener implements CustomEventListene
                     catch (BinaryObjectException err) {
                         log.warning("Exception with merging metadata for typeId: " + typeId, err);
 
-                        msg.markRejected(err);
+                        msg.markRejected(err.getMessage());
                     }
                 }
             }
@@ -602,7 +602,7 @@ private final class MetadataUpdateProposedListener implements CustomEventListene
                 MetadataUpdateResultFuture fut = unlabeledFutures.poll();
 
                 if (msg.rejected())
-                    fut.onDone(MetadataUpdateResult.createFailureResult(msg.rejectionError()));
+                    fut.onDone(MetadataUpdateResult.createFailureResult(msg.rejectionErrorMessage()));
                 else {
                     if (clientNode) {
                         boolean success = casBinaryMetadata(typeId, new BinaryMetadataVersionInfo(msg.metadata(), pendingVer, acceptedVer));

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

@@ -17,21 +17,25 @@
 package org.apache.ignite.internal.processors.cache.binary;
 
 import java.util.UUID;
-import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.binary.BinaryMetadata;
 import org.apache.ignite.internal.binary.BinaryMetadataHandler;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 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.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * <b>MetadataUpdateProposedMessage</b> and {@link MetadataUpdateAcceptedMessage} messages make a basis for
  * discovery-based protocol for exchanging {@link BinaryMetadata metadata} describing objects in binary format stored in Ignite caches.
- *
+ * <p>
  * All interactions with binary metadata are performed through {@link BinaryMetadataHandler}
  * interface implemented in {@link CacheObjectBinaryProcessorImpl} processor.
- *
+ * <p>
  * Protocol works as follows:
  * <ol>
  * <li>
@@ -67,33 +71,49 @@
  * it gets blocked until {@link MetadataUpdateAcceptedMessage} arrives with <b>accepted version</b>
  * equals to <b>pending version</b> of this metadata to the moment when is was initially read by the thread.
  */
-public final class MetadataUpdateProposedMessage implements DiscoveryCustomMessage {
+public final class MetadataUpdateProposedMessage implements DiscoveryCustomMessage, MarshallableMessage {
     /** */
     private static final long serialVersionUID = 0L;
 
     /** */
-    private final IgniteUuid id = IgniteUuid.randomUuid();
+    @Order(0)
+    IgniteUuid id;
 
     /** Node UUID which initiated metadata update. */
-    private final UUID origNodeId;
+    @Order(1)
+    UUID origNodeId;
 
     /** */
     private BinaryMetadata metadata;
 
+    /** Serialized {@link #metadata}. */
+    @Order(2)
+    byte[] metadataBytes;
+
     /** Metadata type id. */
-    private final int typeId;
+    @Order(3)
+    int typeId;
 
     /** Metadata version which is pending for update. */
-    private int pendingVer;
+    @Order(4)
+    int pendingVer;
 
     /** Metadata version which is already accepted by entire cluster. */
-    private int acceptedVer;
+    @Order(5)
+    int acceptedVer;
 
     /** Message acceptance status. */
-    private ProposalStatus status = ProposalStatus.SUCCESSFUL;
+    @Order(6)
+    boolean rejected;
 
     /** */
-    private BinaryObjectException err;
+    @Order(7)
+    String errMsg;
+
+    /** Constructor. */
+    public MetadataUpdateProposedMessage() {
+        // No-op.
+    }
 
     /**
      * @param metadata   {@link BinaryMetadata} requested to be updated.
@@ -103,6 +123,7 @@ public MetadataUpdateProposedMessage(BinaryMetadata metadata, UUID origNodeId) {
         assert origNodeId != null;
         assert metadata != null;
 
+        id = IgniteUuid.randomUuid();
         this.origNodeId = origNodeId;
 
         this.metadata = metadata;
@@ -120,7 +141,7 @@ public MetadataUpdateProposedMessage(BinaryMetadata metadata, UUID origNodeId) {
      * {@inheritDoc}
      */
     @Nullable @Override public DiscoveryCustomMessage ackMessage() {
-        return (status == ProposalStatus.SUCCESSFUL) ? new MetadataUpdateAcceptedMessage(typeId, pendingVer) : null;
+        return !rejected ? new MetadataUpdateAcceptedMessage(typeId, pendingVer) : null;
     }
 
     /**
@@ -131,25 +152,25 @@ public MetadataUpdateProposedMessage(BinaryMetadata metadata, UUID origNodeId) {
     }
 
     /**
-     * @param err Error caused this update to be rejected.
+     * @param errMsg Error message caused this update to be rejected.
      */
-    void markRejected(BinaryObjectException err) {
-        status = ProposalStatus.REJECTED;
-        this.err = err;
+    void markRejected(String errMsg) {
+        rejected = true;
+        this.errMsg = errMsg;
     }
 
     /**
      *
      */
     boolean rejected() {
-        return status == ProposalStatus.REJECTED;
+        return rejected;
     }
 
     /**
      *
      */
-    BinaryObjectException rejectionError() {
-        return err;
+    String rejectionErrorMessage() {
+        return errMsg;
     }
 
     /**
@@ -208,13 +229,16 @@ public int typeId() {
         return typeId;
     }
 
-    /** Message acceptance status. */
-    private enum ProposalStatus {
-        /** */
-        SUCCESSFUL,
+    /** {@inheritDoc} */
+    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (metadata != null)
+            metadataBytes = U.marshal(marsh, metadata);
+    }
 
-        /** */
-        REJECTED
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader ldr) throws IgniteCheckedException {
+        if (metadataBytes != null)
+            metadata = U.unmarshal(marsh, metadataBytes, ldr);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/resources/META-INF/classnames.properties

@@ -1030,7 +1030,6 @@ org.apache.ignite.internal.processors.cache.binary.MetadataRequestMessage
 org.apache.ignite.internal.processors.cache.binary.MetadataResponseMessage
 org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessage
 org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessage
-org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessage$ProposalStatus
 org.apache.ignite.internal.processors.cache.binary.MetadataUpdateResult$ResultType
 org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager$BlockSetCallable
 org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager$QueueHeaderPredicate