KAFKA-7011 - Remove ResourceNameType field from Java Resource class.

clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java

@@ -19,48 +19,60 @@
 
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.resource.Resource;
+import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePattern;
 
 import java.util.Objects;
 
 /**
- * Represents a binding between a resource and an access control entry.
+ * Represents a binding between a resource pattern and an access control entry.
  *
  * The API for this class is still evolving and we may break compatibility in minor releases, if necessary.
  */
 @InterfaceStability.Evolving
 public class AclBinding {
-    private final Resource resource;
+    private final ResourcePattern pattern;
     private final AccessControlEntry entry;
 
+    /**
+     * Create an instance of this class with the provided parameters.
+     *
+     * @param pattern non-null resource pattern.
+     * @param entry non-null entry
+     */
+    public AclBinding(ResourcePattern pattern, AccessControlEntry entry) {
+        this.pattern = Objects.requireNonNull(pattern, "pattern");
+        this.entry = Objects.requireNonNull(entry, "entry");
+    }
+
     /**
      * Create an instance of this class with the provided parameters.
      *
      * @param resource non-null resource
      * @param entry non-null entry
+     * @deprecated Since 2.0. Use {@link #AclBinding(ResourcePattern, AccessControlEntry)}
      */
+    @Deprecated
     public AclBinding(Resource resource, AccessControlEntry entry) {
-        Objects.requireNonNull(resource);
-        this.resource = resource;
-        Objects.requireNonNull(entry);
-        this.entry = entry;
+        this(new ResourcePattern(resource.resourceType(), resource.name(), ResourceNameType.LITERAL), entry);
     }
 
     /**
-     * Return true if this binding has any UNKNOWN components.
+     * @return true if this binding has any UNKNOWN components.
      */
     public boolean isUnknown() {
-        return resource.isUnknown() || entry.isUnknown();
+        return pattern.isUnknown() || entry.isUnknown();
     }
 
     /**
-     * Return the resource for this binding.
+     * @return the resource pattern for this binding.
      */
-    public Resource resource() {
-        return resource;
+    public ResourcePattern pattern() {
+        return pattern;
     }
 
     /**
-     * Return the access control entry for this binding.
+     * @return the access control entry for this binding.
      */
     public final AccessControlEntry entry() {
         return entry;
@@ -70,24 +82,25 @@ public final AccessControlEntry entry() {
      * Create a filter which matches only this AclBinding.
      */
     public AclBindingFilter toFilter() {
-        return new AclBindingFilter(resource.toFilter(), entry.toFilter());
+        return new AclBindingFilter(pattern.toFilter(), entry.toFilter());
     }
 
     @Override
     public String toString() {
-        return "(resource=" + resource + ", entry=" + entry + ")";
+        return "(pattern=" + pattern + ", entry=" + entry + ")";
     }
 
     @Override
     public boolean equals(Object o) {
-        if (!(o instanceof AclBinding))
-            return false;
-        AclBinding other = (AclBinding) o;
-        return resource.equals(other.resource) && entry.equals(other.entry);
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        AclBinding that = (AclBinding) o;
+        return Objects.equals(pattern, that.pattern) &&
+            Objects.equals(entry, that.entry);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(resource, entry);
+        return Objects.hash(pattern, entry);
     }
 }

clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java

@@ -19,6 +19,8 @@
 
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.resource.ResourceFilter;
+import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
 
 import java.util.Objects;
 
@@ -29,74 +31,85 @@
  */
 @InterfaceStability.Evolving
 public class AclBindingFilter {
-    private final ResourceFilter resourceFilter;
+    private final ResourcePatternFilter patternFilter;
     private final AccessControlEntryFilter entryFilter;
 
     /**
      * A filter which matches any ACL binding.
      */
-    public static final AclBindingFilter ANY = new AclBindingFilter(ResourceFilter.ANY, AccessControlEntryFilter.ANY);
+    public static final AclBindingFilter ANY = new AclBindingFilter(ResourcePatternFilter.ANY, AccessControlEntryFilter.ANY);
+
+    /**
+     * Create an instance of this filter with the provided parameters.
+     *
+     * @param patternFilter non-null pattern filter
+     * @param entryFilter non-null access control entry filter
+     */
+    public AclBindingFilter(ResourcePatternFilter patternFilter, AccessControlEntryFilter entryFilter) {
+        this.patternFilter = Objects.requireNonNull(patternFilter, "patternFilter");
+        this.entryFilter = Objects.requireNonNull(entryFilter, "entryFilter");
+    }
 
     /**
      * Create an instance of this filter with the provided parameters.
      *
      * @param resourceFilter non-null resource filter
      * @param entryFilter non-null access control entry filter
+     * @deprecated Since 2.0. Use {@link #AclBindingFilter(ResourcePatternFilter, AccessControlEntryFilter)}
      */
+    @Deprecated
     public AclBindingFilter(ResourceFilter resourceFilter, AccessControlEntryFilter entryFilter) {
-        Objects.requireNonNull(resourceFilter);
-        this.resourceFilter = resourceFilter;
-        Objects.requireNonNull(entryFilter);
-        this.entryFilter = entryFilter;
+        this(new ResourcePatternFilter(resourceFilter.resourceType(), resourceFilter.name(), ResourceNameType.LITERAL), entryFilter);
     }
 
     /**
-     * Return true if this filter has any UNKNOWN components.
+     * @return {@code true} if this filter has any UNKNOWN components.
      */
     public boolean isUnknown() {
-        return resourceFilter.isUnknown() || entryFilter.isUnknown();
+        return patternFilter.isUnknown() || entryFilter.isUnknown();
     }
 
     /**
-     * Return the resource filter.
+     * @return the resource pattern filter.
      */
-    public ResourceFilter resourceFilter() {
-        return resourceFilter;
+    public ResourcePatternFilter patternFilter() {
+        return patternFilter;
     }
 
     /**
-     * Return the access control entry filter.
+     * @return the access control entry filter.
      */
     public final AccessControlEntryFilter entryFilter() {
         return entryFilter;
     }
 
     @Override
     public String toString() {
-        return "(resourceFilter=" + resourceFilter + ", entryFilter=" + entryFilter + ")";
+        return "(patternFilter=" + patternFilter + ", entryFilter=" + entryFilter + ")";
     }
 
     @Override
     public boolean equals(Object o) {
-        if (!(o instanceof AclBindingFilter))
-            return false;
-        AclBindingFilter other = (AclBindingFilter) o;
-        return resourceFilter.equals(other.resourceFilter) && entryFilter.equals(other.entryFilter);
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        AclBindingFilter that = (AclBindingFilter) o;
+        return Objects.equals(patternFilter, that.patternFilter) &&
+            Objects.equals(entryFilter, that.entryFilter);
     }
 
     /**
      * Return true if the resource and entry filters can only match one ACE. In other words, if
      * there are no ANY or UNKNOWN fields.
      */
     public boolean matchesAtMostOne() {
-        return resourceFilter.matchesAtMostOne() && entryFilter.matchesAtMostOne();
+        return patternFilter.matchesAtMostOne() && entryFilter.matchesAtMostOne();
     }
 
     /**
      * Return a string describing an ANY or UNKNOWN field, or null if there is no such field.
      */
     public String findIndefiniteField() {
-        String indefinite = resourceFilter.findIndefiniteField();
+        String indefinite = patternFilter.findIndefiniteField();
         if (indefinite != null)
             return indefinite;
         return entryFilter.findIndefiniteField();
@@ -106,11 +119,11 @@ public String findIndefiniteField() {
      * Return true if the resource filter matches the binding's resource and the entry filter matches binding's entry.
      */
     public boolean matches(AclBinding binding) {
-        return resourceFilter.matches(binding.resource()) && entryFilter.matches(binding.entry());
+        return patternFilter.matches(binding.pattern()) && entryFilter.matches(binding.entry());
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(resourceFilter, entryFilter);
+        return Objects.hash(patternFilter, entryFilter);
     }
 }

clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java

@@ -19,13 +19,13 @@
 
 import org.apache.kafka.common.acl.AccessControlEntry;
 import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.types.ArrayOf;
 import org.apache.kafka.common.protocol.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
-import org.apache.kafka.common.resource.Resource;
 import org.apache.kafka.common.resource.ResourceNameType;
 import org.apache.kafka.common.utils.Utils;
 
@@ -81,17 +81,17 @@ public AclCreation(AclBinding acl) {
         }
 
         static AclCreation fromStruct(Struct struct) {
-            Resource resource = RequestUtils.resourceFromStructFields(struct);
+            ResourcePattern pattern = RequestUtils.resourcePatternromStructFields(struct);
             AccessControlEntry entry = RequestUtils.aceFromStructFields(struct);
-            return new AclCreation(new AclBinding(resource, entry));
+            return new AclCreation(new AclBinding(pattern, entry));
         }
 
         public AclBinding acl() {
             return acl;
         }
 
         void setStructFields(Struct struct) {
-            RequestUtils.resourceSetStructFields(acl.resource(), struct);
+            RequestUtils.resourcePatternSetStructFields(acl.pattern(), struct);
             RequestUtils.aceSetStructFields(acl.entry(), struct);
         }
 
@@ -179,12 +179,19 @@ private void validate(List<AclCreation> aclCreations) {
         if (version() == 0) {
             final boolean unsupported = aclCreations.stream()
                 .map(AclCreation::acl)
-                .map(AclBinding::resource)
-                .map(Resource::nameType)
+                .map(AclBinding::pattern)
+                .map(ResourcePattern::nameType)
                 .anyMatch(nameType -> nameType != ResourceNameType.LITERAL);
             if (unsupported) {
                 throw new UnsupportedVersionException("Version 0 only supports literal resource name types");
             }
         }
+
+        final boolean unknown = aclCreations.stream()
+            .map(AclCreation::acl)
+            .anyMatch(AclBinding::isUnknown);
+        if (unknown) {
+            throw new IllegalArgumentException("You can not create ACL bindings with unknown elements");
+        }
     }
 }

clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java

@@ -24,8 +24,8 @@
 import org.apache.kafka.common.protocol.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
-import org.apache.kafka.common.resource.ResourceFilter;
 import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.utils.Utils;
 
 import java.nio.ByteBuffer;
@@ -107,7 +107,7 @@ public DeleteAclsRequest(Struct struct, short version) {
         this.filters = new ArrayList<>();
         for (Object filterStructObj : struct.getArray(FILTERS)) {
             Struct filterStruct = (Struct) filterStructObj;
-            ResourceFilter resourceFilter = RequestUtils.resourceFilterFromStructFields(filterStruct);
+            ResourcePatternFilter resourceFilter = RequestUtils.resourcePatternFilterFromStructFields(filterStruct);
             AccessControlEntryFilter aceFilter = RequestUtils.aceFilterFromStructFields(filterStruct);
             this.filters.add(new AclBindingFilter(resourceFilter, aceFilter));
         }
@@ -123,7 +123,7 @@ protected Struct toStruct() {
         List<Struct> filterStructs = new ArrayList<>();
         for (AclBindingFilter filter : filters) {
             Struct filterStruct = struct.instance(FILTERS);
-            RequestUtils.resourceFilterSetStructFields(filter.resourceFilter(), filterStruct);
+            RequestUtils.resourcePatternFilterSetStructFields(filter.patternFilter(), filterStruct);
             RequestUtils.aceFilterSetStructFields(filter.entryFilter(), filterStruct);
             filterStructs.add(filterStruct);
         }
@@ -156,12 +156,17 @@ public static DeleteAclsRequest parse(ByteBuffer buffer, short version) {
     private void validate(short version, List<AclBindingFilter> filters) {
         if (version == 0) {
             final boolean unsupported = filters.stream()
-                .map(AclBindingFilter::resourceFilter)
-                .map(ResourceFilter::nameType)
+                .map(AclBindingFilter::patternFilter)
+                .map(ResourcePatternFilter::nameType)
                 .anyMatch(nameType -> nameType != ResourceNameType.LITERAL);
             if (unsupported) {
                 throw new UnsupportedVersionException("Version 0 only supports literal resource name types");
             }
         }
+
+        final boolean unknown = filters.stream().anyMatch(AclBindingFilter::isUnknown);
+        if (unknown) {
+            throw new IllegalArgumentException("Filters contain UNKNOWN elements");
+        }
     }
 }

clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java

@@ -18,14 +18,14 @@
 
 import org.apache.kafka.common.acl.AccessControlEntry;
 import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.types.ArrayOf;
 import org.apache.kafka.common.protocol.types.Field;
 import org.apache.kafka.common.protocol.types.Schema;
 import org.apache.kafka.common.protocol.types.Struct;
-import org.apache.kafka.common.resource.Resource;
 import org.apache.kafka.common.resource.ResourceNameType;
 import org.apache.kafka.common.utils.Utils;
 import org.slf4j.Logger;
@@ -180,7 +180,7 @@ public DeleteAclsResponse(Struct struct) {
                 Struct matchingAclStruct = (Struct) matchingAclStructObj;
                 ApiError matchError = new ApiError(matchingAclStruct);
                 AccessControlEntry entry = RequestUtils.aceFromStructFields(matchingAclStruct);
-                Resource resource = RequestUtils.resourceFromStructFields(matchingAclStruct);
+                ResourcePattern resource = RequestUtils.resourcePatternromStructFields(matchingAclStruct);
                 deletions.add(new AclDeletionResult(matchError, new AclBinding(resource, entry)));
             }
             this.responses.add(new AclFilterResponse(error, deletions));
@@ -201,7 +201,7 @@ protected Struct toStruct(short version) {
             for (AclDeletionResult deletion : response.deletions()) {
                 Struct deletionStruct = responseStruct.instance(MATCHING_ACLS_KEY_NAME);
                 deletion.error.write(deletionStruct);
-                RequestUtils.resourceSetStructFields(deletion.acl().resource(), deletionStruct);
+                RequestUtils.resourcePatternSetStructFields(deletion.acl().pattern(), deletionStruct);
                 RequestUtils.aceSetStructFields(deletion.acl().entry(), deletionStruct);
                 deletionStructs.add(deletionStruct);
             }
@@ -247,12 +247,20 @@ private void validate(short version) {
             final boolean unsupported = responses.stream()
                 .flatMap(r -> r.deletions.stream())
                 .map(AclDeletionResult::acl)
-                .map(AclBinding::resource)
-                .map(Resource::nameType)
+                .map(AclBinding::pattern)
+                .map(ResourcePattern::nameType)
                 .anyMatch(nameType -> nameType != ResourceNameType.LITERAL);
             if (unsupported) {
                 throw new UnsupportedVersionException("Version 0 only supports literal resource name types");
             }
         }
+
+        final boolean unknown = responses.stream()
+            .flatMap(r -> r.deletions.stream())
+            .map(AclDeletionResult::acl)
+            .anyMatch(AclBinding::isUnknown);
+        if (unknown) {
+            throw new IllegalArgumentException("Response contains UNKNOWN elements");
+        }
     }
 }