IGNITE-27899 : Make TcpDiscoveryNode implement message v2

modules/codegen/src/main/java/org/apache/ignite/internal/MessageSerializerGenerator.java

@@ -724,7 +724,7 @@ private String messageCollectionItemTypes(VariableElement field, TypeMirror type
         String descName = field.getSimpleName() + "CollDesc";
         String typeName = desc.substring(desc.indexOf(' ') + 1, desc.indexOf('('));
 
-        fields.add("private final static " + typeName + " " + descName + " = " + desc + ";");
+        fields.add("private static final " + typeName + " " + descName + " = " + desc + ";");
 
         return descName;
     }

modules/core/src/main/java/org/apache/ignite/internal/ClusterMetricsSnapshot.java

@@ -109,6 +109,9 @@ public ClusterMetricsSnapshot() {
      * Creates snapshot based on the handled message.
      */
     public ClusterMetricsSnapshot(NodeMetricsMessage m) {
+        // As in #deserialize().
+        m.lastUpdateTime = U.currentTimeMillis();
+
         this.m = m;
     }
 

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

@@ -129,13 +129,17 @@
 import org.apache.ignite.internal.util.distributed.FullMessageSerializer;
 import org.apache.ignite.internal.util.distributed.InitMessage;
 import org.apache.ignite.internal.util.distributed.InitMessageSerializer;
+import org.apache.ignite.lang.IgniteProductVersion;
+import org.apache.ignite.lang.IgniteProductVersionSerializer;
 import org.apache.ignite.marshaller.Marshaller;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 import org.apache.ignite.spi.communication.tcp.internal.TcpConnectionRequestDiscoveryMessage;
 import org.apache.ignite.spi.communication.tcp.internal.TcpConnectionRequestDiscoveryMessageSerializer;
 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.internal.TcpDiscoveryNode;
+import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNodeMarshallableSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.InetAddressMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.InetAddressMessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.InetSocketAddressMessage;
@@ -223,6 +227,8 @@ public DiscoveryMessageFactory(Marshaller marsh, ClassLoader clsLdr) {
         factory.register(-200, TcpDiscoveryCollectionMessage::new,
             new TcpDiscoveryCollectionMessageMarshallableSerializer(marsh, clsLdr));
 
+        factory.register(-117, TcpDiscoveryNode::new, new TcpDiscoveryNodeMarshallableSerializer(marsh, clsLdr));
+        factory.register(-116, IgniteProductVersion::new, new IgniteProductVersionSerializer());
         factory.register(-115, SchemaAlterTableAddColumnOperation::new, new SchemaAlterTableAddColumnOperationSerializer());
         factory.register(-114, SchemaIndexCreateOperation::new, new SchemaIndexCreateOperationMarshallableSerializer(marsh, clsLdr));
         factory.register(-113, SchemaIndexDropOperation::new, new SchemaIndexDropOperationSerializer());

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

@@ -363,7 +363,7 @@ public void cleanCachesAndGroups() {
      */
     public void addCacheGroup(CacheGroupDescriptor grpDesc, IgnitePredicate<ClusterNode> filter, CacheMode cacheMode) {
         CacheGroupAffinity old = registeredCacheGrps.put(grpDesc.groupId(),
-            new CacheGroupAffinity(grpDesc.cacheOrGroupName(), filter, cacheMode, grpDesc.persistenceEnabled()));
+            new CacheGroupAffinity(filter, cacheMode, grpDesc.persistenceEnabled()));
 
         assert old == null : old;
     }
@@ -3389,9 +3389,6 @@ private Snapshot(AffinityTopologyVersion topVer, DiscoCache discoCache) {
      *
      */
     private static class CacheGroupAffinity {
-        /** */
-        private final String name;
-
         /** Nodes filter. */
         private final IgnitePredicate<ClusterNode> cacheFilter;
 
@@ -3402,17 +3399,11 @@ private static class CacheGroupAffinity {
         private final boolean persistentCacheGrp;
 
         /**
-         * @param name Name.
          * @param cacheFilter Node filter.
          * @param cacheMode Cache mode.
          * @param persistentCacheGrp Persistence is configured for cache or not.
          */
-        CacheGroupAffinity(
-                String name,
-                IgnitePredicate<ClusterNode> cacheFilter,
-                CacheMode cacheMode,
-                boolean persistentCacheGrp) {
-            this.name = name;
+        CacheGroupAffinity(IgnitePredicate<ClusterNode> cacheFilter, CacheMode cacheMode, boolean persistentCacheGrp) {
             this.cacheFilter = cacheFilter;
             this.cacheMode = cacheMode;
             this.persistentCacheGrp = persistentCacheGrp;

modules/core/src/main/java/org/apache/ignite/lang/IgniteProductVersion.java

@@ -26,7 +26,10 @@
 import java.util.regex.Pattern;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.IgniteVersionUtils;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.jetbrains.annotations.NotNull;
 
 /**
@@ -37,7 +40,7 @@
  * Two versions are compared in the following order: major number,
  * minor number, maintenance number, revision timestamp.
  */
-public class IgniteProductVersion implements Comparable<IgniteProductVersion>, Externalizable {
+public class IgniteProductVersion implements Comparable<IgniteProductVersion>, Externalizable, Message {
     /** */
     private static final long serialVersionUID = 0L;
 
@@ -52,25 +55,31 @@ public class IgniteProductVersion implements Comparable<IgniteProductVersion>, E
         Pattern.compile("(\\d+)\\.(\\d+)\\.(\\d+)([-.]([^0123456789][^-]+)(-SNAPSHOT)?)?(-(\\d+))?(-([\\da-f]+))?");
 
     /** Major version number. */
-    private byte major;
+    @Order(0)
+    byte major;
 
     /** Minor version number. */
-    private byte minor;
+    @Order(1)
+    byte minor;
 
     /** Maintenance version number. */
-    private byte maintenance;
+    @Order(2)
+    byte maintenance;
 
     /** Stage of development. */
-    private String stage;
+    @Order(3)
+    String stage;
 
     /** Revision timestamp. */
-    private long revTs;
+    @Order(4)
+    long revTs;
 
     /** Revision hash. */
-    private byte[] revHash;
+    @Order(5)
+    byte[] revHash;
 
     /**
-     * Empty constructor required by {@link Externalizable}.
+     * Empty constructor required by {@link Externalizable} and {@link DiscoveryMessageFactory}.
      */
     public IgniteProductVersion() {
         // No-op.
@@ -265,6 +274,11 @@ public int compareToIgnoreTimestamp(@NotNull IgniteProductVersion o) {
         revHash = U.readByteArray(in);
     }
 
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return -116;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         String revTsStr = IgniteVersionUtils.formatBuildTimeStamp(revTs * 1000);

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

@@ -2240,7 +2240,7 @@ private void processNodeAddedMessage(TcpDiscoveryNodeAddedMessage msg) {
                         nodeAdded = true;
 
                         if (msg.topologyHistory() != null)
-                            topHist.putAll(msg.topologyHistory());
+                            topHist.putAll(upcast(msg.topologyHistory()));
                     }
                     else {
                         if (log.isDebugEnabled())

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

@@ -643,12 +643,11 @@
 
                     processed.add(n);
 
-                    List<ClusterNode> top = U.arrayList(nodes, F.notIn(processed));
+                    Collection<ClusterNode> top = upcast(U.arrayList(nodes, F.notIn(processed)));
 
                     topVer++;
 
-                    NavigableMap<Long, Collection<ClusterNode>> hist = updateTopologyHistory(topVer,
-                        Collections.unmodifiableList(top));
+                    NavigableMap<Long, Collection<ClusterNode>> hist = updateTopologyHistory(topVer, top);
 
                     lsnr.onDiscovery(
                         new DiscoveryNotification(EVT_NODE_FAILED, topVer, n, top, hist, null, null)
@@ -1907,7 +1906,7 @@
                     hist = new TreeMap<>(topHist);
                 }
 
-                nodeAddedMsg.topologyHistory(hist);
+                nodeAddedMsg.topologyHistory(downcast(hist));
             }
         }
     }
@@ -2451,6 +2450,31 @@
         return spi.ignite() instanceof IgniteEx ? ((IgniteEx)spi.ignite()).context().workersRegistry() : null;
     }
 
+    /**
+     * Upcasts collection type.
+     *
+     * @param <P> Parent type.
+     * @param <C> Child type.
+     * @param c Initial collection.
+     * @return Resulting collection.
+     */
+    private static <P, C extends P> Collection<P> upcast(Collection<C> c) {
+        return (Collection<P>)c;
+    }
+
+    /**
+     * Downcasts type of map's collection value.
+     *
+     * @param <K> Map key type.
+     * @param <P> Parent type.
+     * @param <C> Child type.
+     * @param m Initial collections map.
+     * @return Resulting map.
+     */
+    private static <K, P, C extends P> Map<K, Collection<C>> downcast(Map<K, Collection<P>> m) {
+        return (Map<K, Collection<C>>)(Map)m;
+    }
+
     /**
      * Discovery messages history used for client reconnect.
      */
@@ -5086,7 +5110,7 @@
                             joiningNodesDiscoDataList = new ArrayList<>();
 
                             topHist.clear();
-                            topHist.putAll(msg.topologyHistory());
+                            topHist.putAll(upcast(msg.topologyHistory()));
 
                             pendingMsgs.reset(msg.messages());
                         }

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

@@ -43,7 +43,6 @@
 import org.apache.ignite.internal.processors.tracing.NoopTracing;
 import org.apache.ignite.internal.processors.tracing.Tracing;
 import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.LT;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.spi.IgniteSpiContext;
@@ -140,18 +139,6 @@
     /** Tracing. */
     protected Tracing tracing;
 
-    /**
-     * Upcasts collection type.
-     *
-     * @param c Initial collection.
-     * @return Resulting collection.
-     */
-    protected static <T extends R, R> Collection<R> upcast(Collection<T> c) {
-        A.notNull(c, "c");
-
-        return (Collection<R>)c;
-    }
-
     /**
      * @param spi Adapter.
      */
@@ -495,6 +482,19 @@
         return msg.traceLogLevel() ? traceLog : debugLog;
     }
 
+    /**
+     * Upcasts type of map's collection value.
+     *
+     * @param <K> Map key type.
+     * @param <P> Parent type.
+     * @param <C> Child type.
+     * @param m Initial map of collections.
+     * @return Resulting map.
+     */
+    protected static <K, P, C extends P> Map<K, Collection<P>> upcast(Map<K, Collection<C>> m) {
+        return (Map<K, Collection<P>>)(Map)m;
+    }
+
     /**
      *
      */