fix performance degradation for BinaryMetadataUpdatesFlowTest#testConcurrentMetadataUpdates

Author: wernerdv

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

@@ -56,7 +56,7 @@
 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.binary.MetadataUpdateProposedMessageSerializer;
 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;
@@ -363,8 +363,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(534, DynamicCacheChangeBatch::new,
-            new DynamicCacheChangeBatchMarshallableSerializer(marsh, clsLdr));
-        factory.register(535, MetadataUpdateProposedMessage::new, new MetadataUpdateProposedMessageMarshallableSerializer(marsh, clsLdr));
+        factory.register(539, MetadataUpdateProposedMessage::new, new MetadataUpdateProposedMessageSerializer());
     }
 }

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

@@ -26,7 +26,7 @@
 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.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -70,8 +70,11 @@
  * (with <b>pending version</b> strictly greater than <b>accepted version</b>)
  * 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.
+ * <p>
+ * We don't implement MarshallableMessage for this message because it leads to performance degradation when updating BinaryMetadata
+ * (see test: BinaryMetadataUpdatesFlowTest#testConcurrentMetadataUpdates).
  */
-public final class MetadataUpdateProposedMessage implements DiscoveryCustomMessage, MarshallableMessage {
+public final class MetadataUpdateProposedMessage implements DiscoveryCustomMessage, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -222,26 +225,28 @@ public void metadata(BinaryMetadata metadata) {
         this.metadata = metadata;
     }
 
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
-        if (metadata != null)
-            metadataBytes = U.marshal(marsh, metadata);
+    /**
+     *
+     */
+    public int typeId() {
+        return typeId;
     }
 
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader ldr) throws IgniteCheckedException {
-        if (metadataBytes != null) {
-            metadata = U.unmarshal(marsh, metadataBytes, ldr);
-
-            metadataBytes = null;
-        }
+    /**
+     * @param marsh Marshaller.
+     */
+    public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+        if (metadata != null)
+            metadataBytes = U.marshal(marsh, metadata);
     }
 
     /**
-     *
+     * @param marsh Marshaller.
+     * @param ldr Class loader.
      */
-    public int typeId() {
-        return typeId;
+    public void finishUnmarshal(Marshaller marsh, ClassLoader ldr) throws IgniteCheckedException {
+        if (metadataBytes != null)
+            metadata = U.unmarshal(marsh, metadataBytes, ldr);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java

@@ -23,6 +23,8 @@
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.managers.discovery.IncompleteDeserializationException;
+import org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
+import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateProposedMessage;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.Marshaller;
@@ -104,8 +106,15 @@ public DiscoverySpiCustomMessage message() {
     @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
         super.prepareMarshal(marsh);
 
-        if (msg instanceof Message)
+        if (msg instanceof Message) {
+            if (msg instanceof MetadataUpdateProposedMessage)
+                ((MetadataUpdateProposedMessage)msg).prepareMarshal(marsh);
+            else if (msg instanceof SecurityAwareCustomMessageWrapper &&
+                ((SecurityAwareCustomMessageWrapper)msg).delegate() instanceof MetadataUpdateProposedMessage)
+                ((MetadataUpdateProposedMessage)((SecurityAwareCustomMessageWrapper)msg).delegate()).prepareMarshal(marsh);
+
             serMsg = (Message)msg;
+        }
         else {
             if (msg != null)
                 msgBytes = U.marshal(marsh, msg);
@@ -125,8 +134,15 @@ public DiscoverySpiCustomMessage message() {
         if (msg != null)
             return;
 
-        if (serMsg != null)
+        if (serMsg != null) {
             msg = (DiscoverySpiCustomMessage)serMsg;
+
+            if (msg instanceof MetadataUpdateProposedMessage)
+                ((MetadataUpdateProposedMessage)msg).finishUnmarshal(marsh, ldr);
+            else if (msg instanceof SecurityAwareCustomMessageWrapper &&
+                ((SecurityAwareCustomMessageWrapper)msg).delegate() instanceof MetadataUpdateProposedMessage)
+                ((MetadataUpdateProposedMessage)((SecurityAwareCustomMessageWrapper)msg).delegate()).finishUnmarshal(marsh, ldr);
+        }
         else {
             try {
                 if (msgBytes != null)

modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/IgniteDiscoveryMessageSerializationTest.java

@@ -21,7 +21,6 @@
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 
-import static org.apache.ignite.internal.util.IgniteUtils.toBytes;
 import static org.apache.ignite.marshaller.Marshallers.jdk;
 
 /** Serialization test for discovery messages. */
@@ -30,27 +29,4 @@ public class IgniteDiscoveryMessageSerializationTest extends AbstractMessageSeri
     @Override protected MessageFactoryProvider messageFactory() {
         return new DiscoveryMessageFactory(jdk(), U.gridClassLoader());
     }
-
-    /** {@inheritDoc} */
-    @Override protected AbstractTestMessageReader createMessageReader(int capacity) {
-        return new TestIoMessageReader(capacity);
-    }
-
-    /** */
-    private static class TestIoMessageReader extends AbstractTestMessageReader {
-        /** */
-        private static final byte[] BYTE_ARR = toBytes(null);
-
-        /** */
-        public TestIoMessageReader(int capacity) {
-            super(capacity);
-        }
-
-        /** {@inheritDoc} */
-        @Override public byte[] readByteArray() {
-            super.readByteArray();
-
-            return BYTE_ARR;
-        }
-    }
 }