avoid unsafe split when validate hostname which might be ipv6 address (…

…apache#5713)

### Motivation
Pulsar doesn't support deploy on IPv6 network environment.
This PR just makes an effort to make it.
The error happens if the client connect to brokers by IPv6 address, like fec0:0:0:ffff::1.
- Wrong format: fec0:0:0:ffff::1:6650
- Correct format: [fec0:0:0:ffff::1]:6650

Cause the split regex is ':', brackets are needed and the ip:port can't split by ':' directly.

### Modifications

validateHostName in ServiceURI

pulsar-common/src/main/java/org/apache/pulsar/common/net/ServiceURI.java

@@ -28,6 +28,7 @@
 import java.net.URI;
 import java.util.List;
 import java.util.stream.Collectors;
+
 import lombok.AccessLevel;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -134,21 +135,21 @@ public static ServiceURI create(URI uri) {
     private static String validateHostName(String serviceName,
                                            String[] serviceInfos,
                                            String hostname) {
-        String[] parts = hostname.split(":");
-        if (parts.length >= 3) {
+        URI uri = null;
+        try {
+            uri = URI.create("dummyscheme://" + hostname);
+        } catch (IllegalArgumentException e) {
             throw new IllegalArgumentException("Invalid hostname : " + hostname);
-        } else if (parts.length == 2) {
-            try {
-                Integer.parseUnsignedInt(parts[1]);
-            } catch (NumberFormatException nfe) {
-                throw new IllegalArgumentException("Invalid hostname : " + hostname);
-            }
-            return hostname;
-        } else if (parts.length == 1) {
-            return hostname + ":" + getServicePort(serviceName, serviceInfos);
-        } else {
-            return hostname;
         }
+        String host = uri.getHost();
+        if (host == null) {
+            throw new IllegalArgumentException("Invalid hostname : " + hostname);
+        }
+        int port = uri.getPort();
+        if (port == -1) {
+            port = getServicePort(serviceName, serviceInfos);
+        }
+        return host + ":" + port;
     }
 
     private final String serviceName;

pulsar-common/src/test/java/org/apache/pulsar/common/net/ServiceURITest.java

@@ -18,8 +18,7 @@
  */
 package org.apache.pulsar.common.net;
 
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.*;
 import static org.testng.internal.junit.ArrayAsserts.assertArrayEquals;
 
 import java.net.URI;
@@ -55,6 +54,7 @@ public void testInvalidServiceUris() {
             "pulsar://localhost:6650:6651/",    // invalid hostname pair
             "pulsar://localhost:xyz/",          // invalid port
             "pulsar://localhost:-6650/",        // negative port
+            "pulsar://fec0:0:0:ffff::1:6650",   // missing brackets
         };
 
         for (String uri : uris) {
@@ -122,6 +122,30 @@ public void testUserInfo() {
             "/path/to/namespace");
     }
 
+    @Test
+    public void testIpv6Uri() {
+        String serviceUri = "pulsar://pulsaruser@[fec0:0:0:ffff::1]:6650/path/to/namespace";
+        assertServiceUri(
+            serviceUri,
+            "pulsar",
+            new String[0],
+            "pulsaruser",
+            new String[] { "[fec0:0:0:ffff::1]:6650" },
+            "/path/to/namespace");
+    }
+
+    @Test
+    public void testIpv6UriWithoutPulsarPort() {
+        String serviceUri = "pulsar://[fec0:0:0:ffff::1]/path/to/namespace";
+        assertServiceUri(
+            serviceUri,
+            "pulsar",
+            new String[0],
+            null,
+            new String[] { "[fec0:0:0:ffff::1]:6650" },
+            "/path/to/namespace");
+    }
+
     @Test
     public void testMultipleHostsSemiColon() {
         String serviceUri = "pulsar://host1:6650;host2:6650;host3:6650/path/to/namespace";