DiscoveryNode and TransportAddress should preserve host information

In some cases, such as the creation of DiscoveryNode instances for unicast ping requests, the
host information was not being populated properly and instead the address string was being used.
Additionally, when serializing a DiscoveryNode and in turn a transport address, the host was not
being set on the InetAddress when deserializing the object, so even if the address was created
from a hostname, the address in the deserialized instance had no knowledge of the hostname that
was originally used.

core/src/main/java/org/elasticsearch/Version.java

@@ -97,6 +97,8 @@ public class Version {
     public static final Version V_5_0_1 = new Version(V_5_0_1_ID, org.apache.lucene.util.Version.LUCENE_6_2_1);
     public static final int V_5_0_2_ID_UNRELEASED = 5000299;
     public static final Version V_5_0_2_UNRELEASED = new Version(V_5_0_2_ID_UNRELEASED, org.apache.lucene.util.Version.LUCENE_6_2_1);
+    public static final int V_5_0_3_ID_UNRELEASED = 5000399;
+    public static final Version V_5_0_3_UNRELEASED = new Version(V_5_0_3_ID_UNRELEASED, org.apache.lucene.util.Version.LUCENE_6_2_1);
     public static final int V_5_1_0_ID_UNRELEASED = 5010099;
     public static final Version V_5_1_0_UNRELEASED = new Version(V_5_1_0_ID_UNRELEASED, org.apache.lucene.util.Version.LUCENE_6_3_0);
     public static final int V_5_2_0_ID_UNRELEASED = 5020099;
@@ -125,6 +127,8 @@ public class Version {
                 return V_5_2_0_UNRELEASED;
             case V_5_1_0_ID_UNRELEASED:
                 return V_5_1_0_UNRELEASED;
+            case V_5_0_3_ID_UNRELEASED:
+                return V_5_0_3_UNRELEASED;
             case V_5_0_2_ID_UNRELEASED:
                 return V_5_0_2_UNRELEASED;
             case V_5_0_1_ID:

core/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java

@@ -135,7 +135,8 @@ public class DiscoveryNode implements Writeable, ToXContent {
      */
     public DiscoveryNode(String nodeName, String nodeId, TransportAddress address,
                          Map<String, String> attributes, Set<Role> roles, Version version) {
-        this(nodeName, nodeId, UUIDs.randomBase64UUID(), address.getAddress(), address.getAddress(), address, attributes, roles, version);
+        this(nodeName, nodeId, UUIDs.randomBase64UUID(), address.address().getHostString(), address.getAddress(), address, attributes,
+            roles, version);
     }
 
     /**

core/src/main/java/org/elasticsearch/common/transport/TransportAddress.java

@@ -78,7 +78,13 @@ public final class TransportAddress implements Writeable {
         final int len = in.readByte();
         final byte[] a = new byte[len]; // 4 bytes (IPv4) or 16 bytes (IPv6)
         in.readFully(a);
-        InetAddress inetAddress = InetAddress.getByAddress(a);
+        final InetAddress inetAddress;
+        if (in.getVersion().onOrAfter(Version.V_5_0_3_UNRELEASED)) {
+            String host = in.readString();
+            inetAddress = InetAddress.getByAddress(host, a);
+        } else {
+            inetAddress = InetAddress.getByAddress(a);
+        }
         int port = in.readInt();
         this.address = new InetSocketAddress(inetAddress, port);
     }
@@ -91,6 +97,9 @@ public final class TransportAddress implements Writeable {
         byte[] bytes = address.getAddress().getAddress();  // 4 bytes (IPv4) or 16 bytes (IPv6)
         out.writeByte((byte) bytes.length); // 1 byte
         out.write(bytes, 0, bytes.length);
+        if (out.getVersion().onOrAfter(Version.V_5_0_3_UNRELEASED)) {
+            out.writeString(address.getHostString());
+        }
         // don't serialize scope ids over the network!!!!
         // these only make sense with respect to the local machine, and will only formulate
         // the address incorrectly remotely.

core/src/test/java/org/elasticsearch/cluster/node/DiscoveryNodeTests.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.cluster.node;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.test.ESTestCase;
+
+import java.net.InetAddress;
+
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+
+public class DiscoveryNodeTests extends ESTestCase {
+
+    public void testDiscoveryNodeIsCreatedWithHostFromInetAddress() throws Exception {
+        InetAddress inetAddress = randomBoolean() ? InetAddress.getByName("192.0.2.1") :
+            InetAddress.getByAddress("name1", new byte[] { (byte) 192, (byte) 168, (byte) 0, (byte) 1});
+        TransportAddress transportAddress = new TransportAddress(inetAddress, randomIntBetween(0, 65535));
+        DiscoveryNode node = new DiscoveryNode("name1", "id1", transportAddress, emptyMap(), emptySet(), Version.CURRENT);
+        assertEquals(transportAddress.address().getHostString(), node.getHostName());
+        assertEquals(transportAddress.getAddress(), node.getHostAddress());
+    }
+
+    public void testDiscoveryNodeSerializationKeepsHost() throws Exception {
+        InetAddress inetAddress = InetAddress.getByAddress("name1", new byte[] { (byte) 192, (byte) 168, (byte) 0, (byte) 1});
+        TransportAddress transportAddress = new TransportAddress(inetAddress, randomIntBetween(0, 65535));
+        DiscoveryNode node = new DiscoveryNode("name1", "id1", transportAddress, emptyMap(), emptySet(), Version.CURRENT);
+
+        BytesStreamOutput streamOutput = new BytesStreamOutput();
+        streamOutput.setVersion(Version.CURRENT);
+        node.writeTo(streamOutput);
+
+        StreamInput in = StreamInput.wrap(streamOutput.bytes().toBytesRef().bytes);
+        DiscoveryNode serialized = new DiscoveryNode(in);
+        assertEquals(transportAddress.address().getHostString(), serialized.getHostName());
+        assertEquals(transportAddress.address().getHostString(), serialized.getAddress().address().getHostString());
+        assertEquals(transportAddress.getAddress(), serialized.getHostAddress());
+        assertEquals(transportAddress.getAddress(), serialized.getAddress().getAddress());
+        assertEquals(transportAddress.getPort(), serialized.getAddress().getPort());
+    }
+
+    public void testDiscoveryNodeSerializationToOldVersion() throws Exception {
+        InetAddress inetAddress = InetAddress.getByAddress("name1", new byte[] { (byte) 192, (byte) 168, (byte) 0, (byte) 1});
+        TransportAddress transportAddress = new TransportAddress(inetAddress, randomIntBetween(0, 65535));
+        DiscoveryNode node = new DiscoveryNode("name1", "id1", transportAddress, emptyMap(), emptySet(), Version.CURRENT);
+
+        BytesStreamOutput streamOutput = new BytesStreamOutput();
+        streamOutput.setVersion(Version.V_5_0_0);
+        node.writeTo(streamOutput);
+
+        StreamInput in = StreamInput.wrap(streamOutput.bytes().toBytesRef().bytes);
+        in.setVersion(Version.V_5_0_0);
+        DiscoveryNode serialized = new DiscoveryNode(in);
+        assertEquals(transportAddress.address().getHostString(), serialized.getHostName());
+        assertNotEquals(transportAddress.address().getHostString(), serialized.getAddress().address().getHostString());
+        assertEquals(transportAddress.getAddress(), serialized.getHostAddress());
+        assertEquals(transportAddress.getAddress(), serialized.getAddress().getAddress());
+        assertEquals(transportAddress.getPort(), serialized.getAddress().getPort());
+    }
+}