KAFKA-3448; Support zone index in IPv6 regex

When an address is written textually, the zone index is appended to the address, separated by a percent sign (%). The actual syntax of zone indices depends on the operating system.

Author: Som Sahu <sosahu@microsoft.com>
Author: Soumyajit Sahu <soumyajit-sahu@users.noreply.github.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #1305 from soumyajit-sahu/fixIPV6RegexPattern_trunk

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -50,7 +50,7 @@ public class Utils {
 
     // This matches URIs of formats: host:port and protocol:\\host:port
     // IPv6 is supported with [ip] pattern
-    private static final Pattern HOST_PORT_PATTERN = Pattern.compile(".*?\\[?([0-9a-zA-Z\\-.:]*)\\]?:([0-9]+)");
+    private static final Pattern HOST_PORT_PATTERN = Pattern.compile(".*?\\[?([0-9a-zA-Z\\-%.:]*)\\]?:([0-9]+)");
 
     public static final String NL = System.getProperty("line.separator");
 

clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java

@@ -37,6 +37,7 @@ public class UtilsTest {
         assertEquals("::1", getHost("[::1]:1234"));
         assertEquals("2001:db8:85a3:8d3:1319:8a2e:370:7348", getHost("PLAINTEXT://[2001:db8:85a3:8d3:1319:8a2e:370:7348]:5678"));
         assertEquals("2001:DB8:85A3:8D3:1319:8A2E:370:7348", getHost("PLAINTEXT://[2001:DB8:85A3:8D3:1319:8A2E:370:7348]:5678"));
+        assertEquals("fe80::b1da:69ca:57f7:63d8%3", getHost("PLAINTEXT://[fe80::b1da:69ca:57f7:63d8%3]:5678"));
     }
 
     @Test
@@ -47,6 +48,7 @@ public class UtilsTest {
         assertEquals(1234, getPort("[::1]:1234").intValue());
         assertEquals(5678, getPort("[2001:db8:85a3:8d3:1319:8a2e:370:7348]:5678").intValue());
         assertEquals(5678, getPort("[2001:DB8:85A3:8D3:1319:8A2E:370:7348]:5678").intValue());
+        assertEquals(5678, getPort("[fe80::b1da:69ca:57f7:63d8%3]:5678").intValue());
     }
 
     @Test

core/src/main/scala/kafka/cluster/BrokerEndPoint.scala

@@ -24,7 +24,7 @@ import org.apache.kafka.common.utils.Utils._
 
 object BrokerEndPoint {
 
-  private val uriParseExp = """\[?([0-9a-zA-Z\-.:]*)\]?:([0-9]+)""".r
+  private val uriParseExp = """\[?([0-9a-zA-Z\-%.:]*)\]?:([0-9]+)""".r
 
   /**
    * BrokerEndPoint URI is host:port or [ipv6_host]:port

core/src/main/scala/kafka/cluster/EndPoint.scala

@@ -26,7 +26,7 @@ import org.apache.kafka.common.utils.Utils
 
 object EndPoint {
 
-  private val uriParseExp = """^(.*)://\[?([0-9a-zA-Z\-.:]*)\]?:(-?[0-9]+)""".r
+  private val uriParseExp = """^(.*)://\[?([0-9a-zA-Z\-%.:]*)\]?:(-?[0-9]+)""".r
 
   def readFrom(buffer: ByteBuffer): EndPoint = {
     val port = buffer.getInt()

core/src/test/scala/unit/kafka/cluster/BrokerEndPointTest.scala

@@ -102,6 +102,11 @@ class BrokerEndPointTest extends Logging {
     endpoint = BrokerEndPoint.createBrokerEndPoint(1, connectionString)
     assert(endpoint.host == "::1")
     assert(endpoint.port == 9092)
+    // test for ipv6 with % character
+    connectionString = "[fe80::b1da:69ca:57f7:63d8%3]:9092"
+    endpoint = BrokerEndPoint.createBrokerEndPoint(1, connectionString)
+    assert(endpoint.host == "fe80::b1da:69ca:57f7:63d8%3")
+    assert(endpoint.port == 9092)
     // add test for uppercase in hostname
     connectionString = "MyHostname:9092"
     endpoint = BrokerEndPoint.createBrokerEndPoint(1, connectionString)
@@ -128,6 +133,12 @@ class BrokerEndPointTest extends Logging {
     assert(endpoint.host == "::1")
     assert(endpoint.port == 9092)
     assert(endpoint.connectionString ==  "PLAINTEXT://[::1]:9092")
+    // test for ipv6 with % character
+    connectionString = "PLAINTEXT://[fe80::b1da:69ca:57f7:63d8%3]:9092"
+    endpoint = EndPoint.createEndPoint(connectionString)
+    assert(endpoint.host == "fe80::b1da:69ca:57f7:63d8%3")
+    assert(endpoint.port == 9092)
+    assert(endpoint.connectionString ==  "PLAINTEXT://[fe80::b1da:69ca:57f7:63d8%3]:9092")
     // test hostname
     connectionString = "PLAINTEXT://MyHostname:9092"
     endpoint = EndPoint.createEndPoint(connectionString)