KAFKA-2838; Allow comma in super users, allow comma in CLI authz prop…

…erties. Author: Parth Brahmbhatt <[email protected]> Reviewers: Sriharsha Chintalapani <[email protected]>, Jun Rao <[email protected]> Closes apache#529 from Parth-Brahmbhatt/KAFKA-2838
LioRoger · Nov 14, 2015 · 8e6bae2 · 8e6bae2
1 parent 5d87eac
commit 8e6bae2
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 5 deletions.
diff --git a/core/src/main/scala/kafka/admin/AclCommand.scala b/core/src/main/scala/kafka/admin/AclCommand.scala
@@ -257,12 +257,10 @@ object AclCommand {
       .defaultsTo(classOf[SimpleAclAuthorizer].getName)
 
     val authorizerPropertiesOpt = parser.accepts("authorizer-properties", "REQUIRED: properties required to configure an instance of Authorizer. " +
-      "These are comma separated key=val pairs. For the default authorizer the example values are: " +
-      "zookeeper.connect=localhost:2181")
+      "These are key=val pairs. For the default authorizer the example values are: zookeeper.connect=localhost:2181")
       .withRequiredArg
       .describedAs("authorizer-properties")
       .ofType(classOf[String])
-      .withValuesSeparatedBy(Delimiter)
 
     val topicOpt = parser.accepts("topic", "Comma separated list of topic to which acls should be added or removed. " +
       "A value of * indicates acl should apply to all topics.")

diff --git a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
@@ -84,7 +84,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
     val kafkaConfig = KafkaConfig.fromProps(props)
 
     superUsers = configs.get(SimpleAclAuthorizer.SuperUsersProp).collect {
-      case str: String if str.nonEmpty => str.split(",").map(s => KafkaPrincipal.fromString(s.trim)).toSet
+      case str: String if str.nonEmpty => str.split(";").map(s => KafkaPrincipal.fromString(s.trim)).toSet
     }.getOrElse(Set.empty[KafkaPrincipal])
 
     shouldAllowEveryoneIfNoAclIsFound = configs.get(SimpleAclAuthorizer.AllowEveryoneIfNoAclIsFoundProp).map(_.toString.toBoolean).getOrElse(false)

diff --git a/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala b/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala
@@ -34,7 +34,7 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
   val testHostName = "test.host.com"
   var session = new Session(testPrincipal, testHostName)
   var resource: Resource = null
-  val superUsers = "User:superuser1, User:superuser2"
+  val superUsers = "User:superuser1; User:superuser2"
   val username = "alice"
   var config: KafkaConfig = null