IGNITE-27414 : Use MessageSerializer for TcpDiscoveryClientReconnectMessage v2

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

@@ -17,14 +17,21 @@
 
 package org.apache.ignite.internal.managers.discovery;
 
+import java.util.function.Supplier;
+import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.processors.cache.CacheStatisticsModeChangeMessage;
 import org.apache.ignite.internal.processors.cache.CacheStatisticsModeChangeMessageSerializer;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveAcceptedMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveAcceptedMessageSerializer;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposedMessage;
 import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposedMessageSerializer;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
 import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacket;
 import org.apache.ignite.spi.discovery.tcp.internal.DiscoveryDataPacketSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.InetAddressMessage;
@@ -49,6 +56,10 @@
 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.TcpDiscoveryClientReconnectMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientReconnectMessageSerializer;
+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 +76,7 @@
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryJoinRequestMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryLoopbackProblemMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryLoopbackProblemMessageSerializer;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryMarshallableMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryMetricsUpdateMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryMetricsUpdateMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
@@ -87,11 +99,35 @@
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryServerOnlyCustomEventMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryStatusCheckMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryStatusCheckMessageSerializer;
+import org.jetbrains.annotations.Nullable;
 
-/** Message factory for discovery messages. */
+/**
+ * Message factory for discovery messages. Allows to create an enhanced {@link MessageFactory} allowing to create
+ * automated pre- and post- marshalling message serializer for {@link TcpDiscoveryMarshallableMessage}.
+ */
 public class DiscoveryMessageFactory implements MessageFactoryProvider {
+    /** Custom data marshaller. */
+    private final @Nullable Marshaller cstDataMarshall;
+
+    /** Class loader for the custom data marshalling. */
+    private final @Nullable ClassLoader cstDataMarshallClsLdr;
+
+    /**
+     * @param cstDataMarshall Custom data marshaller.
+     * @param cstDataMarshallClsLdr Class loader for the custom data marshalling.
+     */
+    public DiscoveryMessageFactory(@Nullable Marshaller cstDataMarshall, @Nullable ClassLoader cstDataMarshallClsLdr) {
+        assert cstDataMarshall == null && cstDataMarshallClsLdr == null || cstDataMarshall != null && cstDataMarshallClsLdr != null;
+
+        this.cstDataMarshall = cstDataMarshall;
+        this.cstDataMarshallClsLdr = cstDataMarshallClsLdr;
+    }
+
     /** {@inheritDoc} */
     @Override public void registerAll(MessageFactory factory) {
+        factory = enhanceMessageFactory(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,
@@ -127,11 +163,75 @@ 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, TcpDiscoveryClientReconnectMessage::new, new TcpDiscoveryClientReconnectMessageSerializer());
 
         // DiscoveryCustomMessage
         factory.register((short)500, CacheStatisticsModeChangeMessage::new, new CacheStatisticsModeChangeMessageSerializer());
         factory.register((short)501, SecurityAwareCustomMessageWrapper::new, new SecurityAwareCustomMessageWrapperSerializer());
         factory.register((short)502, MetadataRemoveAcceptedMessage::new, new MetadataRemoveAcceptedMessageSerializer());
         factory.register((short)503, MetadataRemoveProposedMessage::new, new MetadataRemoveProposedMessageSerializer());
     }
+
+    /**
+     * @return Enhanced {@link MessageFactory} allowing to create automated pre- and post- marshalling message serializer
+     * for {@link TcpDiscoveryMarshallableMessage}.
+     */
+    private MessageFactory enhanceMessageFactory(MessageFactory mf) {
+        if (cstDataMarshall == null || cstDataMarshallClsLdr == null)
+            return mf;
+
+        return new MessageFactory() {
+            @Override public void register(
+                short directType,
+                Supplier<Message> supplier,
+                MessageSerializer serializer
+            ) throws IgniteException {
+                if (supplier.get() instanceof TcpDiscoveryMarshallableMessage) {
+                    final MessageSerializer serializer0 = serializer;
+
+                    serializer = new MessageSerializer() {
+                        private boolean marshallableMsg;
+
+                        @Override public boolean writeTo(Message msg, MessageWriter writer) {
+                            if (msg instanceof TcpDiscoveryMarshallableMessage && !marshallableMsg) {
+                                marshallableMsg = true;
+
+                                ((TcpDiscoveryMarshallableMessage)msg).prepareMarshal(cstDataMarshall);
+                            }
+
+                            boolean res = serializer0.writeTo(msg, writer);
+
+                            if (res && marshallableMsg)
+                                marshallableMsg = false;
+
+                            return res;
+                        }
+
+                        @Override public boolean readFrom(Message msg, MessageReader reader) {
+                            boolean res = serializer0.readFrom(msg, reader);
+
+                            if (res && msg instanceof TcpDiscoveryMarshallableMessage)
+                                ((TcpDiscoveryMarshallableMessage)msg).finishUnmarshal(cstDataMarshall, cstDataMarshallClsLdr);
+
+                            return res;
+                        }
+                    };
+                }
+
+                mf.register(directType, supplier, serializer);
+            }
+
+            @Override public void register(short directType, Supplier<Message> supplier) throws IgniteException {
+                mf.register(directType, supplier);
+            }
+
+            @Override public Message create(short type) {
+                return mf.create(type);
+            }
+
+            @Override public MessageSerializer serializer(short type) {
+                return mf.serializer(type);
+            }
+        };
+    }
 }

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

