IGNITE-28791 : Use MessageSerializer to transfer metastorage DTO's in DataBag

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

@@ -220,6 +220,9 @@
 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.DistributedMetaStorageClusterNodeData;
+import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageHistoryItemMessage;
+import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageJoiningNodeData;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateAckMessage;
 import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateMessage;
 import org.apache.ignite.internal.processors.plugin.PluginsDataBagItem;
@@ -727,6 +730,9 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) {
         register(BaselineTopologyHistory.class);
         register(BaselineTopologyHistoryItem.class);
         register(DiscoveryDataClusterState.class);
+        register(DistributedMetaStorageHistoryItemMessage.class);
+        register(DistributedMetaStorageJoiningNodeData.class);
+        register(DistributedMetaStorageClusterNodeData.class);
 
         // [13400 - 13500]: Operation context messages.
         msgIdx = 13400;

modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageClusterNodeData.java

@@ -17,51 +17,85 @@
 
 package org.apache.ignite.internal.processors.metastorage.persistence;
 
-import java.io.Serializable;
+import java.io.Externalizable;
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
 
 /**
- * Distributed metastorage data that cluster sends to joining node.
+ * Distributed metastorage data that cluster sends to joining node. To reduce messages number, contains plain representation
+ * of {@link DistributedMetaStorageVersion}, arrays of plain representations of Distributed MetaStorage's key-value pairs.
+ * And wrapped {@link DistributedMetaStorageHistoryItem}s. The version and the full data holders are {@link Externalizable}s
+ * persistent by {@link MetaStorage} with the dedicated code-generated serializers. Thus, we do not make them directly a {@link Message}.
+ *
+ * @see DmsDataWriter#write(String, byte[])
+ * @see MetaStorage#write(String, Serializable)
  */
-@SuppressWarnings({"PublicField", "AssignmentOrReturnOfFieldWithMutableType"})
-class DistributedMetaStorageClusterNodeData implements Serializable {
-    /** */
-    private static final long serialVersionUID = 0L;
-
-    /**
-     * Distributed metastorage version of cluster. If {@link #fullData} is not null then this version corresponds to
-     * its content.
-     */
-    public final DistributedMetaStorageVersion ver;
-
-    /**
-     * Full data is sent if there's not enough history items on local node.
-     */
-    public final DistributedMetaStorageKeyValuePair[] fullData;
-
-    /**
-     * Required updates for joining nodes or full available history of local node if {@link #fullData} is
-     * not {@code null}.
-     */
-    public final DistributedMetaStorageHistoryItem[] hist;
-
-    /**
-     * Additional updates. Makes sence only if {@link #fullData} is not {@code null}.
-     */
-    public DistributedMetaStorageHistoryItem[] updates;
+public class DistributedMetaStorageClusterNodeData implements Message {
+    /** @see DistributedMetaStorageVersion#id */
+    @Order(0)
+    @GridToStringInclude
+    long dVerId;
+
+    /** @see DistributedMetaStorageVersion#hash */
+    @Order(1)
+    @GridToStringInclude
+    long dVerHash;
+
+    /** Array of the full data keys. */
+    @GridToStringInclude
+    @Order(2)
+    @Nullable String[] fullDataKeys;
+
+    /** Arrays of the full data bytes. */
+    @GridToStringInclude
+    @Order(3)
+    @Nullable byte[][] fullDataValsBytes;
+
+    /** Required updates for joining nodes or full available history of local node if the full data is not {@code null}. */
+    @Order(4)
+    @Nullable DistributedMetaStorageHistoryItemMessage[] hist;
+
+    /** Additional updates. Makes sense only if the full data is not {@code null}. */
+    @Order(5)
+    @Nullable DistributedMetaStorageHistoryItemMessage[] updates;
+
+    /** Empty constructor for serialization purposes. */
+    public DistributedMetaStorageClusterNodeData() {
+        // No-op.
+    }
 
     /** */
     public DistributedMetaStorageClusterNodeData(
         DistributedMetaStorageVersion ver,
-        DistributedMetaStorageKeyValuePair[] fullData,
-        DistributedMetaStorageHistoryItem[] hist,
-        DistributedMetaStorageHistoryItem[] updates
+        @Nullable Map<String, byte[]> fullData,
+        @Nullable DistributedMetaStorageHistoryItem[] hist,
+        @Nullable DistributedMetaStorageHistoryItem[] updates
     ) {
         assert ver != null;
         assert fullData == null || hist != null;
 
-        this.fullData = fullData;
-        this.ver = ver;
-        this.hist = hist;
-        this.updates = updates;
+        dVerId = ver.id;
+        dVerHash = ver.hash;
+
+        if (fullData != null) {
+            fullDataKeys = new String[fullData.size()];
+            fullDataValsBytes = new byte[fullData.size()][];
+
+            int i = 0;
+
+            for (var e : fullData.entrySet()) {
+                fullDataKeys[i] = e.getKey();
+                fullDataValsBytes[i] = e.getValue();
+
+                ++i;
+            }
+        }
+
+        this.hist = DistributedMetaStorageHistoryItemMessage.toMessages(hist);
+        this.updates = DistributedMetaStorageHistoryItemMessage.toMessages(updates);
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItem.java

@@ -19,10 +19,19 @@
 
 import java.util.Arrays;
 import org.apache.ignite.internal.dto.IgniteDataTransferObject;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
-/** */
+/**
+ * Distributed Metastorage history items holder. Is a persistent {@link IgniteDataTransferObject} stored by {@link MetaStorage}
+ * using the dedicated code-generated DTO-serializer. Then, has a transfer wrap {@link DistributedMetaStorageHistoryItemMessage}.
+ *
+ * @see DistributedMetaStorageHistoryItemMessage
+ * @see DmsDataWriter#write(String, byte[])
+ * @see MetaStorage#write(String, Serializable)
+ */
 final class DistributedMetaStorageHistoryItem extends IgniteDataTransferObject {
     /** */
     private static final long serialVersionUID = 0L;
@@ -62,6 +71,16 @@ public DistributedMetaStorageHistoryItem(String[] keys, byte[][] valBytesArr) {
         this.valBytesArr = valBytesArr;
     }
 
+    /** @return Array of {@link DistributedMetaStorageHistoryItem} created of the related {@link Message} transfer wraps. */
+    static DistributedMetaStorageHistoryItem[] fromMessages(DistributedMetaStorageHistoryItemMessage[] histMsgs) {
+        DistributedMetaStorageHistoryItem[] res = new DistributedMetaStorageHistoryItem[histMsgs.length];
+
+        for (int i = 0; i < histMsgs.length; ++i)
+            res[i] = new DistributedMetaStorageHistoryItem(histMsgs[i].keys, histMsgs[i].valBytes);
+
+        return res;
+    }
+
     /** */
     public long estimateSize() {
         int len = keys.length;

modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItemMessage.java

@@ -0,0 +1,67 @@
+/*
+ * 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.metastorage.persistence;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.dto.IgniteDataTransferObject;
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Transfer wrap for {@link DistributedMetaStorageHistoryItem} which is a persistent {@link IgniteDataTransferObject} stored
+ * by {@link MetaStorage} using the dedicated code-generated DTO-serializer.
+ *
+ * @see DistributedMetaStorageHistoryItem
+ * @see DmsDataWriter#write(String, byte[])
+ * @see MetaStorage#write(String, Serializable)
+ */
+public class DistributedMetaStorageHistoryItemMessage implements Message {
+    /** */
+    @Order(0)
+    String[] keys;
+
+    /** */
+    @Order(1)
+    byte[][] valBytes;
+
+    /** Empty constructor for serialization purposes. */
+    public DistributedMetaStorageHistoryItemMessage() {
+        // No-op.
+    }
+
+    /** */
+    DistributedMetaStorageHistoryItemMessage(String[] keys, byte[][] valBytes) {
+        this.keys = keys;
+        this.valBytes = valBytes;
+    }
+
+    /** @return {@link Message} wraps array for {@code hist}. */
+    static @Nullable DistributedMetaStorageHistoryItemMessage[] toMessages(@Nullable DistributedMetaStorageHistoryItem[] hist) {
+        if (hist == null)
+            return null;
+
+        var res = new DistributedMetaStorageHistoryItemMessage[hist.length];
+
+        for (int i = 0; i < hist.length; ++i)
+            res[i] = new DistributedMetaStorageHistoryItemMessage(hist[i].keys, hist[i].valBytesArr);
+
+        return res;
+    }
+}