Support es7 node http publish_address format (#49279)

Add parsing support to node http publish_address format cname/ip:port.

client/sniffer/src/main/java/org/elasticsearch/client/sniff/ElasticsearchNodesSniffer.java

@@ -164,9 +164,21 @@ public final class ElasticsearchNodesSniffer implements NodesSniffer {
                 if ("http".equals(fieldName)) {
                     while (parser.nextToken() != JsonToken.END_OBJECT) {
                         if (parser.getCurrentToken() == JsonToken.VALUE_STRING && "publish_address".equals(parser.getCurrentName())) {
-                            URI publishAddressAsURI = URI.create(scheme + "://" + parser.getValueAsString());
-                            publishedHost = new HttpHost(publishAddressAsURI.getHost(), publishAddressAsURI.getPort(),
-                                    publishAddressAsURI.getScheme());
+                            String address = parser.getValueAsString();
+                            String host;
+                            URI publishAddressAsURI;
+
+                            // ES7 cname/ip:port format
+                            if(address.contains("/")) {
+                                String[] cnameAndURI = address.split("/", 2);
+                                publishAddressAsURI = URI.create(scheme + "://" + cnameAndURI[1]);
+                                host = cnameAndURI[0];
+                            }
+                            else {
+                                publishAddressAsURI = URI.create(scheme + "://" + address);
+                                host = publishAddressAsURI.getHost();
+                            }
+                            publishedHost = new HttpHost(host, publishAddressAsURI.getPort(), publishAddressAsURI.getScheme());
                         } else if (parser.currentToken() == JsonToken.START_ARRAY && "bound_address".equals(parser.getCurrentName())) {
                             while (parser.nextToken() != JsonToken.END_ARRAY) {
                                 URI boundAddressAsURI = URI.create(scheme + "://" + parser.getValueAsString());

client/sniffer/src/test/java/org/elasticsearch/client/sniff/ElasticsearchNodesSnifferParseTests.java

@@ -107,6 +107,28 @@ public class ElasticsearchNodesSnifferParseTests extends RestClientTestCase {
                 node(9207, "c2", "6.0.0", false, false, true));
     }
 
+    public void testParsingPublishAddressWithPreES7Format() throws IOException {
+        InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream("es6_nodes_publication_address_format.json");
+
+        HttpEntity entity = new InputStreamEntity(in, ContentType.APPLICATION_JSON);
+        List<Node> nodes = ElasticsearchNodesSniffer.readHosts(entity, Scheme.HTTP, new JsonFactory());
+
+        assertEquals("127.0.0.1", nodes.get(0).getHost().getHostName());
+        assertEquals(9200, nodes.get(0).getHost().getPort());
+        assertEquals("http", nodes.get(0).getHost().getSchemeName());
+    }
+
+    public void testParsingPublishAddressWithES7Format() throws IOException {
+        InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream("es7_nodes_publication_address_format.json");
+
+        HttpEntity entity = new InputStreamEntity(in, ContentType.APPLICATION_JSON);
+        List<Node> nodes = ElasticsearchNodesSniffer.readHosts(entity, Scheme.HTTP, new JsonFactory());
+
+        assertEquals("elastic.test", nodes.get(0).getHost().getHostName());
+        assertEquals(9200, nodes.get(0).getHost().getPort());
+        assertEquals("http", nodes.get(0).getHost().getSchemeName());
+    }
+
     private Node node(int port, String name, String version, boolean master, boolean data, boolean ingest) {
         HttpHost host = new HttpHost("127.0.0.1", port);
         Set<HttpHost> boundHosts = new HashSet<>(2);

client/sniffer/src/test/resources/es6_nodes_publication_address_format.json

@@ -0,0 +1,30 @@
+{
+  "_nodes": {
+    "total": 8,
+    "successful": 8,
+    "failed": 0
+  },
+  "cluster_name": "elasticsearch",
+  "nodes": {
+    "ikXK_skVTfWkhONhldnbkw": {
+      "name": "m1",
+      "transport_address": "127.0.0.1:9300",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "master",
+        "ingest"
+      ],
+      "attributes": { },
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9200"
+        ],
+        "publish_address": "127.0.0.1:9200",
+        "max_content_length_in_bytes": 104857600
+      }
+    }
+  }
+}

client/sniffer/src/test/resources/es7_nodes_publication_address_format.json

@@ -0,0 +1,30 @@
+{
+  "_nodes": {
+    "total": 8,
+    "successful": 8,
+    "failed": 0
+  },
+  "cluster_name": "elasticsearch",
+  "nodes": {
+    "ikXK_skVTfWkhONhldnbkw": {
+      "name": "m1",
+      "transport_address": "127.0.0.1:9300",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "master",
+        "ingest"
+      ],
+      "attributes": { },
+      "http": {
+        "bound_address": [
+            "elastic.test:9200"
+        ],
+        "publish_address": "elastic.test/127.0.0.1:9200",
+        "max_content_length_in_bytes": 104857600
+      }
+    }
+  }
+}