KAFKA-7185: Allow empty resource name when matching ACLs (#5400)

Reviewers: Ismael Juma <ismael@juma.me.uk>, Rajini Sivaram <rajinisivaram@googlemail.com>

core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala

@@ -238,7 +238,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
 
       val prefixed = aclCache.range(
         Resource(resourceType, resourceName, PatternType.PREFIXED),
-        Resource(resourceType, resourceName.substring(0, 1), PatternType.PREFIXED)
+        Resource(resourceType, resourceName.take(1), PatternType.PREFIXED)
       )
         .filterKeys(resource => resourceName.startsWith(resource.name))
         .flatMap { case (resource, versionedAcls) => versionedAcls.acls }

core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala

@@ -92,6 +92,19 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
     simpleAclAuthorizer.authorize(session, Read, Resource(Topic, "something", PREFIXED))
   }
 
+  @Test
+  def testAuthorizeWithEmptyResourceName(): Unit = {
+    assertFalse(simpleAclAuthorizer.authorize(session, Read, Resource(Group, "", LITERAL)))
+    simpleAclAuthorizer.addAcls(Set[Acl](allowReadAcl), Resource(Group, WildCardResource, LITERAL))
+    assertTrue(simpleAclAuthorizer.authorize(session, Read, Resource(Group, "", LITERAL)))
+  }
+
+  // Authorizing the empty resource is not supported because we create a znode with the resource name.
+  @Test(expected = classOf[IllegalArgumentException])
+  def testEmptyAclThrowsException(): Unit = {
+    simpleAclAuthorizer.addAcls(Set[Acl](allowReadAcl), Resource(Group, "", LITERAL))
+  }
+
   @Test
   def testTopicAcl() {
     val user1 = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)