@@ -788,8 +788,6 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws
 
                     TcpDiscoveryJoinRequestMessage joinReqMsg = new TcpDiscoveryJoinRequestMessage(node, discoveryData);
 
-                    joinReqMsg.prepareMarshal(spi.marshaller());
-
                     TcpDiscoveryNode nodef = node;
 
                     joinReqMsg.spanContainer().span(
@@ -2241,7 +2239,7 @@ private void processNodeAddedMessage(TcpDiscoveryNodeAddedMessage msg) {
 
                         nodeAdded = true;
 
-                        if (msg.topologyHistory() != null)
+                        if (!F.isEmpty(msg.topologyHistory()))
                             topHist.putAll(msg.topologyHistory());
                     }
                     else {
@@ -2310,8 +2308,6 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
                         delayDiscoData.clear();
                     }
 
-                    msg.finishUnmarshal(spi.marshaller(), U.resolveClassLoader(spi.ignite().configuration()));
-
                     locNode.setAttributes(msg.clientNodeAttributes());
 
                     clearNodeSensitiveData(locNode);
@@ -2548,6 +2544,10 @@ private void processClientReconnectMessage(TcpDiscoveryClientReconnectMessage ms
                 return;
 
             if (getLocalNodeId().equals(msg.creatorNodeId())) {
+                Collection<TcpDiscoveryAbstractMessage> pendingMsgs = msg.pendingMsgsMsg == null
+                    ? Collections.emptyList()
+                    : msg.pendingMsgsMsg.messages();
+
                 if (reconnector != null) {
                     assert msg.success() : msg;
 
@@ -2558,7 +2558,7 @@ private void processClientReconnectMessage(TcpDiscoveryClientReconnectMessage ms
 
                     reconnector = null;
 
-                    for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
+                    for (TcpDiscoveryAbstractMessage pendingMsg : pendingMsgs) {
                         if (log.isDebugEnabled())
                             log.debug("Process pending message on reconnect [msg=" + pendingMsg + ']');
 
@@ -2568,7 +2568,7 @@ private void processClientReconnectMessage(TcpDiscoveryClientReconnectMessage ms
                 else {
                     if (joinLatch.getCount() > 0) {
                         if (msg.success()) {
-                            for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
+                            for (TcpDiscoveryAbstractMessage pendingMsg : pendingMsgs) {
                                 if (log.isDebugEnabled())
                                     log.debug("Process pending message on connect [msg=" + pendingMsg + ']');
 

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

@@ -137,6 +137,7 @@
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingRequest;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingResponse;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientReconnectMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCollectionMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryConnectionCheckMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryDiscardMessage;
@@ -1124,8 +1125,6 @@ private void joinTopology() throws IgniteSpiException {
 
         TcpDiscoveryJoinRequestMessage joinReqMsg = new TcpDiscoveryJoinRequestMessage(locNode, discoveryData);
 
-        joinReqMsg.prepareMarshal(spi.marshaller());
-
         joinReqMsg.spanContainer().span(
             tracing.create(TraceableMessagesTable.traceName(joinReqMsg.getClass()))
                 .addTag(SpanTags.tag(SpanTags.EVENT_NODE, SpanTags.ID), () -> locNode.id().toString())
@@ -2493,8 +2492,6 @@ else if (msg instanceof TcpDiscoveryNodeAddFinishedMessage) {
                 if (addFinishMsg.clientDiscoData() != null) {
                     addFinishMsg = new TcpDiscoveryNodeAddFinishedMessage(addFinishMsg);
 
-                    addFinishMsg.prepareMarshal(spi.marshaller());
-
                     msg = addFinishMsg;
 
                     DiscoveryDataPacket discoData = addFinishMsg.clientDiscoData();
@@ -3308,9 +3305,6 @@ private void sendMessageAcrossRing(TcpDiscoveryAbstractMessage msg) {
             if (msg instanceof TraceableMessage)
                 tracing.messages().beforeSend((TraceableMessage)msg);
 
-            if (msg instanceof TcpDiscoveryJoinRequestMessage)
-                ((TcpDiscoveryJoinRequestMessage)msg).prepareMarshal(spi.marshaller());
-
             sendMessageToClients(msg);
 
             List<TcpDiscoveryNode> failedNodes;
@@ -4144,7 +4138,7 @@ private void processJoinRequestMessage(final TcpDiscoveryJoinRequestMessage msg)
                         Collection<TcpDiscoveryAbstractMessage> msgs = msgHist.messages(null, node);
 
                         if (msgs != null) {
-                            reconMsg.pendingMessages(msgs);
+                            reconMsg.pendingMsgsMsg = new TcpDiscoveryCollectionMessage(msgs);
 
                             reconMsg.success(true);
                         }
@@ -4869,8 +4863,6 @@ private void processNodeAddedMessage(TcpDiscoveryNodeAddedMessage msg) {
                         addFinishMsg.clientDiscoData(msg.gridDiscoveryData());
 
                         addFinishMsg.clientNodeAttributes(node.attributes());
-
-                        addFinishMsg.prepareMarshal(spi.marshaller());
                     }
 
                     addFinishMsg = tracing.messages().branch(addFinishMsg, msg);
@@ -5086,7 +5078,9 @@ else if (spiState == CONNECTING)
                             joiningNodesDiscoDataList = new ArrayList<>();
 
                             topHist.clear();
-                            topHist.putAll(msg.topologyHistory());
+
+                            if (!F.isEmpty(msg.topologyHistory()))
+                                topHist.putAll(msg.topologyHistory());
 
                             pendingMsgs.reset(msg.messages());
                         }
@@ -6966,8 +6960,6 @@ else if (e.hasCause(ObjectStreamException.class) ||
                         else if (msg instanceof TcpDiscoveryJoinRequestMessage) {
                             TcpDiscoveryJoinRequestMessage req = (TcpDiscoveryJoinRequestMessage)msg;
 
-                            req.finishUnmarshal(spi.marshaller(), U.resolveClassLoader(spi.ignite().configuration()));
-
                             // Current node holds connection with the node that is joining the cluster. Therefore, it can
                             // save certificates with which the connection was established to joining node attributes.
                             if (spi.nodeAuth != null && nodeId.equals(req.node().id()))
@@ -7379,7 +7371,7 @@ private void processClientReconnectMessage(TcpDiscoveryClientReconnectMessage ms
 
                         if (pending != null) {
                             msg.verifierNodeId(locNodeId);
-                            msg.pendingMessages(pending);
+                            msg.pendingMsgsMsg = new TcpDiscoveryCollectionMessage(pending);
                             msg.success(true);
 
                             if (log.isDebugEnabled()) {

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

@@ -79,7 +79,7 @@ public class TcpDiscoveryIoSession {
     private final Socket sock;
 
     /** */
-    final DirectMessageWriter msgWriter;
+    private final DirectMessageWriter msgWriter;
 
     /** */
     private final DirectMessageReader msgReader;
@@ -91,7 +91,7 @@ public class TcpDiscoveryIoSession {
     private final CompositeInputStream in;
 
     /** Intermediate buffer for serializing discovery messages. */
-    final ByteBuffer msgBuf;
+    private final ByteBuffer msgBuf;
 
     /**
      * Creates a new discovery I/O session bound to the given socket.

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

@@ -2119,7 +2119,7 @@ protected void onExchange(DiscoveryDataPacket dataPacket, ClassLoader clsLdr) {
         registerMBean(igniteInstanceName, new TcpDiscoverySpiMBeanImpl(this), TcpDiscoverySpiMBean.class);
 
         msgFactory = new IgniteMessageFactoryImpl(
-            new MessageFactoryProvider[] { new DiscoveryMessageFactory() });
+            new MessageFactoryProvider[] { new DiscoveryMessageFactory(marshaller(), U.resolveClassLoader(ignite().configuration())) });
 
         impl.spiStart(igniteInstanceName);
     }