IGNITE-28594 Use Message DTO for DiscoveryDataBag#JoiningNodeDiscoveryData

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -38,7 +38,9 @@
 import org.apache.ignite.internal.managers.encryption.ChangeCacheEncryptionRequest;
 import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest;
 import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyResponse;
+import org.apache.ignite.internal.managers.encryption.GroupKeyEncrypted;
 import org.apache.ignite.internal.managers.encryption.MasterKeyChangeRequest;
+import org.apache.ignite.internal.managers.encryption.NodeEncryptionKeys;
 import org.apache.ignite.internal.managers.eventstorage.GridEventStorageMessage;
 import org.apache.ignite.internal.plugin.AbstractMarshallableMessageFactoryProvider;
 import org.apache.ignite.internal.processors.authentication.User;
@@ -68,6 +70,7 @@
 import org.apache.ignite.internal.processors.cache.WalStateFinishMessage;
 import org.apache.ignite.internal.processors.cache.WalStateProposeMessage;
 import org.apache.ignite.internal.processors.cache.binary.BinaryMetadataVersionInfo;
+import org.apache.ignite.internal.processors.cache.binary.BinaryMetadataVersionsData;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveAcceptedMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposedMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRequestMessage;
@@ -186,15 +189,18 @@
 import org.apache.ignite.internal.processors.datastreamer.DataStreamerEntry;
 import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
 import org.apache.ignite.internal.processors.datastreamer.DataStreamerResponse;
+import org.apache.ignite.internal.processors.marshaller.MappedName;
 import org.apache.ignite.internal.processors.marshaller.MappingAcceptedMessage;
 import org.apache.ignite.internal.processors.marshaller.MappingProposedMessage;
 import org.apache.ignite.internal.processors.marshaller.MarshallerMappingItem;
+import org.apache.ignite.internal.processors.marshaller.MarshallerMappingsData;
 import org.apache.ignite.internal.processors.marshaller.MissingMappingRequestMessage;
 import org.apache.ignite.internal.processors.marshaller.MissingMappingResponseMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageCasAckMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageCasMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateAckMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateMessage;
+import org.apache.ignite.internal.processors.query.InlineSizesData;
 import org.apache.ignite.internal.processors.query.QueryField;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryCancelRequest;
 import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryFailResponse;
@@ -243,6 +249,7 @@
 import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
 import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
 import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
+import org.apache.ignite.spi.discovery.ObjectData;
 import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket;
 import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
 import org.apache.ignite.spi.discovery.tcp.messages.InetAddressMessage;
@@ -343,6 +350,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(CacheVersionedValue.class);
         withNoSchema(GridCacheVersion.class);
         withNoSchema(GridCacheVersionEx.class);
+        withNoSchemaResolvedClassLoader(ObjectData.class);
 
         // [5700 - 5900]: Discovery originated messages.
         msgIdx = 5700;
@@ -560,6 +568,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(StatisticsResponse.class);
         withNoSchema(CacheContinuousQueryBatchAck.class);
         withSchema(CacheContinuousQueryEntry.class);
+        withNoSchema(InlineSizesData.class);
 
         // [11200 - 11300]: Compute, distributed process messages.
         msgIdx = 11200;
@@ -624,14 +633,19 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(MetadataRequestMessage.class);
         withNoSchema(MetadataResponseMessage.class);
         withNoSchema(MarshallerMappingItem.class);
-        withNoSchema(BinaryMetadataVersionInfo.class);
+        withSchemaResolvedClassLoader(BinaryMetadataVersionInfo.class);
+        withNoSchema(BinaryMetadataVersionsData.class);
+        withNoSchema(MappedName.class);
+        withNoSchema(MarshallerMappingsData.class);
 
         // [12400 - 12500]: Encryption messages.
         msgIdx = 12400;
         withNoSchema(GenerateEncryptionKeyRequest.class);
         withNoSchema(GenerateEncryptionKeyResponse.class);
         withNoSchema(ChangeCacheEncryptionRequest.class);
         withNoSchema(MasterKeyChangeRequest.class);
