IGNITE-26111 TcpDiscoverySpi uses MessageSerializer

Author: timoninmaxim

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

@@ -49,6 +49,7 @@
 import org.apache.ignite.internal.codegen.MissingMappingResponseMessageSerializer;
 import org.apache.ignite.internal.codegen.NearCacheUpdatesSerializer;
 import org.apache.ignite.internal.codegen.SessionChannelMessageSerializer;
+import org.apache.ignite.internal.codegen.TcpDiscoveryCheckFailedMessageSerializer;
 import org.apache.ignite.internal.codegen.TcpInverseConnectionResponseMessageSerializer;
 import org.apache.ignite.internal.codegen.TxLockSerializer;
 import org.apache.ignite.internal.codegen.TxLocksRequestSerializer;
@@ -199,6 +200,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.tcp.messages.TcpDiscoveryCheckFailedMessage;
 
 /**
  * Message factory implementation.
@@ -379,5 +381,8 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
         // [2048..2053] - Snapshots
         // [-42..-37] - former hadoop.
         // [64..71] - former IGFS.
+
+        // Discovery messages.
+        factory.register((short)-1000, TcpDiscoveryCheckFailedMessage::new, new TcpDiscoveryCheckFailedMessageSerializer());
     }
 }

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

@@ -345,6 +345,7 @@ public GridDiscoveryManager(GridKernalContext ctx) {
         DiscoverySpi spi = getSpi();
 
         spi.setNodeAttributes(ctx.nodeAttributes(), VER);
+        spi.setMessageFactory(ctx.io().messageFactory());
     }
 
     /**

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

@@ -24,6 +24,7 @@
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.lang.IgniteProductVersion;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.spi.IgniteSpiAdapter;
 import org.apache.ignite.spi.IgniteSpiException;
 import org.apache.ignite.spi.IgniteSpiNoop;
@@ -70,6 +71,11 @@ public class StandaloneNoopDiscoverySpi extends IgniteSpiAdapter implements Disc
 
     }
 
+    /** {@inheritDoc} */
+    @Override public void setMessageFactory(MessageFactory msgFactory) {
+
+    }
+
     /** {@inheritDoc} */
     @Override public void setListener(@Nullable DiscoverySpiListener lsnr) {
 

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

@@ -24,6 +24,7 @@
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.lang.IgniteProductVersion;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.spi.IgniteSpi;
 import org.apache.ignite.spi.IgniteSpiException;
 import org.jetbrains.annotations.Nullable;
@@ -94,6 +95,11 @@ public interface DiscoverySpi extends IgniteSpi {
      */
     public void setNodeAttributes(Map<String, Object> attrs, IgniteProductVersion ver);
 
+    /**
+     * @param msgFactory Message factory.
+     */
+    public void setMessageFactory(MessageFactory msgFactory);
+
     /**
      * Sets a listener for discovery events. Refer to
      * {@link org.apache.ignite.events.DiscoveryEvent} for a set of all possible

modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedDiscoverySpi.java

@@ -35,6 +35,7 @@
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.lang.IgniteProductVersion;
 import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.security.SecurityCredentials;
 import org.apache.ignite.spi.IgniteSpiAdapter;
 import org.apache.ignite.spi.IgniteSpiContext;
@@ -129,6 +130,11 @@ public class IsolatedDiscoverySpi extends IgniteSpiAdapter implements IgniteDisc
         locNode = new IsolatedNode(ignite.configuration().getNodeId(), attrs, ver);
     }
 
+    /** {@inheritDoc} */
+    @Override public void setMessageFactory(MessageFactory msgFactory) {
+        // Np-op.
+    }
+
     /** {@inheritDoc} */
     @Override public void setListener(@Nullable DiscoverySpiListener lsnr) {
         this.lsnr = lsnr;

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

@@ -1186,7 +1186,7 @@ private void forceStopRead() throws InterruptedException {
                         TcpDiscoveryAbstractMessage msg;
 
                         try {
-                            msg = U.unmarshal(spi.marshaller(), in, U.resolveClassLoader(spi.ignite().configuration()));
+                            msg = spi.readMessage(sock, in, sock.getSoTimeout());
                         }
                         catch (IgniteCheckedException e) {
                             if (log.isDebugEnabled())
@@ -1616,8 +1616,7 @@ public void cancel() {
                         List<TcpDiscoveryAbstractMessage> msgs = null;
 
                         while (!isInterrupted()) {
-                            TcpDiscoveryAbstractMessage msg = U.unmarshal(spi.marshaller(), in,
-                                U.resolveClassLoader(spi.ignite().configuration()));
+                            TcpDiscoveryAbstractMessage msg = spi.readMessage(sock, in, sock.getSoTimeout());
 
                             if (msg instanceof TcpDiscoveryClientReconnectMessage) {
                                 TcpDiscoveryClientReconnectMessage res = (TcpDiscoveryClientReconnectMessage)msg;

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

@@ -3376,7 +3376,6 @@ private void sendMessageToClients(TcpDiscoveryAbstractMessage msg) {
                     }
 
                     TcpDiscoveryAbstractMessage msg0 = msg;
-                    byte[] msgBytes0 = msgBytes;
 
                     if (msg instanceof TcpDiscoveryNodeAddedMessage) {
                         TcpDiscoveryNodeAddedMessage nodeAddedMsg = (TcpDiscoveryNodeAddedMessage)msg;
@@ -3389,16 +3388,14 @@ private void sendMessageToClients(TcpDiscoveryAbstractMessage msg) {
                                     U.resolveClassLoader(spi.ignite().configuration()));
 
                                 prepareNodeAddedMessage(msg0, clientMsgWorker.clientNodeId, null, null);
-
-                                msgBytes0 = null;
                             }
                             catch (IgniteCheckedException e) {
                                 U.error(log, "Failed to create message copy: " + msg, e);
                             }
                         }
                     }
 
-                    clientMsgWorker.addMessage(msg0, msgBytes0);
+                    clientMsgWorker.addMessage(msg0);
                 }
             }
         }
@@ -6985,8 +6982,7 @@ else if (e.hasCause(ObjectStreamException.class) ||
                     try {
                         SecurityUtils.serializeVersion(1);
 
-                        TcpDiscoveryAbstractMessage msg = U.unmarshal(spi.marshaller(), in,
-                            U.resolveClassLoader(spi.ignite().configuration()));
+                        TcpDiscoveryAbstractMessage msg = spi.readMessage(sock, in, sock.getSoTimeout());
 
                         msg.senderNodeId(nodeId);
 
@@ -7758,11 +7754,6 @@ void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[] msgBytes) {
             try {
                 assert msg.verified() : msg;
 
-                byte[] msgBytes = msgT.get2();
-
-                if (msgBytes == null)
-                    msgBytes = U.marshal(spi.marshaller(), msg);
-
                 DebugLogger msgLog = messageLogger(msg);
 
                 if (msg instanceof TcpDiscoveryClientAckResponse) {
@@ -7784,7 +7775,7 @@ else if (msgLog.isDebugEnabled()) {
                                 + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
                         }
 
-                        spi.writeToSocket(sock, msg, msgBytes, spi.failureDetectionTimeoutEnabled() ?
+                        spi.writeToSocket(sock, msg, spi.failureDetectionTimeoutEnabled() ?
                             spi.clientFailureDetectionTimeout() : spi.getSocketTimeout());
                     }
                 }
@@ -7796,7 +7787,7 @@ else if (msgLog.isDebugEnabled()) {
 
                     assert topologyInitialized(msg) : msg;
 
-                    spi.writeToSocket(sock, msg, msgBytes, spi.getEffectiveSocketTimeout(false));
+                    spi.writeToSocket(sock, msg, spi.getEffectiveSocketTimeout(false));
                 }
 
                 boolean clientFailed = msg instanceof TcpDiscoveryNodeFailedMessage &&