KAFKA-19782: improve Authorizer by using (Patricia) trie

build.gradle

@@ -1858,6 +1858,7 @@ project(':clients') {
     implementation libs.opentelemetryProto
     implementation libs.protobuf
     implementation libs.slf4jApi
+    implementation libs.commonsCollections4
 
     // libraries which should be added as runtime dependencies in generated pom.xml should be defined here:
     shadowed libs.zstd

checkstyle/import-control.xml

@@ -201,6 +201,7 @@
     <subpackage name="utils">
       <allow pkg="org.apache.kafka.common" />
       <allow pkg="org.apache.logging.log4j" />
+      <allow pkg="org.apache.commons.collections4.trie" />
     </subpackage>
 
     <subpackage name="quotas">
@@ -253,6 +254,10 @@
     <!-- This is required to make AlterConfigPolicyTest work. -->
     <allow pkg="org.apache.kafka.server.policy" />
 
+    <subpackage name="authorizer">
+      <allow pkg="org.apache.commons.collections4.trie" />
+    </subpackage>
+
     <subpackage name="telemetry">
       <allow class="org.apache.kafka.server.authorizer.AuthorizableRequestContext" />
     </subpackage>

clients/src/main/java/org/apache/kafka/common/utils/TrieSet.java

@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.utils;
+
+import org.apache.commons.collections4.trie.PatriciaTrie;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A set of strings backed by a PATRICIA trie.
+ * <p>
+ * A {@link org.apache.commons.collections4.trie.PatriciaTrie}
+ * (Practical Algorithm to Retrieve Information
+ * Coded in Alphanumeric) implements efficient worst-case O(K)-time operations, where K is the max key bit-length in the
+ * tree.
+ * </p>
+ */
+public class TrieSet implements Set<String> {
+    private final PatriciaTrie<String> trie;
+
+    public TrieSet() {
+        trie = new PatriciaTrie<>();
+    }
+
+    /**
+     * Get a set view of all strings with the same prefix.
+     * <p>
+     * The view is backed by the trie. If you want to modify the trie while iterating the view, create a copy first.
+     * </p>
+     * @param key the prefix to search for
+     * @return a set view of all strings with the given prefix
+     */
+    public Set<String> prefixSet(String key) {
+        return trie.prefixMap(key).keySet();
+    }
+
+    @Override
+    public int size() {
+        return trie.size();
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return size() == 0;
+    }
+
+    @Override
+    public boolean contains(Object o) {
+        return trie.containsKey(o);
+    }
+
+    @Override
+    public Iterator<String> iterator() {
+        return trie.keySet().iterator();
+    }
+
+    @Override
+    public Object[] toArray() {
+        return trie.keySet().toArray();
+    }
+
+    @Override
+    public <T> T[] toArray(T[] a) {
+        Objects.requireNonNull(a);
+        return trie.keySet().toArray(a);
+    }
+
+    @Override
+    public boolean add(String s) {
+        Objects.requireNonNull(s, "Expect non-null element to add");
+        return trie.putIfAbsent(s, s) == null;
+    }
+
+    @Override
+    public boolean remove(Object o) {
+        return trie.remove(o) != null;
+    }
+
+    @Override
+    public boolean containsAll(Collection<?> c) {
+        for (final Object k : c) {
+            if (!trie.containsKey(k))
+                return false;
+        }
+        return true;
+    }
+
+    @Override
+    public boolean addAll(Collection<? extends String> c) {
+        boolean mutated = false;
+        for (final String k : c)
+            mutated |= add(k);
+        return mutated;
+    }
+
+    @Override
+    public boolean retainAll(Collection<?> c) {
+        boolean mutated = false;
+        Iterator<String> it = iterator();
+        while (it.hasNext()) {
+            String element = it.next();
+            if (!c.contains(element)) {
+                it.remove();
+                mutated = true;
+            }
+        }
+        return mutated;
+    }
+
+    @Override
+    public boolean removeAll(Collection<?> c) {
+        boolean mutated = false;
+        for (final Object k : c)
+            mutated |= remove(k);
+        return mutated;
+    }
+
+    @Override
+    public void clear() {
+        trie.clear();
+    }
+}

clients/src/main/java/org/apache/kafka/server/authorizer/Authorizer.java

@@ -30,8 +30,10 @@
 import org.apache.kafka.common.resource.ResourceType;
 import org.apache.kafka.common.security.auth.KafkaPrincipal;
 import org.apache.kafka.common.utils.SecurityUtils;
+import org.apache.kafka.common.utils.TrieSet;
 
 import java.io.Closeable;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumMap;
 import java.util.HashSet;
@@ -205,10 +207,10 @@ op, new ResourcePattern(resourceType, "hardcode", PatternType.LITERAL),
                     put(PatternType.LITERAL, new HashSet<>());
                     put(PatternType.PREFIXED, new HashSet<>());
                 }};
-        EnumMap<PatternType, Set<String>> allowPatterns =
+        EnumMap<PatternType, TrieSet> allowPatterns =
             new EnumMap<>(PatternType.class) {{
-                    put(PatternType.LITERAL, new HashSet<>());
-                    put(PatternType.PREFIXED, new HashSet<>());
+                    put(PatternType.LITERAL, new TrieSet());
+                    put(PatternType.PREFIXED, new TrieSet());
                 }};
 
         boolean hasWildCardAllow = false;
@@ -270,23 +272,16 @@ op, new ResourcePattern(resourceType, "hardcode", PatternType.LITERAL),
 
         // For any literal allowed, if there's no dominant literal and prefix denied, return allow.
         // For any prefix allowed, if there's no dominant prefix denied, return allow.
-        for (Map.Entry<PatternType, Set<String>> entry : allowPatterns.entrySet()) {
-            for (String allowStr : entry.getValue()) {
-                if (entry.getKey() == PatternType.LITERAL
-                        && denyPatterns.get(PatternType.LITERAL).contains(allowStr))
-                    continue;
-                StringBuilder sb = new StringBuilder();
-                boolean hasDominatedDeny = false;
-                for (char ch : allowStr.toCharArray()) {
-                    sb.append(ch);
-                    if (denyPatterns.get(PatternType.PREFIXED).contains(sb.toString())) {
-                        hasDominatedDeny = true;
-                        break;
-                    }
-                }
-                if (!hasDominatedDeny)
-                    return AuthorizationResult.ALLOWED;
+        for (Map.Entry<PatternType, TrieSet> entry : allowPatterns.entrySet()) {
+            TrieSet toAllow = entry.getValue();
+            if (entry.getKey() == PatternType.LITERAL)
+                toAllow.removeAll(denyPatterns.get(PatternType.LITERAL));
+            for (final String d : denyPatterns.get(PatternType.PREFIXED)) {
+                List<String> toDeny = new ArrayList<>(toAllow.prefixSet(d));
+                toAllow.removeAll(toDeny);
             }
+            if (!toAllow.isEmpty())
+                return AuthorizationResult.ALLOWED;
         }
 
         return AuthorizationResult.DENIED;