diff --git a/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala b/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala index 5f2be9053515d..d9e89e1d11957 100644 --- a/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala +++ b/core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala @@ -40,6 +40,7 @@ import org.apache.kafka.server.authorizer._ import org.apache.zookeeper.client.ZKClientConfig import scala.annotation.nowarn +import scala.collection.mutable.ArrayBuffer import scala.collection.{Seq, mutable} import scala.jdk.CollectionConverters._ import scala.util.{Failure, Random, Success, Try} @@ -63,9 +64,15 @@ object AclAuthorizer { def exists: Boolean = zkVersion != ZkVersion.UnknownVersion } - class AclSeqs(classes: Seq[AclEntry]*) { - def find(p: AclEntry => Boolean): Option[AclEntry] = classes.flatMap(_.find(p)).headOption - def isEmpty: Boolean = !classes.exists(_.nonEmpty) + class AclSeqs(seqs: Seq[AclEntry]*) { + def find(p: AclEntry => Boolean): Option[AclEntry] = { + // Lazily iterate through the inner `Seq` elements and stop as soon as we find a match + val it = seqs.iterator.flatMap(_.find(p)) + if (it.hasNext) Some(it.next) + else None + } + + def isEmpty: Boolean = !seqs.exists(_.nonEmpty) } val NoAcls = VersionedAcls(Set.empty, ZkVersion.UnknownVersion) @@ -344,8 +351,10 @@ class AclAuthorizer extends Authorizer with Logging { } else false } - @nowarn("cat=deprecation") + @nowarn("cat=deprecation&cat=optimizer") private def matchingAcls(resourceType: ResourceType, resourceName: String): AclSeqs = { + // this code is performance sensitive, make sure to run AclAuthorizerBenchmark after any changes + // save aclCache reference to a local val to get a consistent view of the cache during acl updates. val aclCacheSnapshot = aclCache val wildcard = aclCacheSnapshot.get(new ResourcePattern(resourceType, ResourcePattern.WILDCARD_RESOURCE, PatternType.LITERAL)) @@ -356,11 +365,13 @@ class AclAuthorizer extends Authorizer with Logging { .map(_.acls.toBuffer) .getOrElse(mutable.Buffer.empty) - val prefixed = aclCacheSnapshot + val prefixed = new ArrayBuffer[AclEntry] + aclCacheSnapshot .from(new ResourcePattern(resourceType, resourceName, PatternType.PREFIXED)) .to(new ResourcePattern(resourceType, resourceName.take(1), PatternType.PREFIXED)) - .flatMap { case (resource, acls) => if (resourceName.startsWith(resource.name)) acls.acls else Seq.empty } - .toBuffer + .foreach { case (resource, acls) => + if (resourceName.startsWith(resource.name)) prefixed ++= acls.acls + } new AclSeqs(prefixed, wildcard, literal) } diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AclAuthorizerBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AclAuthorizerBenchmark.java index c36fb314a45b0..12ac24327d465 100644 --- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AclAuthorizerBenchmark.java +++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AclAuthorizerBenchmark.java @@ -68,18 +68,18 @@ @Measurement(iterations = 15) @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.MILLISECONDS) - public class AclAuthorizerBenchmark { - @Param({"5000", "10000", "50000"}) + @Param({"10000", "50000", "200000"}) private int resourceCount; //no. of. rules per resource - @Param({"5", "10", "15"}) + @Param({"10", "50"}) private int aclCount; - private int hostPreCount = 1000; + private final int hostPreCount = 1000; + private final String resourceNamePrefix = "foo-bar35_resource-"; - private AclAuthorizer aclAuthorizer = new AclAuthorizer(); - private KafkaPrincipal principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "test-user"); + private final AclAuthorizer aclAuthorizer = new AclAuthorizer(); + private final KafkaPrincipal principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "test-user"); private List actions = new ArrayList<>(); private RequestContext context; @@ -87,11 +87,16 @@ public class AclAuthorizerBenchmark { public void setup() throws Exception { setFieldValue(aclAuthorizer, AclAuthorizer.class.getDeclaredField("aclCache").getName(), prepareAclCache()); - actions = Collections.singletonList(new Action(AclOperation.WRITE, new ResourcePattern(ResourceType.TOPIC, "resource-1", PatternType.LITERAL), - 1, true, true)); - context = new RequestContext(new RequestHeader(ApiKeys.PRODUCE, Integer.valueOf(1).shortValue(), "someclient", 1), - "1", InetAddress.getLocalHost(), KafkaPrincipal.ANONYMOUS, - ListenerName.normalised("listener"), SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY); + // By adding `-95` to the resource name prefix, we cause the `TreeMap.from/to` call to return + // most map entries. In such cases, we rely on the filtering based on `String.startsWith` + // to return the matching ACLs. Using a more efficient data structure (e.g. a prefix + // tree) should improve performance significantly). + actions = Collections.singletonList(new Action(AclOperation.WRITE, + new ResourcePattern(ResourceType.TOPIC, resourceNamePrefix + 95, PatternType.LITERAL), + 1, true, true)); + context = new RequestContext(new RequestHeader(ApiKeys.PRODUCE, Integer.valueOf(1).shortValue(), + "someclient", 1), "1", InetAddress.getLocalHost(), KafkaPrincipal.ANONYMOUS, + ListenerName.normalised("listener"), SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY); } private void setFieldValue(Object obj, String fieldName, Object value) throws Exception { @@ -103,10 +108,9 @@ private void setFieldValue(Object obj, String fieldName, Object value) throws Ex private TreeMap prepareAclCache() { Map> aclEntries = new HashMap<>(); for (int resourceId = 0; resourceId < resourceCount; resourceId++) { - ResourcePattern resource = new ResourcePattern( (resourceId % 10 == 0) ? ResourceType.GROUP : ResourceType.TOPIC, - "resource-" + resourceId, + resourceNamePrefix + resourceId, (resourceId % 5 == 0) ? PatternType.PREFIXED : PatternType.LITERAL); Set entries = aclEntries.computeIfAbsent(resource, k -> new HashSet<>()); @@ -118,20 +122,22 @@ private TreeMap prepareAclCache() { } } - ResourcePattern resourceWildcard = new ResourcePattern(ResourceType.TOPIC, ResourcePattern.WILDCARD_RESOURCE, PatternType.LITERAL); - ResourcePattern resourcePrefix = new ResourcePattern(ResourceType.TOPIC, "resource-", PatternType.PREFIXED); - - Set entriesWildcard = aclEntries.computeIfAbsent(resourceWildcard, k -> new HashSet<>()); + ResourcePattern resourcePrefix = new ResourcePattern(ResourceType.TOPIC, resourceNamePrefix, + PatternType.PREFIXED); Set entriesPrefix = aclEntries.computeIfAbsent(resourcePrefix, k -> new HashSet<>()); - for (int hostId = 0; hostId < hostPreCount; hostId++) { - AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId, AclOperation.READ, AclPermissionType.ALLOW); + AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId, + AclOperation.READ, AclPermissionType.ALLOW); entriesPrefix.add(new AclEntry(ace)); } + ResourcePattern resourceWildcard = new ResourcePattern(ResourceType.TOPIC, ResourcePattern.WILDCARD_RESOURCE, + PatternType.LITERAL); + Set entriesWildcard = aclEntries.computeIfAbsent(resourceWildcard, k -> new HashSet<>()); // get dynamic entries number for wildcard acl for (int hostId = 0; hostId < resourceCount / 10; hostId++) { - AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId, AclOperation.READ, AclPermissionType.ALLOW); + AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId, + AclOperation.READ, AclPermissionType.ALLOW); entriesWildcard.add(new AclEntry(ace)); } @@ -155,7 +161,7 @@ public void testAclsIterator() { } @Benchmark - public void testAuthorizer() throws Exception { + public void testAuthorizer() { aclAuthorizer.authorize(context, actions); } }