IGNITE-27992 Use MessageSerializer for WalStateAbstractMessage and successors

Author: wernerdv

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

@@ -17,6 +17,10 @@
 
 package org.apache.ignite.internal.managers.discovery;
 
+import org.apache.ignite.internal.processors.cache.WalStateFinishMessage;
+import org.apache.ignite.internal.processors.cache.WalStateFinishMessageSerializer;
+import org.apache.ignite.internal.processors.cache.WalStateProposeMessage;
+import org.apache.ignite.internal.processors.cache.WalStateProposeMessageSerializer;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket;
@@ -113,5 +117,8 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
         factory.register((short)18, TcpDiscoveryStatusCheckMessage::new, new TcpDiscoveryStatusCheckMessageSerializer());
         factory.register((short)19, TcpDiscoveryNodeAddFinishedMessage::new, new TcpDiscoveryNodeAddFinishedMessageSerializer());
         factory.register((short)20, TcpDiscoveryJoinRequestMessage::new, new TcpDiscoveryJoinRequestMessageSerializer());
+
+        factory.register((short)500, WalStateFinishMessage::new, new WalStateFinishMessageSerializer());
+        factory.register((short)501, WalStateProposeMessage::new, new WalStateProposeMessageSerializer());
     }
 }

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

@@ -18,35 +18,46 @@
 package org.apache.ignite.internal.processors.cache;
 
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.Nullable;
 
 /**
  * WAL state change abstract message.
  */
-public abstract class WalStateAbstractMessage implements DiscoveryCustomMessage {
+public abstract class WalStateAbstractMessage implements DiscoveryCustomMessage, Message {
     /** Message ID */
-    private final IgniteUuid id = IgniteUuid.randomUuid();
+    @Order(0)
+    IgniteUuid id;
 
     /** Unique operation ID. */
-    private final UUID opId;
+    @Order(1)
+    UUID opId;
 
     /** Group ID. */
-    private int grpId;
+    @Order(2)
+    int grpId;
 
     /** Group deployment ID. */
-    private IgniteUuid grpDepId;
+    @Order(3)
+    IgniteUuid grpDepId;
 
     /** Message that should be processed through exchange thread. */
     @GridToStringExclude
     private transient WalStateProposeMessage exchangeMsg;
 
+    /** Constructor. */
+    protected WalStateAbstractMessage() {
+        // No-op.
+    }
+
     /**
      * Constructor.
      *
@@ -55,6 +66,7 @@ public abstract class WalStateAbstractMessage implements DiscoveryCustomMessage
      * @param grpDepId Group deployment ID.
      */
     protected WalStateAbstractMessage(UUID opId, int grpId, IgniteUuid grpDepId) {
+        id = IgniteUuid.randomUuid();
         this.opId = opId;
         this.grpId = grpId;
         this.grpDepId = grpDepId;

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

@@ -18,6 +18,7 @@
 package org.apache.ignite.internal.processors.cache;
 
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.lang.IgniteUuid;
 import org.jetbrains.annotations.Nullable;
@@ -30,10 +31,17 @@ public class WalStateFinishMessage extends WalStateAbstractMessage {
     private static final long serialVersionUID = 0L;
 
     /** Whether WAL state was changed as a result of this call. */
-    private final boolean changed;
+    @Order(4)
+    boolean changed;
 
     /** Error message. */
-    private final String errMsg;
+    @Order(5)
+    String errMsg;
+
+    /** Constructor. */
+    public WalStateFinishMessage() {
+        // No-op.
+    }
 
     /**
      * Constructor.
@@ -69,4 +77,9 @@ public boolean changed() {
     @Override public String toString() {
         return S.toString(WalStateFinishMessage.class, this, "super", super.toString());
     }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 500;
+    }
 }

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

@@ -19,6 +19,7 @@
 
 import java.util.Map;
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.lang.IgniteUuid;
 
@@ -30,23 +31,29 @@ public class WalStateProposeMessage extends WalStateAbstractMessage {
     private static final long serialVersionUID = 0L;
 
     /** Cache names which are expected to be in the group along with their deployment IDs. */
-    private Map<String, IgniteUuid> caches;
+    @Order(4)
+    Map<String, IgniteUuid> caches;
 
     /** Whether WAL should be enabled or disabled. */
-    private final boolean enable;
+    @Order(5)
+    boolean enable;
 
     /** Whether message is being handled on cache affinity node. */
     private transient boolean affNode;
 
+    /** Constructor. */
+    public WalStateProposeMessage() {
+        // No-op.
+    }
+
     /**
      * Constructor.
      *
-     * @param opId Operation IDs.
+     * @param opId Operation ID.
      * @param grpId Expected group ID.
      * @param grpDepId Expected group deployment ID.
      * @param nodeId Node ID.
      * @param caches Expected cache names and their relevant deployment IDs.
-     *
      * @param enable WAL state flag.
      */
     public WalStateProposeMessage(UUID opId, int grpId, IgniteUuid grpDepId, UUID nodeId,
@@ -89,4 +96,9 @@ public void affinityNode(boolean affNode) {
     @Override public String toString() {
         return S.toString(WalStateProposeMessage.class, this, "super", super.toString());
     }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 501;
+    }
 }