Add a test to verify produce, consume with ACLs for topic/group wildcard resources
@@ -67,7 +67,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
val numRecords = 1
val group = "group"
val topic = "e2etopic"
- val topicWildcard = "*"
+ val wildcard = "*"
val part = 0
val tp = new TopicPartition(topic, part)
val topicAndPartition = TopicAndPartition(topic, part)
@@ -79,6 +79,8 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
val topicResource = new Resource(Topic, topic)
val groupResource = new Resource(Group, group)
val clusterResource = Resource.ClusterResource
+ val wildcardTopicResource = new Resource(Topic, wildcard)
+ val wildcardGroupResource = new Resource(Group, wildcard)
// Arguments to AclCommand to set ACLs. There are three definitions here:
// 1- Provides read and write access to topic
@@ -93,7 +95,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
def topicBrokerReadAclArgs: Array[String] = Array("--authorizer-properties",
s"zookeeper.connect=$zkConnect",
s"--add",
- s"--topic=$topicWildcard",
+ s"--topic=$wildcard",
s"--operation=Read",
s"--allow-principal=$kafkaPrincipalType:$kafkaPrincipal")
def produceAclArgs: Array[String] = Array("--authorizer-properties",
@@ -135,6 +137,15 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
s"--group=$group",
s"--operation=Read",
s"--allow-principal=$kafkaPrincipalType:$clientPrincipal")
+ def produceConsumeWildcardAclArgs: Array[String] = Array("--authorizer-properties",
+ s"zookeeper.connect=$zkConnect",
+ s"--add",
+ s"--topic=$wildcard",
+ s"--group=$wildcard",
+ s"--consumer",
+ s"--producer",
+ s"--allow-principal=$kafkaPrincipalType:$clientPrincipal")
+
def ClusterActionAcl = Set(new Acl(new KafkaPrincipal(kafkaPrincipalType, kafkaPrincipal), Allow, Acl.WildCardHost, ClusterAction))
def TopicBrokerReadAcl = Set(new Acl(new KafkaPrincipal(kafkaPrincipalType, kafkaPrincipal), Allow, Acl.WildCardHost, Read))
def GroupReadAcl = Set(new Acl(new KafkaPrincipal(kafkaPrincipalType, clientPrincipal), Allow, Acl.WildCardHost, Read))
@@ -201,6 +212,22 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
consumeRecords(this.consumers.head, numRecords)
}
+ @Test
+ def testProduceConsumeWithWildcardAcls(): Unit = {
+ setWildcardResourceAcls()
+ sendRecords(numRecords, tp)
+ consumers.head.subscribe(List(topic).asJava)
+ consumeRecords(this.consumers.head, numRecords)
+ }
+
+ private def setWildcardResourceAcls() {
+ AclCommand.main(produceConsumeWildcardAclArgs)
+ servers.foreach { s =>
+ TestUtils.waitAndVerifyAcls(TopicReadAcl ++ TopicWriteAcl ++ TopicDescribeAcl ++ TopicBrokerReadAcl, s.apis.authorizer.get, wildcardTopicResource)
+ TestUtils.waitAndVerifyAcls(GroupReadAcl, s.apis.authorizer.get, wildcardGroupResource)
+ }
+ }
+
protected def setAclsAndProduce() {
AclCommand.main(produceAclArgs)
AclCommand.main(consumeAclArgs)
