IGNITE-27556 : Use MessageSerializer for TcpDiscoveryNodeAddedMessag v3 minimal

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

@@ -45,6 +45,8 @@
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingRequestSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingResponse;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingResponseSerializer;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCollectionMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCollectionMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryConnectionCheckMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryConnectionCheckMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
@@ -65,6 +67,8 @@
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryMetricsUpdateMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessageSerializer;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddedMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddedMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeFailedMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeFailedMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeFullMetricsMessage;
@@ -88,6 +92,7 @@
 public class DiscoveryMessageFactory implements MessageFactoryProvider {
     /** {@inheritDoc} */
     @Override public void registerAll(MessageFactory factory) {
+        factory.register((short)-108, TcpDiscoveryCollectionMessage::new, new TcpDiscoveryCollectionMessageSerializer());
         factory.register((short)-107, NodeSpecificData::new, new NodeSpecificDataSerializer());
         factory.register((short)-106, DiscoveryDataPacket::new, new DiscoveryDataPacketSerializer());
         factory.register((short)-105, TcpDiscoveryNodeFullMetricsMessage::new,
@@ -123,6 +128,7 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
         factory.register((short)21, TcpDiscoveryCustomEventMessage::new, new TcpDiscoveryCustomEventMessageSerializer());
         factory.register((short)22, TcpDiscoveryServerOnlyCustomEventMessage::new,
             new TcpDiscoveryServerOnlyCustomEventMessageSerializer());
+        factory.register((short)23, TcpDiscoveryNodeAddedMessage::new, new TcpDiscoveryNodeAddedMessageSerializer());
 
         // DiscoveryCustomMessage
         factory.register((short)500, CacheStatisticsModeChangeMessage::new, new CacheStatisticsModeChangeMessageSerializer());

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

@@ -2163,8 +2163,12 @@ protected void processDiscoveryMessage(TcpDiscoveryAbstractMessage msg) {
             if (msg instanceof TraceableMessage)
                 tracing.messages().beforeSend((TraceableMessage)msg);
 
-            if (msg instanceof TcpDiscoveryNodeAddedMessage)
+            if (msg instanceof TcpDiscoveryNodeAddedMessage) {
+                ((TcpDiscoveryNodeAddedMessage)msg).finishUnmarshal(spi.marshaller(),
+                    U.resolveClassLoader(spi.ignite().configuration()));
+
                 processNodeAddedMessage((TcpDiscoveryNodeAddedMessage)msg);
+            }
             else if (msg instanceof TcpDiscoveryNodeAddFinishedMessage)
                 processNodeAddFinishedMessage((TcpDiscoveryNodeAddFinishedMessage)msg);
             else if (msg instanceof TcpDiscoveryNodeLeftMessage)
@@ -2216,16 +2220,16 @@ private void processNodeAddedMessage(TcpDiscoveryNodeAddedMessage msg) {
             if (spi.getSpiContext().isStopping())
                 return;
 
-            TcpDiscoveryNode node = msg.node();
+            TcpDiscoveryNode node = msg.node;
 
             UUID newNodeId = node.id();
 
             if (getLocalNodeId().equals(newNodeId)) {
                 if (joining()) {
-                    Collection<TcpDiscoveryNode> top = msg.topology();
+                    Collection<TcpDiscoveryNode> top = msg.top;
 
                     if (top != null) {
-                        spi.gridStartTime = msg.gridStartTime();
+                        spi.gridStartTime = msg.gridStartTime;
 
                         if (disconnected())
                             rmtNodes.clear();
@@ -2241,8 +2245,8 @@ private void processNodeAddedMessage(TcpDiscoveryNodeAddedMessage msg) {
 
                         nodeAdded = true;
 
-                        if (msg.topologyHistory() != null)
-                            topHist.putAll(msg.topologyHistory());
+                        if (!F.isEmpty(msg.topHist))
+                            topHist.putAll(msg.topHist);
                     }
                     else {
                         if (log.isDebugEnabled())
@@ -2261,7 +2265,7 @@ else if (log.isDebugEnabled())
                         if (log.isDebugEnabled())
                             log.debug("Added new node to topology: " + node);
 
-                        DiscoveryDataPacket dataPacket = msg.gridDiscoveryData();
+                        DiscoveryDataPacket dataPacket = msg.dataPacket;
 
                         if (dataPacket != null && dataPacket.hasJoiningNodeData()) {
                             if (joining())
@@ -2296,9 +2300,9 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
                 }
             }
 
-            if (getLocalNodeId().equals(msg.nodeId())) {
+            if (getLocalNodeId().equals(msg.nodeId)) {
                 if (joining()) {
-                    DiscoveryDataPacket dataContainer = msg.clientDiscoData();
+                    DiscoveryDataPacket dataContainer = msg.clientDiscoData;
 
                     if (dataContainer != null)
                         spi.onExchange(dataContainer, U.resolveClassLoader(spi.ignite().configuration()));
@@ -2312,7 +2316,7 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
 
                     msg.finishUnmarshal(spi.marshaller(), U.resolveClassLoader(spi.ignite().configuration()));
 
-                    locNode.setAttributes(msg.clientNodeAttributes());
+                    locNode.setAttributes(msg.clientNodeAttrs);
 
                     clearNodeSensitiveData(locNode);
 
@@ -2355,7 +2359,7 @@ else if (log.isDebugEnabled())
             }
             else {
                 if (nodeAdded()) {
-                    TcpDiscoveryNode node = rmtNodes.get(msg.nodeId());
+                    TcpDiscoveryNode node = rmtNodes.get(msg.nodeId);
 
                     if (node == null) {
                         if (log.isDebugEnabled())