Skip to content

Kafka-2587: Only notification handler will update the cache and all verifications will use waitUntilTrue. #277

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Parth-Brahmbhatt wants to merge 4 commits into from
Closed

Kafka-2587: Only notification handler will update the cache and all verifications will use waitUntilTrue. #277

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bd26222
KAFKA-2587:Increasing timeout for the test verification.
Parth-Brahmbhatt Sep 29, 2015
736e132
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka …
Parth-Brahmbhatt Oct 5, 2015
4830bc8
KAFKA-2587: Only notification handler will update the cache and all v…
Parth-Brahmbhatt Oct 5, 2015
a9b03a6
Addressing comments.
Parth-Brahmbhatt Oct 7, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,6 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
ZkUtils.createPersistentPath(zkClient, path, Json.encode(Acl.toJsonCompatibleMap(updatedAcls)))

updateAclChangedFlag(resource)
updateCache(resource, updatedAcls)
}
}

Expand All @@ -186,7 +185,6 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
ZkUtils.deletePath(zkClient, toResourcePath(resource))

updateAclChangedFlag(resource)
updateCache(resource, filteredAcls)
}

aclNeedsRemoval
Expand All @@ -197,7 +195,6 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
if (ZkUtils.pathExists(zkClient, toResourcePath(resource))) {
ZkUtils.deletePath(zkClient, toResourcePath(resource))
updateAclChangedFlag(resource)
updateCache(resource, Set.empty[Acl])
true
} else false
}
Expand Down
6 changes: 3 additions & 3 deletions core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ class AclCommandTest extends ZooKeeperTestHarness with Logging {
val (acls, cmd) = getAclToCommand(permissionType, operationToCmd._1)
AclCommand.main(args ++ cmd ++ resourceCmd ++ operationToCmd._2 :+ "--add")
for (resource <- resources) {
Assert.assertEquals(acls, getAuthorizer(brokerProps).getAcls(resource))
TestUtils.waitAndVerifyAcls(acls, getAuthorizer(brokerProps), resource)
}

testRemove(resources, resourceCmd, args, brokerProps)
Expand All @@ -97,7 +97,7 @@ class AclCommandTest extends ZooKeeperTestHarness with Logging {
AclCommand.main(args ++ getCmd(Allow) ++ resourceCommand ++ cmd :+ "--add")
for ((resources, acls) <- resourcesToAcls) {
for (resource <- resources) {
Assert.assertEquals(acls, getAuthorizer(brokerProps).getAcls(resource))
TestUtils.waitAndVerifyAcls(acls, getAuthorizer(brokerProps), resource)
}
}
testRemove(resourcesToAcls.keys.flatten.toSet, resourceCommand, args, brokerProps)
Expand All @@ -108,7 +108,7 @@ class AclCommandTest extends ZooKeeperTestHarness with Logging {
for (resource <- resources) {
Console.withIn(new StringReader(s"y${AclCommand.Newline}" * resources.size)) {
AclCommand.main(args ++ resourceCmd :+ "--remove")
Assert.assertEquals(Set.empty[Acl], getAuthorizer(brokerProps).getAcls(resource))
TestUtils.waitAndVerifyAcls(Set.empty[Acl], getAuthorizer(brokerProps), resource)
}
}
}
Expand Down
7 changes: 3 additions & 4 deletions core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,14 +211,14 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
)

resourceToAcls foreach { case (key, value) => changeAclAndVerify(Set.empty[Acl], value, Set.empty[Acl], key) }
assertEquals(resourceToAcls + (resource -> acls), simpleAclAuthorizer.getAcls())
TestUtils.waitUntilTrue(() => resourceToAcls + (resource -> acls) == simpleAclAuthorizer.getAcls(), "changes not propagated in timeout period.")

//test remove acl from existing acls.
acls = changeAclAndVerify(acls, Set.empty[Acl], Set(acl1, acl5))

//test remove all acls for resource
simpleAclAuthorizer.removeAcls(resource)
TestUtils.waitUntilTrue(() => simpleAclAuthorizer.getAcls(resource) == Set.empty[Acl], "changes not propagated in timeout period.")
TestUtils.waitAndVerifyAcls(Set.empty[Acl], simpleAclAuthorizer, resource)
assertTrue(!ZkUtils.pathExists(zkClient, simpleAclAuthorizer.toResourcePath(resource)))

//test removing last acl also deletes zookeeper path
Expand Down Expand Up @@ -261,8 +261,7 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
acls --=removedAcls
}

TestUtils.waitUntilTrue(() => simpleAclAuthorizer.getAcls(resource) == acls,
s"changes not propagated in timeout period. expected $acls but got ${simpleAclAuthorizer.getAcls(resource)}", waitTime = 10000)
TestUtils.waitAndVerifyAcls(acls, simpleAclAuthorizer, resource)

acls
}
Expand Down
6 changes: 6 additions & 0 deletions core/src/test/scala/unit/kafka/utils/TestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import java.security.cert.X509Certificate
import javax.net.ssl.X509TrustManager
import charset.Charset

import kafka.security.auth.{Resource, Authorizer, Acl}
import org.apache.kafka.common.protocol.SecurityProtocol
import org.apache.kafka.common.utils.Utils._

Expand Down Expand Up @@ -942,6 +943,11 @@ object TestUtils extends Logging {
trustManager
}

def waitAndVerifyAcls(expected: Set[Acl], authorizer: Authorizer, resource: Resource) = {
TestUtils.waitUntilTrue(() => authorizer.getAcls(resource) == expected,
s"expected acls $expected but got ${authorizer.getAcls(resource)}", waitTime = 10000)
}

}

class IntEncoder(props: VerifiableProperties = null) extends Encoder[Int] {
Expand Down