Handle wildcard parameter in ACL list API

src/main/java/com/michelin/ns4kafka/controller/acl/AclController.java

@@ -1,6 +1,5 @@
 package com.michelin.ns4kafka.controller.acl;
 
-import static com.michelin.ns4kafka.service.AclService.PUBLIC_GRANTED_TO;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidAclDeleteOnlyAdmin;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidImmutableField;
 import static com.michelin.ns4kafka.util.FormatErrorUtils.invalidNotFound;
@@ -49,32 +48,24 @@ public class AclController extends NamespacedResourceController {
      * @return A list of ACLs
      */
     @Get("{?limit}")
-    public List<AccessControlEntry> list(String namespace, Optional<AclLimit> limit) {
-        if (limit.isEmpty()) {
-            limit = Optional.of(AclLimit.ALL);
-        }
-
+    public List<AccessControlEntry> list(String namespace,
+                                         Optional<AclLimit> limit,
+                                         @QueryValue(defaultValue = "*") String name) {
         Namespace ns = getNamespace(namespace);
-        return switch (limit.get()) {
-            case GRANTEE -> aclService.findAllGrantedToNamespace(ns)
+        return switch (limit.orElse(AclLimit.ALL)) {
+            case GRANTEE -> aclService.findAllGrantedToNamespaceByWildcardName(ns, name)
                 .stream()
-                .sorted(Comparator.comparing(o -> o.getMetadata().getNamespace()))
+                .sorted(Comparator.comparing((AccessControlEntry acl) -> acl.getMetadata().getNamespace()))
                 .toList();
-            case GRANTOR -> aclService.findAllForCluster(ns.getMetadata().getCluster())
+            case GRANTOR -> aclService.findAllGrantedByNamespaceByWildcardName(ns, name)
                 .stream()
-                // granted by me
-                .filter(accessControlEntry -> accessControlEntry.getMetadata().getNamespace().equals(namespace))
-                // without the granted to me
-                .filter(accessControlEntry -> !accessControlEntry.getSpec().getGrantedTo().equals(namespace))
-                .sorted(Comparator.comparing(o -> o.getSpec().getGrantedTo()))
+                .sorted(Comparator.comparing(acl -> acl.getSpec().getGrantedTo()))
                 .toList();
-            default -> aclService.findAllForCluster(ns.getMetadata().getCluster())
+            default -> aclService.findAllRelatedToNamespaceByWildcardName(ns, name)
                 .stream()
-                .filter(accessControlEntry ->
-                    accessControlEntry.getMetadata().getNamespace().equals(namespace)
-                        || accessControlEntry.getSpec().getGrantedTo().equals(namespace)
-                        || accessControlEntry.getSpec().getGrantedTo().equals(PUBLIC_GRANTED_TO))
-                .sorted(Comparator.comparing(o -> o.getMetadata().getNamespace()))
+                .sorted(Comparator
+                    .comparing((AccessControlEntry acl) -> acl.getMetadata().getNamespace())
+                    .thenComparing(acl -> acl.getSpec().getGrantedTo()))
                 .toList();
         };
     }
@@ -83,12 +74,14 @@ public List<AccessControlEntry> list(String namespace, Optional<AclLimit> limit)
      * Get an ACL by namespace and name.
      *
      * @param namespace The name
-     * @param acl       The ACL name
+     * @param acl   The ACL name
      * @return The ACL
+     * @deprecated use list(String, Optional ALL, String name) instead.
      */
     @Get("/{acl}")
+    @Deprecated(since = "1.12.0")
     public Optional<AccessControlEntry> get(String namespace, String acl) {
-        return list(namespace, Optional.of(AclLimit.ALL))
+        return aclService.findAllRelatedToNamespace(getNamespace(namespace))
             .stream()
             .filter(accessControlEntry -> accessControlEntry.getMetadata().getName().equals(acl))
             .findFirst();

src/main/java/com/michelin/ns4kafka/repository/kafka/KafkaStore.java

@@ -141,7 +141,7 @@ private void verifyInternalTopic()
                 "The topic " + kafkaTopic + " should have only 1 partition but has " + numPartitions + ".");
         }
 
-        if (description.partitions().get(0).replicas().size() < kafkaStoreProperties.getReplicationFactor()
+        if (description.partitions().getFirst().replicas().size() < kafkaStoreProperties.getReplicationFactor()
             && log.isWarnEnabled()) {
             log.warn("The replication factor of the topic " + kafkaTopic + " is less than the desired one of "
                 + kafkaStoreProperties.getReplicationFactor()

src/main/java/com/michelin/ns4kafka/service/AclService.java

@@ -12,6 +12,7 @@
 import com.michelin.ns4kafka.model.Namespace;
 import com.michelin.ns4kafka.repository.AccessControlEntryRepository;
 import com.michelin.ns4kafka.service.executor.AccessControlEntryAsyncExecutor;
+import com.michelin.ns4kafka.util.RegexUtils;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.inject.qualifiers.Qualifiers;
 import jakarta.inject.Inject;
@@ -247,7 +248,7 @@ public void delete(AccessControlEntry accessControlEntry) {
     }
 
     /**
-     * Find all ACLs granted to given namespace.
+     * Find all ACLs granted to a given namespace.
      * Will also return public granted ACLs.
      *
      * @param namespace The namespace
@@ -256,9 +257,83 @@ public void delete(AccessControlEntry accessControlEntry) {
     public List<AccessControlEntry> findAllGrantedToNamespace(Namespace namespace) {
         return accessControlEntryRepository.findAll()
             .stream()
-            .filter(accessControlEntry ->
-                accessControlEntry.getSpec().getGrantedTo().equals(namespace.getMetadata().getName())
-                    || accessControlEntry.getSpec().getGrantedTo().equals(PUBLIC_GRANTED_TO))
+            .filter(acl -> acl.getSpec().getGrantedTo().equals(namespace.getMetadata().getName())
+                || acl.getSpec().getGrantedTo().equals(PUBLIC_GRANTED_TO))
+            .toList();
+    }
+
+    /**
+     * Find all ACLs that a given namespace granted to other namespaces.
+     *
+     * @param namespace The namespace
+     * @return A list of ACLs
+     */
+    public List<AccessControlEntry> findAllGrantedByNamespace(Namespace namespace) {
+        return accessControlEntryRepository.findAll()
+            .stream()
+            .filter(acl -> acl.getMetadata().getNamespace().equals(namespace.getMetadata().getName()))
+            .filter(acl -> !acl.getSpec().getGrantedTo().equals(namespace.getMetadata().getName()))
+            .toList();
+    }
+
+    /**
+     * Find all ACLs where the given namespace is either the grantor or the grantee, or the ACL is public.
+     *
+     * @param namespace The namespace
+     * @return A list of ACLs
+     */
+    public List<AccessControlEntry> findAllRelatedToNamespace(Namespace namespace) {
+        return accessControlEntryRepository.findAll()
+            .stream()
+            .filter(acl -> acl.getMetadata().getNamespace().equals(namespace.getMetadata().getName())
+                || acl.getSpec().getGrantedTo().equals(namespace.getMetadata().getName())
+                || acl.getSpec().getGrantedTo().equals(PUBLIC_GRANTED_TO))
+            .toList();
+    }
+
+    /**
+     * Find all ACLs granted to given namespace, filtered by name parameter.
+     * Will also return public granted ACLs.
+     *
+     * @param namespace The namespace
+     * @param name      The name parameter
+     * @return A list of ACLs
+     */
+    public List<AccessControlEntry> findAllGrantedToNamespaceByWildcardName(Namespace namespace, String name) {
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
+        return findAllGrantedToNamespace(namespace)
+            .stream()
+            .filter(acl -> RegexUtils.isResourceCoveredByRegex(acl.getMetadata().getName(), nameFilterPatterns))
+            .toList();
+    }
+
+    /**
+     * Find all ACLs that a given namespace granted to other namespaces, filtered by name parameter.
+     *
+     * @param namespace The namespace
+     * @param name      The name parameter
+     * @return A list of ACLs
+     */
+    public List<AccessControlEntry> findAllGrantedByNamespaceByWildcardName(Namespace namespace, String name) {
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
+        return findAllGrantedByNamespace(namespace)
+            .stream()
+            .filter(acl -> RegexUtils.isResourceCoveredByRegex(acl.getMetadata().getName(), nameFilterPatterns))
+            .toList();
+    }
+
+    /**
+     * Find all ACLs that a given namespace granted to other namespaces, filtered by name parameter.
+     *
+     * @param namespace The namespace
+     * @param name      The name parameter
+     * @return A list of ACLs
+     */
+    public List<AccessControlEntry> findAllRelatedToNamespaceByWildcardName(Namespace namespace, String name) {
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
+        return findAllRelatedToNamespace(namespace)
+            .stream()
+            .filter(acl -> RegexUtils.isResourceCoveredByRegex(acl.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 
@@ -312,7 +387,8 @@ public List<AccessControlEntry> findAllForNamespace(Namespace namespace) {
      * @return A list of ACLs
      */
     public List<AccessControlEntry> findAllForCluster(String cluster) {
-        return accessControlEntryRepository.findAll().stream()
+        return accessControlEntryRepository.findAll()
+            .stream()
             .filter(accessControlEntry -> accessControlEntry.getMetadata().getCluster().equals(cluster))
             .toList();
     }
@@ -360,13 +436,13 @@ public Optional<AccessControlEntry> findByName(String namespace, String name) {
     }
 
     /**
-     * Check if there is any ACL concerning the given resource.
+     * Check if the given resource is covered by any given ACLs.
      *
      * @param acls         The OWNER ACL list on resource
      * @param resourceName The resource name to check ACL against
      * @return true if there is any OWNER ACL concerning the given resource, false otherwise
      */
-    public boolean isAnyAclOfResource(List<AccessControlEntry> acls, String resourceName) {
+    public boolean isResourceCoveredByAcls(List<AccessControlEntry> acls, String resourceName) {
         return acls
             .stream()
             .anyMatch(acl -> switch (acl.getSpec().getResourcePatternType()) {

src/main/java/com/michelin/ns4kafka/service/ConnectClusterService.java

@@ -133,7 +133,7 @@ public List<ConnectCluster> findAllForNamespaceByPermissions(Namespace namespace
 
         return connectClusterRepository.findAllForCluster(namespace.getMetadata().getCluster())
             .stream()
-            .filter(connectCluster -> aclService.isAnyAclOfResource(acls, connectCluster.getMetadata().getName()))
+            .filter(connectCluster -> aclService.isResourceCoveredByAcls(acls, connectCluster.getMetadata().getName()))
             .toList();
     }
 
@@ -157,10 +157,10 @@ public List<ConnectCluster> findAllForNamespaceWithOwnerPermission(Namespace nam
      * @return The list of owned Connect cluster
      */
     public List<ConnectCluster> findByWildcardNameWithOwnerPermission(Namespace namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespaceWithOwnerPermission(namespace)
             .stream()
-            .filter(cc -> RegexUtils.filterByPattern(cc.getMetadata().getName(), nameFilterPatterns))
+            .filter(cc -> RegexUtils.isResourceCoveredByRegex(cc.getMetadata().getName(), nameFilterPatterns))
             .map(this::buildConnectClusterWithDecryptedInformation)
             .toList();
     }

src/main/java/com/michelin/ns4kafka/service/ConnectorService.java

@@ -62,7 +62,7 @@ public List<Connector> findAllForNamespace(Namespace namespace) {
             .findResourceOwnerGrantedToNamespace(namespace, AccessControlEntry.ResourceType.CONNECT);
         return connectorRepository.findAllForCluster(namespace.getMetadata().getCluster())
             .stream()
-            .filter(connector -> aclService.isAnyAclOfResource(acls, connector.getMetadata().getName()))
+            .filter(connector -> aclService.isResourceCoveredByAcls(acls, connector.getMetadata().getName()))
             .toList();
     }
 
@@ -74,10 +74,11 @@ public List<Connector> findAllForNamespace(Namespace namespace) {
      * @return A list of connectors
      */
     public List<Connector> findByWildcardName(Namespace namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespace(namespace)
             .stream()
-            .filter(connector -> RegexUtils.filterByPattern(connector.getMetadata().getName(), nameFilterPatterns))
+            .filter(connector -> RegexUtils
+                .isResourceCoveredByRegex(connector.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/service/NamespaceService.java

@@ -73,10 +73,10 @@ public List<Namespace> findAll() {
      * @return The list of namespaces
      */
     public List<Namespace> findByWildcardName(String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAll()
             .stream()
-            .filter(ns -> RegexUtils.filterByPattern(ns.getMetadata().getName(), nameFilterPatterns))
+            .filter(ns -> RegexUtils.isResourceCoveredByRegex(ns.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/service/ResourceQuotaService.java

@@ -72,10 +72,10 @@ public Optional<ResourceQuota> findForNamespace(String namespace) {
      * @return The researched resource quota
      */
     public List<ResourceQuota> findByWildcardName(String namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findForNamespace(namespace)
             .stream()
-            .filter(quota -> RegexUtils.filterByPattern(quota.getMetadata().getName(), nameFilterPatterns))
+            .filter(quota -> RegexUtils.isResourceCoveredByRegex(quota.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/service/RoleBindingService.java

@@ -35,10 +35,10 @@ public List<RoleBinding> findAllForNamespace(String namespace) {
      * @return The list of associated role bindings
      */
     public List<RoleBinding> findByWildcardName(String namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespace(namespace)
             .stream()
-            .filter(rb -> RegexUtils.filterByPattern(rb.getMetadata().getName(), nameFilterPatterns))
+            .filter(rb -> RegexUtils.isResourceCoveredByRegex(rb.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/service/SchemaService.java

@@ -55,7 +55,7 @@ public Flux<SchemaList> findAllForNamespace(Namespace namespace) {
             .getSubjects(namespace.getMetadata().getCluster())
             .filter(subject -> {
                 String underlyingTopicName = subject.replaceAll("-(key|value)$", "");
-                return aclService.isAnyAclOfResource(acls, underlyingTopicName);
+                return aclService.isResourceCoveredByAcls(acls, underlyingTopicName);
             })
             .map(subject -> SchemaList.builder()
                 .metadata(Metadata.builder()
@@ -74,9 +74,10 @@ public Flux<SchemaList> findAllForNamespace(Namespace namespace) {
      * @return A list of schemas
      */
     public Flux<SchemaList> findByWildcardName(Namespace namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespace(namespace)
-            .filter(schemaList -> RegexUtils.filterByPattern(schemaList.getMetadata().getName(), nameFilterPatterns));
+            .filter(schemaList -> RegexUtils
+                .isResourceCoveredByRegex(schemaList.getMetadata().getName(), nameFilterPatterns));
     }
 
     /**

src/main/java/com/michelin/ns4kafka/service/StreamService.java

@@ -48,10 +48,10 @@ public List<KafkaStream> findAllForNamespace(Namespace namespace) {
      * @return A list of Kafka Streams
      */
     public List<KafkaStream> findByWildcardName(Namespace namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespace(namespace)
             .stream()
-            .filter(stream -> RegexUtils.filterByPattern(stream.getMetadata().getName(), nameFilterPatterns))
+            .filter(stream -> RegexUtils.isResourceCoveredByRegex(stream.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/service/TopicService.java

@@ -68,7 +68,7 @@ public List<Topic> findAllForNamespace(Namespace namespace) {
             .findResourceOwnerGrantedToNamespace(namespace, AccessControlEntry.ResourceType.TOPIC);
         return topicRepository.findAllForCluster(namespace.getMetadata().getCluster())
             .stream()
-            .filter(topic -> aclService.isAnyAclOfResource(acls, topic.getMetadata().getName()))
+            .filter(topic -> aclService.isResourceCoveredByAcls(acls, topic.getMetadata().getName()))
             .toList();
     }
 
@@ -80,10 +80,10 @@ public List<Topic> findAllForNamespace(Namespace namespace) {
      * @return A list of topics
      */
     public List<Topic> findByWildcardName(Namespace namespace, String name) {
-        List<String> nameFilterPatterns = RegexUtils.wildcardStringsToRegexPatterns(List.of(name));
+        List<String> nameFilterPatterns = RegexUtils.convertWildcardStringsToRegex(List.of(name));
         return findAllForNamespace(namespace)
             .stream()
-            .filter(topic -> RegexUtils.filterByPattern(topic.getMetadata().getName(), nameFilterPatterns))
+            .filter(topic -> RegexUtils.isResourceCoveredByRegex(topic.getMetadata().getName(), nameFilterPatterns))
             .toList();
     }
 

src/main/java/com/michelin/ns4kafka/util/RegexUtils.java

@@ -16,8 +16,9 @@ public class RegexUtils {
      * @param wildcardStrings The list of wildcard strings
      * @return A list of regex patterns
      */
-    public static List<String> wildcardStringsToRegexPatterns(List<String> wildcardStrings) {
-        return wildcardStrings.stream()
+    public static List<String> convertWildcardStringsToRegex(List<String> wildcardStrings) {
+        return wildcardStrings
+            .stream()
             .map(wildcardString -> "^" + wildcardString
                 .replace(".", "\\.")
                 .replace("*", ".*")
@@ -33,8 +34,9 @@ public static List<String> wildcardStringsToRegexPatterns(List<String> wildcardS
      * @param regexPatterns The regex patterns
      * @return true if any regex pattern matches the resourceName, false otherwise
      */
-    public static boolean filterByPattern(String resourceName, List<String> regexPatterns) {
-        return regexPatterns.stream()
+    public static boolean isResourceCoveredByRegex(String resourceName, List<String> regexPatterns) {
+        return regexPatterns
+            .stream()
             .anyMatch(pattern -> Pattern.compile(pattern).matcher(resourceName).matches());
     }
 }