IGNITE-27722 Investigate duplicated serialization in RingMessageWorker#sendMessageToClients

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

@@ -7649,6 +7649,16 @@ void metrics(ClusterMetrics metrics) {
          * @param msg Message.
          */
         void addMessage(TcpDiscoveryAbstractMessage msg) {
+            try {
+                ses.serializeMessage(msg);
+            }
+            catch (IgniteCheckedException | IOException e) {
+                if (log.isDebugEnabled())
+                    U.error(log, "Serialization failed [msg=" + msg + ']', e);
+
+                onException("Serialization failed [msg=" + msg + ']', e);
+            }
+
             if (msg.highPriority())
                 queue.addFirst(msg);
             else
@@ -7667,8 +7677,6 @@ void addMessage(TcpDiscoveryAbstractMessage msg) {
             try {
                 assert msg.verified() : msg;
 
-                byte[] msgBytes = ses.serializeMessage(msg);
-
                 DebugLogger msgLog = messageLogger(msg);
 
                 if (msg instanceof TcpDiscoveryClientAckResponse) {
@@ -7690,7 +7698,7 @@ else if (msgLog.isDebugEnabled()) {
                                 + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
                         }
 
-                        spi.writeToSocket(sock, msg, msgBytes, spi.failureDetectionTimeoutEnabled() ?
+                        spi.writeToSocket(sock, msg, msg.serializedData(), spi.failureDetectionTimeoutEnabled() ?
                             spi.clientFailureDetectionTimeout() : spi.getSocketTimeout());
                     }
                 }
@@ -7702,7 +7710,7 @@ else if (msgLog.isDebugEnabled()) {
 
                     assert topologyInitialized(msg) : msg;
 
-                    spi.writeToSocket(sock, msg, msgBytes, spi.getEffectiveSocketTimeout(false));
+                    spi.writeToSocket(sock, msg, msg.serializedData(), spi.getEffectiveSocketTimeout(false));
                 }
 
                 boolean clientFailed = msg instanceof TcpDiscoveryNodeFailedMessage &&
@@ -7712,7 +7720,7 @@ else if (msgLog.isDebugEnabled()) {
 
                 success = !clientFailed;
             }
-            catch (IgniteCheckedException | IOException e) {
+            catch (IOException e) {
                 if (log.isDebugEnabled())
                     U.error(log, "Client connection failed [sock=" + sock + ", locNodeId="
                         + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']', e);

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

@@ -242,19 +242,21 @@ <T> T readMessage() throws IgniteCheckedException, IOException {
      * Serializes a discovery message into a byte array.
      *
      * @param msg Discovery message to serialize.
-     * @return Serialized byte array containing the message data.
      * @throws IgniteCheckedException If serialization fails.
      * @throws IOException If serialization fails.
      */
-    byte[] serializeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException, IOException {
-        if (!(msg instanceof Message))
-            return U.marshal(spi.marshaller(), msg);
+    void serializeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException, IOException {
+        if (msg.serializedData() != null)
+            return;
 
-        try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
-            serializeMessage((Message)msg, out);
+        if (!(msg instanceof Message))
+            msg.serializedData(U.marshal(spi.marshaller(), msg));
+        else
+            try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+                serializeMessage((Message)msg, out);
 
-            return out.toByteArray();
-        }
+                msg.serializedData(out.toByteArray());
+            }
     }
 
     /** @return Socket. */

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

@@ -79,6 +79,11 @@ public abstract class TcpDiscoveryAbstractMessage implements Serializable {
     @Order(4)
     private Set<UUID> failedNodes;
 
+    /** Helps to avoid multiply serialization in case message is sent more than once using different workers. */
+    public byte[] serializedData() {
+        return serData;
+    }
+
     /**
      * Default no-arg constructor for {@link Externalizable} interface.
      */
@@ -105,6 +110,14 @@ protected TcpDiscoveryAbstractMessage(TcpDiscoveryAbstractMessage msg) {
         flags = msg.flags;
     }
 
+    /** */
+    public void serializedData(byte[] serData) {
+        this.serData = serData;
+    }
+
+    /** */
+    private byte[] serData;
+
     /**
      * @return {@code True} if need use trace logging for this message (to reduce amount of logging with debug level).
      */