KAFKA-697 ConsoleConsumer throws InvalidConfigException for . in client id; reviewed by Jun Rao, Neha Narkhede, John Fung

core/src/main/scala/kafka/common/Config.scala

@@ -23,14 +23,14 @@ import kafka.utils.Logging
 trait Config extends Logging {
 
   def validateChars(prop: String, value: String) {
-    val legalChars = "[a-zA-Z0-9_-]"
+    val legalChars = "[a-zA-Z0-9\\._\\-]"
     val rgx = new Regex(legalChars + "*")
 
     rgx.findFirstIn(value) match {
       case Some(t) =>
         if (!t.equals(value))
-          throw new InvalidConfigException(prop + " " + value + " is illegal, contains a character other than ASCII alphanumerics, _ and -")
-      case None => throw new InvalidConfigException(prop + " " + value + " is illegal, contains a character other than ASCII alphanumerics, _ and -")
+          throw new InvalidConfigException(prop + " " + value + " is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'")
+      case None => throw new InvalidConfigException(prop + " " + value + " is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'")
     }
   }
 }

core/src/main/scala/kafka/common/Topic.scala

@@ -20,21 +20,23 @@ package kafka.common
 import util.matching.Regex
 
 object Topic {
-  private val legalChars = "[a-zA-Z0-9_-]"
+  private val legalChars = "[a-zA-Z0-9\\._\\-]"
   private val maxNameLength = 255
   private val rgx = new Regex(legalChars + "+")
 
   def validate(topic: String) {
     if (topic.length <= 0)
       throw new InvalidTopicException("topic name is illegal, can't be empty")
+    else if (topic.equals(".") || topic.equals(".."))
+      throw new InvalidTopicException("topic name cannot be \".\" or \"..\"")
     else if (topic.length > maxNameLength)
       throw new InvalidTopicException("topic name is illegal, can't be longer than " + maxNameLength + " characters")
 
     rgx.findFirstIn(topic) match {
       case Some(t) =>
         if (!t.equals(topic))
-          throw new InvalidTopicException("topic name " + topic + " is illegal, contains a character other than ASCII alphanumerics, _ and -")
-      case None => throw new InvalidTopicException("topic name " + topic + " is illegal,  contains a character other than ASCII alphanumerics, _ and -")
+          throw new InvalidTopicException("topic name " + topic + " is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'")
+      case None => throw new InvalidTopicException("topic name " + topic + " is illegal,  contains a character other than ASCII alphanumerics, '.', '_' and '-'")
     }
   }
 }

core/src/test/scala/unit/kafka/common/ConfigTest.scala

@@ -29,7 +29,7 @@ class ConfigTest {
   @Test
   def testInvalidClientIds() {
     val invalidClientIds = new ArrayBuffer[String]()
-    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', '.', ' ', '\t', '\r', '\n', '=')
+    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', ' ', '\t', '\r', '\n', '=')
     for (weirdChar <- badChars) {
       invalidClientIds += "Is" + weirdChar + "illegal"
     }
@@ -45,7 +45,7 @@ class ConfigTest {
     }
 
     val validClientIds = new ArrayBuffer[String]()
-    validClientIds += ("valid", "CLIENT", "iDs", "ar6", "VaL1d", "_0-9_", "")
+    validClientIds += ("valid", "CLIENT", "iDs", "ar6", "VaL1d", "_0-9_.", "")
     for (i <- 0 until validClientIds.size) {
       try {
         ProducerConfig.validateClientId(validClientIds(i))
@@ -59,7 +59,7 @@ class ConfigTest {
   @Test
   def testInvalidGroupIds() {
     val invalidGroupIds = new ArrayBuffer[String]()
-    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', '.', ' ', '\t', '\r', '\n', '=')
+    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', ' ', '\t', '\r', '\n', '=')
     for (weirdChar <- badChars) {
       invalidGroupIds += "Is" + weirdChar + "illegal"
     }
@@ -75,7 +75,7 @@ class ConfigTest {
     }
 
     val validGroupIds = new ArrayBuffer[String]()
-    validGroupIds += ("valid", "GROUP", "iDs", "ar6", "VaL1d", "_0-9_", "")
+    validGroupIds += ("valid", "GROUP", "iDs", "ar6", "VaL1d", "_0-9_.", "")
     for (i <- 0 until validGroupIds.size) {
       try {
         ConsumerConfig.validateGroupId(validGroupIds(i))

core/src/test/scala/unit/kafka/common/TopicTest.scala

@@ -32,7 +32,7 @@ class TopicTest {
     for (i <- 1 to 6)
       longName += longName
     invalidTopicNames += longName
-    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', '.')
+    val badChars = Array('/', '\\', ',', '\0', ':', "\"", '\'', ';', '*', '?', ' ', '\t', '\r', '\n', '=')
     for (weirdChar <- badChars) {
       invalidTopicNames += "Is" + weirdChar + "illegal"
     }
@@ -48,7 +48,7 @@ class TopicTest {
     }
 
     val validTopicNames = new ArrayBuffer[String]()
-    validTopicNames += ("valid", "TOPIC", "nAmEs", "ar6", "VaL1d", "_0-9_")
+    validTopicNames += ("valid", "TOPIC", "nAmEs", "ar6", "VaL1d", "_0-9_.")
     for (i <- 0 until validTopicNames.size) {
       try {
         Topic.validate(validTopicNames(i))