IGNITE-28440 Use message serializer for test discovery messages (#12974)

Author: NSAmelchev

modules/core/src/main/java/org/apache/ignite/plugin/extensions/communication/Message.java

@@ -58,11 +58,6 @@ default boolean readFrom(ByteBuffer buf, MessageReader reader) {
         throw new UnsupportedOperationException();
     }
 
-    /**
-     * Gets message type.
-     *
-     * @return Message type.
-     */
     /**
      * Gets message type.
      *
@@ -72,9 +67,8 @@ default short directType() {
         var clazz = getClass();
         Short type = REGISTRATIONS.get(clazz);
 
-        if (type == null) {
+        if (type == null)
             throw new IgniteException("No registration for class " + clazz.getSimpleName());
-        }
 
         return type;
     }
@@ -91,8 +85,7 @@ default void registerAsDirectType(short directType) {
         var clazz = getClass();
         var type = REGISTRATIONS.putIfAbsent(clazz, directType);
 
-        if ((type != null) && (type != directType)) {
+        if ((type != null) && (type != directType))
             throw new IgniteException(clazz.getSimpleName() + " is already registered for direct type " + type);
-        }
     }
 }

modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/messages/HandshakeWaitMessage.java

@@ -31,5 +31,4 @@ public class HandshakeWaitMessage implements Message {
     @Override public String toString() {
         return S.toString(HandshakeWaitMessage.class, this);
     }
-
 }

modules/core/src/test/java/org/apache/ignite/internal/metric/OutboundIoMessageQueueSizeTest.java

@@ -23,18 +23,17 @@
 import org.apache.ignite.cache.query.ContinuousQuery;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric;
 import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.MessagesPluginProvider;
 import org.apache.ignite.spi.discovery.tcp.BlockTcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.DummyCustomDiscoveryMessage;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.apache.ignite.testframework.ListeningTestLogger;
 import org.apache.ignite.testframework.LogListener;
 import org.apache.ignite.testframework.junits.WithSystemProperty;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.jetbrains.annotations.Nullable;
 import org.junit.Test;
 
 import static org.apache.ignite.internal.managers.discovery.GridDiscoveryManager.DISCO_METRICS;
@@ -58,6 +57,8 @@ public class OutboundIoMessageQueueSizeTest extends GridCommonAbstractTest {
         cfg.setDiscoverySpi(new BlockTcpDiscoverySpi().setIpFinder(((TcpDiscoverySpi)cfg.getDiscoverySpi()).getIpFinder()));
         cfg.setGridLogger(log);
 
+        cfg.setPluginProviders(new MessagesPluginProvider(DummyCustomDiscoveryMessage.class));
+
         return cfg;
     }
 
@@ -124,7 +125,7 @@ public void testDiscoveryMsgQueue() throws Exception {
 
         metric.reset(); // Reset value accumulated before discovery SPI startup.
 
-        srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage(IgniteUuid.randomUuid()));
+        srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage());
 
         // Assume our message can be added to queue concurrently with other messages
         // (for example, with metrics update message).
@@ -142,35 +143,12 @@ public void testDiscoveryMsgQueue() throws Exception {
 
         try {
             for (int i = 0; i <= MSG_LIMIT; i++)
-                srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage(IgniteUuid.randomUuid()));
+                srv0.context().discovery().sendCustomEvent(new DummyCustomDiscoveryMessage());
 
             assertTrue(metric.value() >= MSG_LIMIT);
         }
         finally {
             latch.countDown();
         }
     }
-
-    /** */
-    private static class DummyCustomDiscoveryMessage implements DiscoveryCustomMessage {
-        /** */
-        private final IgniteUuid id;
-
-        /**
-         * @param id Message id.
-         */
-        DummyCustomDiscoveryMessage(IgniteUuid id) {
-            this.id = id;
-        }
-
-        /** {@inheritDoc} */
-        @Override public IgniteUuid id() {
-            return id;
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public DiscoveryCustomMessage ackMessage() {
-            return null;
-        }
-    }
 }

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java

@@ -99,6 +99,7 @@
 import org.apache.ignite.lang.IgnitePredicate;
 import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
 import org.apache.ignite.spi.encryption.keystore.KeystoreEncryptionSpi;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -911,7 +912,7 @@ protected static BlockingCustomMessageDiscoverySpi discoSpi(IgniteEx ignite) {
     }
 
     /** */
-    protected static class BlockingCustomMessageDiscoverySpi extends TcpDiscoverySpi {
+    protected static class BlockingCustomMessageDiscoverySpi extends TestTcpDiscoverySpi {
         /** List of messages which have been blocked. */
         private final List<DiscoveryCustomMessage> blocked = new CopyOnWriteArrayList<>();
 

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotHandlerTest.java

@@ -36,8 +36,7 @@
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
 import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
-import org.apache.ignite.internal.util.distributed.MessagesPluginProvider;
-import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
+import org.apache.ignite.internal.util.distributed.TestIntegerMessage;
 import org.apache.ignite.internal.util.distributed.TestUuidMessage;
 import org.apache.ignite.internal.util.typedef.G;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -48,6 +47,7 @@
 import org.apache.ignite.plugin.PluginContext;
 import org.apache.ignite.plugin.PluginProvider;
 import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.MessagesPluginProvider;
 import org.apache.ignite.testframework.GridTestUtils;
 import org.jetbrains.annotations.Nullable;
 import org.junit.Test;
@@ -76,8 +76,10 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
     /** {@inheritDoc} */
     @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
         return super.getConfiguration(igniteInstanceName)
-            .setPluginProviders(pluginProvider, new MessagesPluginProvider())
-            .setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
+            .setPluginProviders(
+                pluginProvider,
+                new MessagesPluginProvider(TestIntegerMessage.class, TestUuidMessage.class)
+            );
     }
 
     /** {@inheritDoc} */

modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java

@@ -34,10 +34,9 @@
 import org.apache.ignite.failure.StopNodeOrHaltFailureHandler;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.events.DiscoveryCustomEvent;
-import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
 import org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
 import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.MessagesPluginProvider;
 import org.apache.ignite.spi.discovery.DiscoverySpi;
 import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
@@ -87,6 +86,8 @@ public class NodeSecurityContextPropagationTest extends GridCommonAbstractTest {
             .setIpFinder(new TcpDiscoveryVmIpFinder()
                 .setAddresses(Collections.singleton("127.0.0.1:47500")));
 
+        cfg.setPluginProviders(new MessagesPluginProvider(TestDiscoveryMessage.class, TestDiscoveryAcknowledgeMessage.class));
+
         return cfg;
     }
 
@@ -218,30 +219,6 @@ private Object discoveryRingMessageWorker(IgniteEx ignite) {
         return U.field(impl, "msgWorker");
     }
 
-    /** */
-    public static class TestDiscoveryMessage extends AbstractTestDiscoveryMessage {
-        /** {@inheritDoc} */
-        @Override public @Nullable DiscoveryCustomMessage ackMessage() {
-            return new TestDiscoveryAcknowledgeMessage();
-        }
-    }
-
-    /** */
-    public static class TestDiscoveryAcknowledgeMessage extends AbstractTestDiscoveryMessage { }
-
-    /** */
-    public abstract static class AbstractTestDiscoveryMessage implements DiscoveryCustomMessage {
-        /** {@inheritDoc} */
-        @Override public IgniteUuid id() {
-            return IgniteUuid.randomUuid();
-        }
-
-        /** {@inheritDoc} */
-        @Override public @Nullable DiscoveryCustomMessage ackMessage() {
-            return null;
-        }
-    }
-
     /** */
     public static class BlockingDequeWrapper<T> implements BlockingDeque<T> {
         /** */

modules/core/src/test/java/org/apache/ignite/internal/processors/security/TestDiscoveryAcknowledgeMessage.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.security;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class TestDiscoveryAcknowledgeMessage implements DiscoveryCustomMessage, Message {
+    /** */
+    @Order(0)
+    IgniteUuid id = IgniteUuid.randomUuid();
+
+    /** Constructor for {@link MessageFactory}. */
+    public TestDiscoveryAcknowledgeMessage() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteUuid id() {
+        return id;
+    }
+
+    /** {@inheritDoc} */
+    @Override public @Nullable DiscoveryCustomMessage ackMessage() {
+        return null;
+    }
+}

modules/core/src/test/java/org/apache/ignite/internal/processors/security/TestDiscoveryMessage.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.security;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class TestDiscoveryMessage implements DiscoveryCustomMessage, Message {
+    /** */
+    @Order(0)
+    IgniteUuid id = IgniteUuid.randomUuid();
+
+    /** Constructor for {@link MessageFactory}. */
+    public TestDiscoveryMessage() {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteUuid id() {
+        return id;
+    }
+
+    /** {@inheritDoc} */
+    @Override public @Nullable DiscoveryCustomMessage ackMessage() {
+        return new TestDiscoveryAcknowledgeMessage();
+    }
+}

modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessClientAwaitTest.java

@@ -28,10 +28,10 @@
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.TestRecordingCommunicationSpi;
-import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.MessagesPluginProvider;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 
@@ -70,9 +70,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
         IgniteConfiguration cfg = super.getConfiguration(instanceName);
 
         cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
-
-        cfg.setPluginProviders(new MessagesPluginProvider());
-        cfg.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
+        cfg.setPluginProviders(new MessagesPluginProvider(TestIntegerMessage.class, TestUuidMessage.class));
 
         return cfg;
     }

modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessCoordinatorLeftTest.java

@@ -28,8 +28,8 @@
 import org.apache.ignite.failure.FailureHandler;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
 import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.spi.MessagesPluginProvider;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 
@@ -86,8 +86,7 @@ public class DistributedProcessCoordinatorLeftTest extends GridCommonAbstractTes
             }
         });
 
-        cfg.setPluginProviders(new MessagesPluginProvider());
-        cfg.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
+        cfg.setPluginProviders(new MessagesPluginProvider(TestIntegerMessage.class, TestUuidMessage.class));
 
         return cfg;
     }