+        withNoSchema(NodeEncryptionKeys.class);
+        withNoSchema(GroupKeyEncrypted.class);
 
         // [13000 - 13300]: Control, configuration, diagnostincs and other messages.
         msgIdx = 13000;
@@ -662,6 +676,11 @@ private <T extends Message> void withNoSchemaResolvedClassLoader(Class<T> cls) {
         register(cls, dfltMarsh, resolvedClsLdr);
     }
 
+    /** Registers message using {@link #schemaAwareMarsh} and {@link #resolvedClsLdr}. */
+    private <T extends Message> void withSchemaResolvedClassLoader(Class<T> cls) {
+        register(cls, schemaAwareMarsh, resolvedClsLdr);
+    }
+
     /** Registers message using incrementing {@link #msgIdx} as the message id/type. */
     private <T extends Message> void register(Class<T> cls, Marshaller marsh, ClassLoader clsLrd) {
         register(factory, cls, msgIdx++, marsh, clsLrd);

modules/core/src/main/java/org/apache/ignite/internal/GridPluginComponent.java

@@ -28,6 +28,7 @@
 import org.apache.ignite.spi.discovery.DiscoveryDataBag;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag.JoiningNodeDiscoveryData;
+import org.apache.ignite.spi.discovery.ObjectData;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -115,7 +116,7 @@ public PluginProvider plugin() {
     @Nullable @Override public IgniteNodeValidationResult validateNode(ClusterNode node,
         JoiningNodeDiscoveryData discoData) {
         try {
-            Map<String, Serializable> map = (Map<String, Serializable>)discoData.joiningNodeData();
+            Map<String, Serializable> map = ObjectData.unwrap(discoData.joiningNodeData());
 
             if (map != null)
                 plugin.validateNewNode(node, map.get(plugin.name()));

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java

@@ -431,9 +431,9 @@ public void onLocalJoin() {
                 "Cache group key change is in progress! Node join is rejected.");
         }
 
-        NodeEncryptionKeys nodeEncKeys = (NodeEncryptionKeys)discoData.joiningNodeData();
+        NodeEncryptionKeys nodeEncKeys = discoData.joiningNodeData();
 
-        if (!discoData.hasJoiningNodeData() || nodeEncKeys == null) {
+        if (nodeEncKeys == null) {
             return new IgniteNodeValidationResult(ctx.localNodeId(),
                 "Joining node doesn't have encryption data [node=" + node.id() + "]",
                 "Joining node doesn't have encryption data.");
@@ -522,7 +522,7 @@ public void onLocalJoin() {
 
     /** {@inheritDoc} */
     @Override public void onJoiningNodeDataReceived(JoiningNodeDiscoveryData data) {
-        NodeEncryptionKeys nodeEncryptionKeys = (NodeEncryptionKeys)data.joiningNodeData();
+        NodeEncryptionKeys nodeEncryptionKeys = data.joiningNodeData();
 
         if (nodeEncryptionKeys == null || nodeEncryptionKeys.newKeys == null || ctx.clientNode())
             return;
@@ -1748,45 +1748,6 @@ private String decryptKeyName(byte[] data) {
         });
     }
 
-    /** */
-    protected static class NodeEncryptionKeys implements Serializable {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        NodeEncryptionKeys(
-            HashMap<Integer, List<GroupKeyEncrypted>> knownKeysWithIds,
-            Map<Integer, byte[]> newKeys,
-            byte[] masterKeyDigest
-        ) {
-            this.newKeys = newKeys;
-            this.masterKeyDigest = masterKeyDigest;
-
-            if (F.isEmpty(knownKeysWithIds))
-                return;
-
-            // To be able to join the old cluster.
-            knownKeys = U.newHashMap(knownKeysWithIds.size());
-
-            for (Map.Entry<Integer, List<GroupKeyEncrypted>> entry : knownKeysWithIds.entrySet())
-                knownKeys.put(entry.getKey(), entry.getValue().get(0).key());
-
-            this.knownKeysWithIds = knownKeysWithIds;
-        }
-
-        /** Known i.e. stored in {@code ReadWriteMetastorage} keys from node (in compatible format). */
-        Map<Integer, byte[]> knownKeys;
-
-        /**  New keys i.e. keys for a local statically configured caches. */
-        Map<Integer, byte[]> newKeys;
-
-        /** Master key digest. */
-        byte[] masterKeyDigest;
-
-        /** Known i.e. stored in {@code ReadWriteMetastorage} keys from node. */
-        Map<Integer, List<GroupKeyEncrypted>> knownKeysWithIds;
-    }
-
     /** */
     private class GenerateEncryptionKeyFuture extends GridFutureAdapter<T2<Collection<byte[]>, byte[]>> {
         /** */

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyEncrypted.java

@@ -18,19 +18,26 @@
 package org.apache.ignite.internal.managers.encryption;
 
 import java.io.Serializable;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
 /**
  * Cache group encryption key with identifier. Key is encrypted.
  */
-public class GroupKeyEncrypted implements Serializable {
+public class GroupKeyEncrypted implements Serializable, Message {
     /** Serial version UID. */
     private static final long serialVersionUID = 0L;
 
     /** Encryption key ID. */
-    private final int id;
+    @Order(0)
+    int id;
 
     /** Encryption key. */
-    private final byte[] key;
+    @Order(1)
+    byte[] key;
+
+    /** */
+    public GroupKeyEncrypted() {}
 
     /**
      * @param id Encryption key ID.

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/NodeEncryptionKeys.java

@@ -0,0 +1,73 @@
+/*
+ * 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.managers.encryption;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/** */
+public class NodeEncryptionKeys implements Serializable, Message {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Known i.e. stored in {@code ReadWriteMetastorage} keys from node (in compatible format). */
+    @Order(0)
+    Map<Integer, byte[]> knownKeys;
+
+    /**  New keys i.e. keys for a local statically configured caches. */
+    @Order(1)
+    Map<Integer, byte[]> newKeys;
+
+    /** Master key digest. */
+    @Order(2)
+    byte[] masterKeyDigest;
+
+    /** Known i.e. stored in {@code ReadWriteMetastorage} keys from node. */
+    @Order(3)
+    Map<Integer, List<GroupKeyEncrypted>> knownKeysWithIds;
+
+    /** */
+    public NodeEncryptionKeys() {}
+
+    /** */
+    NodeEncryptionKeys(
+        HashMap<Integer, List<GroupKeyEncrypted>> knownKeysWithIds,
+        Map<Integer, byte[]> newKeys,
+        byte[] masterKeyDigest
+    ) {
+        this.newKeys = newKeys;
+        this.masterKeyDigest = masterKeyDigest;
+
+        if (F.isEmpty(knownKeysWithIds))
+            return;
+
+        // To be able to join the old cluster.
+        knownKeys = U.newHashMap(knownKeysWithIds.size());
+
+        for (Map.Entry<Integer, List<GroupKeyEncrypted>> entry : knownKeysWithIds.entrySet())
+            knownKeys.put(entry.getKey(), entry.getValue().get(0).key());
+
+        this.knownKeysWithIds = knownKeysWithIds;
+    }
+}

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

@@ -83,6 +83,7 @@
 import org.apache.ignite.plugin.CachePluginProvider;
 import org.apache.ignite.plugin.PluginProvider;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag;
+import org.apache.ignite.spi.discovery.ObjectData;
 import org.apache.ignite.spi.systemview.view.CacheGroupView;
 import org.apache.ignite.spi.systemview.view.CacheView;
 import org.jetbrains.annotations.Nullable;
@@ -1257,7 +1258,7 @@ private boolean validateStartNewCache(
      * @param dataBag Discovery data bag.
      */
     void collectJoiningNodeData(DiscoveryDataBag dataBag) {
-        dataBag.addJoiningNodeData(CACHE_PROC.ordinal(), joinDiscoveryData());
+        dataBag.addJoiningNodeData(CACHE_PROC.ordinal(), new ObjectData(joinDiscoveryData()));
     }
 
     /**
@@ -2038,7 +2039,7 @@ public ExchangeActions onStateChangeRequest(ChangeGlobalStateMessage msg, Affini
      */
     public void onJoiningNodeDataReceived(DiscoveryDataBag.JoiningNodeDiscoveryData data) {
         if (data.hasJoiningNodeData()) {
-            Serializable joiningNodeData = data.joiningNodeData();
+            Serializable joiningNodeData = ObjectData.unwrap(data.joiningNodeData());
 
             if (joiningNodeData instanceof CacheClientReconnectDiscoveryData) {
                 if (disconnectedState()) {
@@ -2062,7 +2063,7 @@ else if (joiningNodeData instanceof CacheJoinNodeDiscoveryData)
      */
     public String validateJoiningNodeData(DiscoveryDataBag.JoiningNodeDiscoveryData data, boolean joiningNodeClient) {
         if (data.hasJoiningNodeData()) {
-            Serializable joiningNodeData = data.joiningNodeData();
+            Serializable joiningNodeData = ObjectData.unwrap(data.joiningNodeData());
 
             if (joiningNodeData instanceof CacheJoinNodeDiscoveryData) {
                 CacheJoinNodeDiscoveryData joinData = (CacheJoinNodeDiscoveryData)joiningNodeData;

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

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.cache;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -60,6 +61,7 @@
 import org.apache.ignite.plugin.security.SecurityException;
 import org.apache.ignite.spi.IgniteNodeValidationResult;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag;
+import org.apache.ignite.spi.discovery.ObjectData;
 import org.apache.ignite.spi.encryption.EncryptionSpi;
 import org.jetbrains.annotations.Nullable;
 
@@ -123,8 +125,10 @@ public class ValidationOnNodeJoinUtils {
         GridKernalContext ctx,
         Function<String, DynamicCacheDescriptor> cacheDescProvider
     ) {
-        if (discoData.hasJoiningNodeData() && discoData.joiningNodeData() instanceof CacheJoinNodeDiscoveryData) {
-            CacheJoinNodeDiscoveryData nodeData = (CacheJoinNodeDiscoveryData)discoData.joiningNodeData();
+        Serializable joiningNodeData = ObjectData.unwrap(discoData.joiningNodeData());
+
+        if (joiningNodeData instanceof CacheJoinNodeDiscoveryData) {
+            CacheJoinNodeDiscoveryData nodeData = (CacheJoinNodeDiscoveryData)joiningNodeData;
 
             boolean isGridActive = ctx.state().clusterState().active();
 

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

@@ -841,15 +841,6 @@ private final class MetadataRequestListener implements GridMessageListener {
 
             MetadataResponseMessage resp = new MetadataResponseMessage(typeId);
 
-            if (metaVerInfo != null) {
-                try {
-                    metaVerInfo.marshalMetadata();
-                }
-                catch (IgniteCheckedException e) {
-                    U.error(log, "Failed to marshal binary metadata for [typeId=" + typeId + ']', e);
-                }
-            }
-
             resp.metadataVersionInfo(metaVerInfo);
 
             try {
@@ -890,16 +881,9 @@ private final class MetadataResponseListener implements GridMessageListener {
                 return;
             }
 
-            try {
-                metaVerInfo.unmarshalMetadata();
-
-                casBinaryMetadata(typeId, metaVerInfo);
+            casBinaryMetadata(typeId, metaVerInfo);
 
-                fut.onDone(MetadataUpdateResult.createSuccessfulResult(-1));
-            }
-            catch (IgniteCheckedException e) {
-                fut.onDone(MetadataUpdateResult.createFailureResult(new BinaryObjectException(e)));
-            }
+            fut.onDone(MetadataUpdateResult.createSuccessfulResult(-1));
         }
     }