IGNITE-27415 Use MessageSerializer for TcpDiscoveryAuthFailedMessage (#12605)

Author: NSAmelchev

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

@@ -18,6 +18,8 @@
 package org.apache.ignite.internal.managers.discovery;
 
 import org.apache.ignite.internal.codegen.InetAddressMessageSerializer;
+import org.apache.ignite.internal.codegen.InetSocketAddressMessageSerializer;
+import org.apache.ignite.internal.codegen.TcpDiscoveryAuthFailedMessageSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryCheckFailedMessageSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryClientPingRequestSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryClientPingResponseSerializer;
@@ -32,6 +34,8 @@
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 import org.apache.ignite.spi.discovery.tcp.messages.InetAddressMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.InetSocketAddressMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAuthFailedMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCheckFailedMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingRequest;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingResponse;
@@ -48,6 +52,7 @@
 public class DiscoveryMessageFactory implements MessageFactoryProvider {
     /** {@inheritDoc} */
     @Override public void registerAll(MessageFactory factory) {
+        factory.register((short)-101, InetSocketAddressMessage::new, new InetSocketAddressMessageSerializer());
         factory.register((short)-100, InetAddressMessage::new, new InetAddressMessageSerializer());
 
         factory.register((short)0, TcpDiscoveryCheckFailedMessage::new, new TcpDiscoveryCheckFailedMessageSerializer());
@@ -61,5 +66,6 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
         factory.register((short)8, TcpDiscoveryHandshakeRequest::new, new TcpDiscoveryHandshakeRequestSerializer());
         factory.register((short)9, TcpDiscoveryDiscardMessage::new, new TcpDiscoveryDiscardMessageSerializer());
         factory.register((short)10, TcpDiscoveryHandshakeResponse::new, new TcpDiscoveryHandshakeResponseSerializer());
+        factory.register((short)11, TcpDiscoveryAuthFailedMessage::new, new TcpDiscoveryAuthFailedMessageSerializer());
     }
 }

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

@@ -3340,7 +3340,7 @@ else if (msg instanceof TcpDiscoveryAuthFailedMessage)
          */
         private void processAuthFailedMessage(TcpDiscoveryAuthFailedMessage authFailedMsg) {
             try {
-                sendDirectlyToClient(authFailedMsg.getTargetNodeId(), authFailedMsg);
+                sendDirectlyToClient(authFailedMsg.targetNodeId(), authFailedMsg);
             }
             catch (IgniteSpiException ex) {
                 log.warning(
@@ -7174,7 +7174,7 @@ else if (msg instanceof TcpDiscoveryAuthFailedMessage) {
                                     mux.notifyAll();
                                 }
                                 else {
-                                    UUID targetNode = ((TcpDiscoveryAuthFailedMessage)msg).getTargetNodeId();
+                                    UUID targetNode = ((TcpDiscoveryAuthFailedMessage)msg).targetNodeId();
 
                                     if (targetNode == null || targetNode.equals(locNodeId)) {
                                         if (log.isDebugEnabled())

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

@@ -2019,7 +2019,7 @@ protected IgniteSpiException authenticationFailedError(TcpDiscoveryAuthFailedMes
         assert msg != null;
 
         return new IgniteSpiException(new IgniteAuthenticationException("Authentication failed [nodeId=" +
-            msg.creatorNodeId() + ", addr=" + msg.address().getHostAddress() + ']'));
+            msg.creatorNodeId() + ", addr=" + msg.creatorAddress().getHostAddress() + ']'));
     }
 
     /**

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

@@ -19,6 +19,7 @@
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.util.typedef.internal.S;
@@ -31,29 +32,17 @@ public class InetAddressMessage implements Message {
     private String hostName;
 
     /** */
-    @Order(1)
-    private int port;
-
-    /** */
-    @Order(value = 2, method = "addressBytes")
+    @Order(value = 1, method = "addressBytes")
     private byte[] addrBytes;
 
-    /**
-     * Default constructor for {@link DiscoveryMessageFactory}.
-     */
+    /** Default constructor for {@link DiscoveryMessageFactory}. */
     public InetAddressMessage() {
         // No-op.
     }
 
-    /**
-     * Constructor.
-     *
-     * @param addr Address.
-     * @param port Port.
-     */
-    public InetAddressMessage(InetAddress addr, int port) {
+    /** @param addr Address. */
+    public InetAddressMessage(InetAddress addr) {
         hostName = addr.getHostName();
-        this.port = port;
         addrBytes = addr.getAddress();
     }
 
@@ -67,14 +56,14 @@ public void addressBytes(byte[] addrBytes) {
         this.addrBytes = addrBytes;
     }
 
-    /** @return port. */
-    public int port() {
-        return port;
-    }
-
-    /** @param port port. */
-    public void port(int port) {
-        this.port = port;
+    /** @return {@link InetAddress#getByAddress(String, byte[])} */
+    public InetAddress address() {
+        try {
+            return addrBytes == null ? null : InetAddress.getByAddress(hostName, addrBytes);
+        }
+        catch (UnknownHostException e) {
+            throw new IgniteException("Failed to read host address.", e);
+        }
     }
 
     /** @return Host name. */
@@ -87,11 +76,6 @@ public void hostName(String hostName) {
         this.hostName = hostName;
     }
 
-    /** @return {@link InetAddress#getByAddress(String, byte[])} */
-    public InetAddress address() throws UnknownHostException {
-        return addrBytes == null ? null : InetAddress.getByAddress(hostName, addrBytes);
-    }
-
     /** {@inheritDoc} */
     @Override public short directType() {
         return -100;

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

@@ -0,0 +1,69 @@
+/*
+ * 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.spi.discovery.tcp.messages;
+
+import java.net.InetAddress;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/** Socket address utility container message. Is not a pure {@link TcpDiscoveryAbstractMessage}. */
+public class InetSocketAddressMessage extends InetAddressMessage {
+    /** */
+    @Order(2)
+    private int port;
+
+    /**
+     * Default constructor for {@link DiscoveryMessageFactory}.
+     */
+    public InetSocketAddressMessage() {
+        // No-op.
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param addr Address.
+     * @param port Port.
+     */
+    public InetSocketAddressMessage(InetAddress addr, int port) {
+        super(addr);
+
+        this.port = port;
+    }
+
+    /** @return Port. */
+    public int port() {
+        return port;
+    }
+
+    /** @param port Port. */
+    public void port(int port) {
+        this.port = port;
+    }
+
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return -101;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(InetSocketAddressMessage.class, this);
+    }
+}

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

@@ -17,71 +17,75 @@
 
 package org.apache.ignite.spi.discovery.tcp.messages;
 
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
 import java.net.InetAddress;
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
 /**
- * Message telling joining node that its authentication failed on coordinator.
+ * Message telling joining node that its authentication failed.
  */
-public class TcpDiscoveryAuthFailedMessage extends TcpDiscoveryAbstractMessage {
+public class TcpDiscoveryAuthFailedMessage extends TcpDiscoveryAbstractMessage implements Message {
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Coordinator address. */
-    private transient InetAddress addr;
+    /** Creator address. */
+    @Order(value = 5, method = "creatorAddressMessage")
+    private InetAddressMessage creatorAddrMsg;
 
     /** Node id for which authentication was failed. */
+    @Order(6)
     private UUID targetNodeId;
 
+    /** Default constructor for {@link DiscoveryMessageFactory}. */
+    public TcpDiscoveryAuthFailedMessage() {
+        // No-op.
+    }
+
     /**
      * Constructor.
      *
      * @param creatorNodeId Creator node ID.
-     * @param addr Coordinator address.
+     * @param creatorAddr Creator address.
      * @param targetNodeId Node for which authentication was failed.
      */
-    public TcpDiscoveryAuthFailedMessage(UUID creatorNodeId, InetAddress addr, UUID targetNodeId) {
+    public TcpDiscoveryAuthFailedMessage(UUID creatorNodeId, InetAddress creatorAddr, UUID targetNodeId) {
         super(creatorNodeId);
 
-        this.addr = addr;
+        this.creatorAddrMsg = new InetAddressMessage(creatorAddr);
         this.targetNodeId = targetNodeId;
     }
 
-    /**
-     * @return Node for which authentication was failed.
-     */
-    public UUID getTargetNodeId() {
+    /** @return Node for which authentication was failed. */
+    public UUID targetNodeId() {
         return targetNodeId;
     }
 
-    /**
-     * @return Coordinator address.
-     */
-    public InetAddress address() {
-        return addr;
+    /** @param targetNodeId Node for which authentication was failed. */
+    public void targetNodeId(UUID targetNodeId) {
+        this.targetNodeId = targetNodeId;
     }
 
-    /**
-     * Serialize this message.
-     */
-    private void writeObject(ObjectOutputStream out) throws IOException {
-        out.defaultWriteObject();
+    /** @return Creator address message. */
+    public InetAddressMessage creatorAddressMessage() {
+        return creatorAddrMsg;
+    }
 
-        U.writeByteArray(out, addr.getAddress());
+    /** @param addr Creator address message. */
+    public void creatorAddressMessage(InetAddressMessage addr) {
+        this.creatorAddrMsg = addr;
     }
 
-    /**
-     * Deserialize this message.
-     */
-    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
-        in.defaultReadObject();
+    /** @return Creator address. */
+    public InetAddress creatorAddress() {
+        return creatorAddrMsg.address();
+    }
 
-        addr = InetAddress.getByAddress(U.readByteArray(in));
+    /** {@inheritDoc} */
+    @Override public short directType() {
+        return 11;
     }
 
     /** {@inheritDoc} */

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

@@ -18,10 +18,8 @@
 package org.apache.ignite.spi.discovery.tcp.messages;
 
 import java.net.InetSocketAddress;
-import java.net.UnknownHostException;
 import java.util.Collection;
 import java.util.UUID;
-import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.util.typedef.F;
@@ -46,7 +44,7 @@ public class TcpDiscoveryHandshakeResponse extends TcpDiscoveryAbstractMessage i
 
     /** Redirect addresses messages serialization holder. */
     @Order(value = 7, method = "redirectAddressesMessages")
-    private @Nullable Collection<InetAddressMessage> redirectAddrsMsgs;
+    private @Nullable Collection<InetSocketAddressMessage> redirectAddrsMsgs;
 
     /**
      * Default constructor for {@link DiscoveryMessageFactory}.
@@ -109,30 +107,23 @@ public void order(long order) {
     public @Nullable Collection<InetSocketAddress> redirectAddresses() {
         return F.isEmpty(redirectAddrsMsgs)
             ? null
-            : F.transform(redirectAddrsMsgs, msg -> {
-                try {
-                    return new InetSocketAddress(msg.address(), msg.port());
-                }
-                catch (UnknownHostException e) {
-                    throw new IgniteException("Failed to read host address.", e);
-                }
-            });
+            : F.transform(redirectAddrsMsgs, msg -> new InetSocketAddress(msg.address(), msg.port()));
     }
 
     /** @param sockAddrs Socket addresses list for redirect. */
     public void redirectAddresses(@Nullable Collection<InetSocketAddress> sockAddrs) {
         redirectAddrsMsgs = sockAddrs == null
             ? null
-            : F.viewReadOnly(sockAddrs, addr -> new InetAddressMessage(addr.getAddress(), addr.getPort()));
+            : F.viewReadOnly(sockAddrs, addr -> new InetSocketAddressMessage(addr.getAddress(), addr.getPort()));
     }
 
     /** @return Collection of {@link InetAddressMessage}. */
-    public @Nullable Collection<InetAddressMessage> redirectAddressesMessages() {
+    public @Nullable Collection<InetSocketAddressMessage> redirectAddressesMessages() {
         return redirectAddrsMsgs;
     }
 
     /** @param redirectAddrsMsgs Collection of {@link InetAddressMessage}. */
-    public void redirectAddressesMessages(@Nullable Collection<InetAddressMessage> redirectAddrsMsgs) {
+    public void redirectAddressesMessages(@Nullable Collection<InetSocketAddressMessage> redirectAddrsMsgs) {
         this.redirectAddrsMsgs = redirectAddrsMsgs;
     }