IGNITE-28511 Use MessageSerializer for SnapshotMetadata and WALPointer (#13076)

Author: Vladsz83

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

@@ -147,6 +147,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesFailureMessage;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesRequestMessage;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponse;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationEndRequest;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationRequest;
@@ -155,6 +156,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponse;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreStartRequest;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotStartDiscoveryMessage;
+import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse;
 import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
@@ -343,6 +345,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(CacheVersionedValue.class);
         withNoSchema(GridCacheVersion.class);
         withNoSchema(GridCacheVersionEx.class);
+        withNoSchema(WALPointer.class);
 
         // [5700 - 5900]: Discovery originated messages.
         msgIdx = 5700;
@@ -384,6 +387,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(SnapshotPartitionsVerifyHandlerResponse.class);
         withNoSchema(SnapshotRestoreOperationResponse.class);
         withNoSchema(SnapshotMetadataResponse.class);
+        withNoSchema(SnapshotMetadata.class);
         withNoSchema(SnapshotCheckPartitionHashesResponse.class);
         withNoSchema(SnapshotCheckHandlersResponse.class);
         withNoSchema(SnapshotFilesRequestMessage.class);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java

@@ -30,94 +30,119 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.pagemem.wal.record.delta.ClusterSnapshotRecord;
 import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
 import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * Snapshot metadata file.
  * <p>
  * All changes must be made with the respect of RU rules.
  */
-public class SnapshotMetadata implements Serializable {
+public class SnapshotMetadata implements Message, Serializable {
     /** Serial version uid. */
     private static final long serialVersionUID = 0L;
 
     /** Unique snapshot request id. */
-    private final UUID rqId;
+    @Order(0)
+    UUID rqId;
 
     /** Snapshot name. */
     @GridToStringInclude
-    private final String snpName;
+    @Order(1)
+    String snpName;
 
     /** Consistent id of a node to which this metadata relates. */
     @GridToStringInclude
-    private final String consId;
+    @Order(2)
+    String consId;
 
     /**
      * Directory related to the current consistent node id on which partition files are stored.
      * For some of the cases, consId doesn't equal the directory name.
      */
-    private final String folderName;
+    @Order(3)
+    String folderName;
 
     /**
      * If {@code true} then compress partition files.
      * This shouldn't be confused with {@link SnapshotMetadata#comprGrpIds} which represents how Ignite keeps data in memory pages
      * while {@link SnapshotMetadata#comprParts} represents how dump files are stored on disk.
      */
-    private final boolean comprParts;
+    @Order(4)
+    boolean comprParts;
 
     /** Page size of stored snapshot data. */
-    private final int pageSize;
+    @Order(5)
+    int pageSize;
 
     /** The list of cache groups ids which were included into snapshot. */
     @GridToStringInclude
-    private final List<Integer> grpIds;
+    @Order(6)
+    List<Integer> grpIds;
 
     /** The set of affected by snapshot baseline nodes. */
     @GridToStringInclude
-    private final Set<String> bltNodes;
+    @Order(7)
+    Set<String> bltNodes;
 
     /** WAL pointer to {@link ClusterSnapshotRecord} if exists. */
-    private final @Nullable WALPointer snpRecPtr;
+    @Order(8)
+    @Nullable WALPointer snpRecPtr;
 
     /**
      * Map of cache group partitions from which snapshot has been taken on the local node. This map can be empty
      * since for instance, due to the node filter there is no cache data on node.
      */
     @GridToStringInclude
-    private transient Map<Integer, Set<Integer>> locParts = new HashMap<>();
+    @Order(9)
+    transient Map<Integer, Set<Integer>> locParts = new HashMap<>();
 
     /** Master key digest for encrypted caches. */
     @GridToStringInclude
-    @Nullable private final byte[] masterKeyDigest;
+    @Order(10)
+    @Nullable byte[] masterKeyDigest;
 
     /** Warnings occurred at snapshot creation. */
     @GridToStringInclude
-    @Nullable private List<String> warnings;
+    @Order(11)
+    @Nullable List<String> warnings;
 
     /** Creation timestamp in milliseconds since Unix epoch. */
-    private long snapshotTime;
+    @Order(12)
+    long snapshotTime;
 
     /** */
-    private transient Set<Integer> comprGrpIds;
+    @Order(13)
+    @Nullable transient Set<Integer> comprGrpIds;
 
     /** */
     private boolean hasComprGrps;
 
     /** If {@code true} snapshot only primary copies of partitions. */
-    private boolean onlyPrimary;
+    @Order(14)
+    boolean onlyPrimary;
 
     /** If {@code true} cache group dump stored. */
-    private boolean dump;
+    @Order(15)
+    boolean dump;
 
     /** Encryption key. */
-    private @Nullable byte[] encKey;
+    @Order(16)
+    @Nullable byte[] encKey;
+
+    /** Empty constructor for a {@link MessageFactory}. */
+    public SnapshotMetadata() {
+        // No-op.
+    }
 
     /**
      * @param rqId Unique request id.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataResponse.java

@@ -18,21 +18,15 @@
 package org.apache.ignite.internal.processors.cache.persistence.snapshot;
 
 import java.util.List;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.MarshallableMessage;
 import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 
 /** */
-public class SnapshotMetadataResponse implements MarshallableMessage {
+public class SnapshotMetadataResponse implements Message {
     /** */
     @Order(0)
-    byte[] metadataBytes;
-
-    /** */
-    private List<SnapshotMetadata> metadata;
+    List<SnapshotMetadata> metadata;
 
     /** Default constructor for {@link MessageFactory}. */
     public SnapshotMetadataResponse() {
@@ -48,16 +42,4 @@ public SnapshotMetadataResponse(List<SnapshotMetadata> metadata) {
     public List<SnapshotMetadata> metadata() {
         return metadata;
     }
-
-    /** {@inheritDoc} */
-    @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
-        metadataBytes = U.marshal(marsh, metadata);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
-        if (metadataBytes != null)
-            metadata = U.unmarshal(marsh, metadataBytes, clsLdr);
-    }
-
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/WALPointer.java

@@ -18,27 +18,38 @@
 package org.apache.ignite.internal.processors.cache.persistence.wal;
 
 import java.io.Serializable;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.jetbrains.annotations.NotNull;
 
 /**
  * File WAL pointer.
  */
-public class WALPointer implements Serializable, Comparable<WALPointer> {
+public class WALPointer implements Message, Comparable<WALPointer>, Serializable {
     /** Serial version uid. */
     private static final long serialVersionUID = 0L;
 
     /** Pointer serialized size. */
     public static final int POINTER_SIZE = 16;
 
     /** Absolute WAL segment file index (incrementing counter) */
-    private final long idx;
+    @Order(0)
+    long idx;
 
     /** */
-    private final int fileOff;
+    @Order(1)
+    int fileOff;
 
     /** Written record length */
-    private int len;
+    @Order(2)
+    int len;
+
+    /** Empty constructor for a {@link MessageFactory}. */
+    public WALPointer() {
+        // No-op.
+    }
 
     /**
      * @param idx Absolute WAL segment file index (incremental counter).