KAFKA-7006 - remove duplicate Scala ResourceNameType in preference to…

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

@@ -39,10 +39,16 @@ public class AclBinding {
      * @param entry non-null entry
      */
     public AclBinding(Resource resource, AccessControlEntry entry) {
-        Objects.requireNonNull(resource);
-        this.resource = resource;
-        Objects.requireNonNull(entry);
-        this.entry = entry;
+        this.resource = Objects.requireNonNull(resource);
+        this.entry = Objects.requireNonNull(entry);
+
+        if (resource.resourceType().isNotSpecific()) {
+            throw new IllegalArgumentException("Resource type must be specific, not: " + resource.resourceType());
+        }
+
+        if (resource.nameType().isNotSpecific()) {
+            throw new IllegalArgumentException("Resource name type must be specific, not: " + resource.nameType());
+        }
     }
 
     /**

clients/src/main/java/org/apache/kafka/common/resource/ResourceNameType.java

@@ -82,6 +82,15 @@ public boolean isUnknown() {
         return this == UNKNOWN;
     }
 
+    /**
+     * Return whether this resource name type is ANY or UKNOWN.
+     *
+     * i.e. it is not a specific resource name type.
+     */
+    public boolean isNotSpecific() {
+        return this == ANY || this == UNKNOWN;
+    }
+
     /**
      * Return the ResourceNameType with the provided code or {@link #UNKNOWN} if one cannot be found.
      */

clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java

@@ -118,4 +118,13 @@ public byte code() {
     public boolean isUnknown() {
         return this == UNKNOWN;
     }
+
+    /**
+     * Return whether this resource name type is ANY or UNKNOWN.
+     *
+     * i.e. it is not a specific resource type.
+     */
+    public boolean isNotSpecific() {
+        return this == ANY || this == UNKNOWN;
+    }
 }

clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java

@@ -22,7 +22,9 @@
 import org.apache.kafka.common.resource.ResourceType;
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -56,29 +58,29 @@ public class AclBindingTest {
         new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY));
 
     @Test
-    public void testMatching() throws Exception {
-        assertTrue(ACL1.equals(ACL1));
+    public void testMatching() {
+        assertEquals(ACL1, ACL1);
         final AclBinding acl1Copy = new AclBinding(
             new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
             new AccessControlEntry("User:ANONYMOUS", "", AclOperation.ALL, AclPermissionType.ALLOW));
-        assertTrue(ACL1.equals(acl1Copy));
-        assertTrue(acl1Copy.equals(ACL1));
-        assertTrue(ACL2.equals(ACL2));
-        assertFalse(ACL1.equals(ACL2));
-        assertFalse(ACL2.equals(ACL1));
+        assertEquals(ACL1, acl1Copy);
+        assertEquals(acl1Copy, ACL1);
+        assertEquals(ACL2, ACL2);
+        assertNotEquals(ACL1, ACL2);
+        assertNotEquals(ACL2, ACL1);
         assertTrue(AclBindingFilter.ANY.matches(ACL1));
-        assertFalse(AclBindingFilter.ANY.equals(ACL1));
+        assertNotEquals(AclBindingFilter.ANY, ACL1);
         assertTrue(AclBindingFilter.ANY.matches(ACL2));
-        assertFalse(AclBindingFilter.ANY.equals(ACL2));
+        assertNotEquals(AclBindingFilter.ANY, ACL2);
         assertTrue(AclBindingFilter.ANY.matches(ACL3));
-        assertFalse(AclBindingFilter.ANY.equals(ACL3));
-        assertTrue(AclBindingFilter.ANY.equals(AclBindingFilter.ANY));
+        assertNotEquals(AclBindingFilter.ANY, ACL3);
+        assertEquals(AclBindingFilter.ANY, AclBindingFilter.ANY);
         assertTrue(ANY_ANONYMOUS.matches(ACL1));
-        assertFalse(ANY_ANONYMOUS.equals(ACL1));
+        assertNotEquals(ANY_ANONYMOUS, ACL1);
         assertFalse(ANY_ANONYMOUS.matches(ACL2));
-        assertFalse(ANY_ANONYMOUS.equals(ACL2));
+        assertNotEquals(ANY_ANONYMOUS, ACL2);
         assertTrue(ANY_ANONYMOUS.matches(ACL3));
-        assertFalse(ANY_ANONYMOUS.equals(ACL3));
+        assertNotEquals(ANY_ANONYMOUS, ACL3);
         assertFalse(ANY_DENY.matches(ACL1));
         assertFalse(ANY_DENY.matches(ACL2));
         assertTrue(ANY_DENY.matches(ACL3));
@@ -87,12 +89,12 @@ public void testMatching() throws Exception {
         assertFalse(ANY_MYTOPIC.matches(ACL3));
         assertTrue(ANY_ANONYMOUS.matches(UNKNOWN_ACL));
         assertTrue(ANY_DENY.matches(UNKNOWN_ACL));
-        assertTrue(UNKNOWN_ACL.equals(UNKNOWN_ACL));
+        assertEquals(UNKNOWN_ACL, UNKNOWN_ACL);
         assertFalse(ANY_MYTOPIC.matches(UNKNOWN_ACL));
     }
 
     @Test
-    public void testUnknowns() throws Exception {
+    public void testUnknowns() {
         assertFalse(ACL1.isUnknown());
         assertFalse(ACL2.isUnknown());
         assertFalse(ACL3.isUnknown());
@@ -103,12 +105,32 @@ public void testUnknowns() throws Exception {
     }
 
     @Test
-    public void testMatchesAtMostOne() throws Exception {
+    public void testMatchesAtMostOne() {
         assertNull(ACL1.toFilter().findIndefiniteField());
         assertNull(ACL2.toFilter().findIndefiniteField());
         assertNull(ACL3.toFilter().findIndefiniteField());
         assertFalse(ANY_ANONYMOUS.matchesAtMostOne());
         assertFalse(ANY_DENY.matchesAtMostOne());
         assertFalse(ANY_MYTOPIC.matchesAtMostOne());
     }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testThrowsOnUnknownResourceNameType() {
+        new AclBinding(new Resource(ResourceType.TOPIC, "foo", ResourceNameType.UNKNOWN), ACL1.entry());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testThrowsOnAnyResourceNameType() {
+        new AclBinding(new Resource(ResourceType.TOPIC, "foo", ResourceNameType.ANY), ACL1.entry());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testThrowsOnUnknownResourceType() {
+        new AclBinding(new Resource(ResourceType.UNKNOWN, "foo", ResourceNameType.LITERAL), ACL1.entry());
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testThrowsOnAnyResourceType() {
+        new AclBinding(new Resource(ResourceType.ANY, "foo", ResourceNameType.LITERAL), ACL1.entry());
+    }
 }

core/src/main/scala/kafka/admin/AclCommand.scala

@@ -25,13 +25,13 @@ import kafka.utils._
 import org.apache.kafka.common.security.JaasUtils
 import org.apache.kafka.common.security.auth.KafkaPrincipal
 import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType => JResourceNameType, ResourceType => JResourceType, Resource => JResource}
+import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType, ResourceType => JResourceType, Resource => JResource}
 
 import scala.collection.JavaConverters._
 
 object AclCommand extends Logging {
 
-  val ClusterResourceFilter = new ResourceFilter(JResourceType.CLUSTER, JResource.CLUSTER_NAME, JResourceNameType.LITERAL)
+  val ClusterResourceFilter = new ResourceFilter(JResourceType.CLUSTER, JResource.CLUSTER_NAME, ResourceNameType.LITERAL)
 
   private val Newline = scala.util.Properties.lineSeparator
 
@@ -87,13 +87,13 @@ object AclCommand extends Logging {
   }
 
   private def addAcl(opts: AclCommandOptions) {
-    if (opts.options.valueOf(opts.resourceNameType) == JResourceNameType.ANY)
+    if (opts.options.valueOf(opts.resourceNameType) == ResourceNameType.ANY)
       CommandLineUtils.printUsageAndDie(opts.parser, "A '--resource-name-type' value of 'Any' is not valid when adding acls.")
 
     withAuthorizer(opts) { authorizer =>
       val resourceToAcl = getResourceFilterToAcls(opts).map {
         case (filter, acls) =>
-          Resource(ResourceType.fromJava(filter.resourceType()), filter.name(), ResourceNameType.fromJava(filter.nameType())) -> acls
+          Resource(ResourceType.fromJava(filter.resourceType()), filter.name(), filter.nameType()) -> acls
       }
 
       if (resourceToAcl.values.exists(_.isEmpty))
@@ -262,13 +262,13 @@ object AclCommand extends Logging {
   }
 
   private def getResourceFilter(opts: AclCommandOptions, dieIfNoResourceFound: Boolean = true): Set[ResourceFilter] = {
-    val resourceNameType: JResourceNameType = opts.options.valueOf(opts.resourceNameType)
+    val resourceNameType: ResourceNameType = opts.options.valueOf(opts.resourceNameType)
 
     var resourceFilters = Set.empty[ResourceFilter]
     if (opts.options.has(opts.topicOpt))
       opts.options.valuesOf(opts.topicOpt).asScala.foreach(topic => resourceFilters += new ResourceFilter(JResourceType.TOPIC, topic.trim, resourceNameType))
 
-    if (resourceNameType == JResourceNameType.LITERAL && (opts.options.has(opts.clusterOpt) || opts.options.has(opts.idempotentOpt)))
+    if (resourceNameType == ResourceNameType.LITERAL && (opts.options.has(opts.clusterOpt) || opts.options.has(opts.idempotentOpt)))
       resourceFilters += ClusterResourceFilter
 
     if (opts.options.has(opts.groupOpt))
@@ -349,7 +349,7 @@ object AclCommand extends Logging {
       .withRequiredArg()
       .ofType(classOf[String])
       .withValuesConvertedBy(new ResourceNameTypeConverter())
-      .defaultsTo(JResourceNameType.LITERAL)
+      .defaultsTo(ResourceNameType.LITERAL)
 
     val addOpt = parser.accepts("add", "Indicates you are trying to add ACLs.")
     val removeOpt = parser.accepts("remove", "Indicates you are trying to remove ACLs.")
@@ -429,17 +429,17 @@ object AclCommand extends Logging {
 
 }
 
-class ResourceNameTypeConverter extends EnumConverter[JResourceNameType](classOf[JResourceNameType]) {
+class ResourceNameTypeConverter extends EnumConverter[ResourceNameType](classOf[ResourceNameType]) {
 
-  override def convert(value: String): JResourceNameType = {
+  override def convert(value: String): ResourceNameType = {
     val nameType = super.convert(value)
     if (nameType.isUnknown)
       throw new ValueConversionException("Unknown resourceNameType: " + value)
 
     nameType
   }
 
-  override def valuePattern: String = JResourceNameType.values
-    .filter(_ != JResourceNameType.UNKNOWN)
+  override def valuePattern: String = ResourceNameType.values
+    .filter(_ != ResourceNameType.UNKNOWN)
     .mkString("|")
 }

core/src/main/scala/kafka/security/SecurityUtils.scala

@@ -17,7 +17,7 @@
 
 package kafka.security
 
-import kafka.security.auth.{Acl, Operation, PermissionType, Resource, ResourceNameType, ResourceType}
+import kafka.security.auth.{Acl, Operation, PermissionType, Resource, ResourceType}
 import org.apache.kafka.common.acl.{AccessControlEntry, AclBinding, AclBindingFilter}
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.ApiError
@@ -32,11 +32,10 @@ object SecurityUtils {
   def convertToResourceAndAcl(filter: AclBindingFilter): Either[ApiError, (Resource, Acl)] = {
     (for {
       resourceType <- Try(ResourceType.fromJava(filter.resourceFilter.resourceType))
-      resourceNameType <- Try(ResourceNameType.fromJava(filter.resourceFilter.nameType))
       principal <- Try(KafkaPrincipal.fromString(filter.entryFilter.principal))
       operation <- Try(Operation.fromJava(filter.entryFilter.operation))
       permissionType <- Try(PermissionType.fromJava(filter.entryFilter.permissionType))
-      resource = Resource(resourceType, filter.resourceFilter.name, resourceNameType)
+      resource = Resource(resourceType, filter.resourceFilter.name, filter.resourceFilter.nameType)
       acl = Acl(principal, permissionType, filter.entryFilter.host, operation)
     } yield (resource, acl)) match {
       case Failure(throwable) => Left(new ApiError(Errors.INVALID_REQUEST, throwable.getMessage))
@@ -45,7 +44,7 @@ object SecurityUtils {
   }
 
   def convertToAclBinding(resource: Resource, acl: Acl): AclBinding = {
-    val adminResource = new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava)
+    val adminResource = new AdminResource(resource.resourceType.toJava, resource.name, resource.nameType)
     val entry = new AccessControlEntry(acl.principal.toString, acl.host.toString,
       acl.operation.toJava, acl.permissionType.toJava)
     new AclBinding(adminResource, entry)

core/src/main/scala/kafka/security/auth/Authorizer.scala

@@ -48,13 +48,13 @@ trait Authorizer extends Configurable {
    *
    * {code}
    * // The following will add ACLs to the literal resource path 'foo', which will only affect the topic named 'foo':
-   * authorizer.addAcls(Set(acl1, acl2), Resource(Topic, "foo", Literal))
+   * authorizer.addAcls(Set(acl1, acl2), Resource(Topic, "foo", LITERAL))
    *
    * // The following will add ACLs to the special literal topic resource path '*', which affects all topics:
-   * authorizer.addAcls(Set(acl1, acl2), Resource(Topic, "*", Literal))
+   * authorizer.addAcls(Set(acl1, acl2), Resource(Topic, "*", LITERAL))
    *
    * // The following will add ACLs to the prefixed resource path 'foo', which affects all topics whose name begins with 'foo':
-   * authorizer.addAcls(Set(acl1, acl2), Resource(Topic, "foo", Prefixed))
+   * authorizer.addAcls(Set(acl1, acl2), Resource(Topic, "foo", PREFIXED))
    * {code}
    *
    * @param acls set of acls to add to existing acls
@@ -67,13 +67,13 @@ trait Authorizer extends Configurable {
    *
    * {code}
    * // The following will remove ACLs from the literal resource path 'foo', which will only affect the topic named 'foo':
-   * authorizer.removeAcls(Set(acl1, acl2), Resource(Topic, "foo", Literal))
+   * authorizer.removeAcls(Set(acl1, acl2), Resource(Topic, "foo", LITERAL))
    *
    * // The following will remove ACLs from the special literal topic resource path '*', which affects all topics:
-   * authorizer.removeAcls(Set(acl1, acl2), Resource(Topic, "*", Literal))
+   * authorizer.removeAcls(Set(acl1, acl2), Resource(Topic, "*", LITERAL))
    *
    * // The following will remove ACLs from the prefixed resource path 'foo', which affects all topics whose name begins with 'foo':
-   * authorizer.removeAcls(Set(acl1, acl2), Resource(Topic, "foo", Prefixed))
+   * authorizer.removeAcls(Set(acl1, acl2), Resource(Topic, "foo", PREFIXED))
    * {code}
    *
    * @param acls set of acls to be removed.
@@ -87,13 +87,13 @@ trait Authorizer extends Configurable {
    *
    * {code}
    * // The following will remove all ACLs from the literal resource path 'foo', which will only affect the topic named 'foo':
-   * authorizer.removeAcls(Resource(Topic, "foo", Literal))
+   * authorizer.removeAcls(Resource(Topic, "foo", LITERAL))
    *
    * // The following will remove all ACLs from the special literal topic resource path '*', which affects all topics:
-   * authorizer.removeAcls(Resource(Topic, "*", Literal))
+   * authorizer.removeAcls(Resource(Topic, "*", LITERAL))
    *
    * // The following will remove all ACLs from the prefixed resource path 'foo', which affects all topics whose name begins with 'foo':
-   * authorizer.removeAcls(Resource(Topic, "foo", Prefixed))
+   * authorizer.removeAcls(Resource(Topic, "foo", PREFIXED))
    * {code}
    *
    * @param resource the resource path from which these acls should be removed.
@@ -106,13 +106,13 @@ trait Authorizer extends Configurable {
    *
    * {code}
    * // The following will get all ACLs from the literal resource path 'foo', which will only affect the topic named 'foo':
-   * authorizer.removeAcls(Resource(Topic, "foo", Literal))
+   * authorizer.removeAcls(Resource(Topic, "foo", LITERAL))
    *
    * // The following will get all ACLs from the special literal topic resource path '*', which affects all topics:
-   * authorizer.removeAcls(Resource(Topic, "*", Literal))
+   * authorizer.removeAcls(Resource(Topic, "*", LITERAL))
    *
    * // The following will get all ACLs from the prefixed resource path 'foo', which affects all topics whose name begins with 'foo':
-   * authorizer.removeAcls(Resource(Topic, "foo", Prefixed))
+   * authorizer.removeAcls(Resource(Topic, "foo", PREFIXED))
    * {code}
    *
    * @param resource the resource path to which the acls belong.

core/src/main/scala/kafka/security/auth/Resource.scala

@@ -17,11 +17,11 @@
 package kafka.security.auth
 
 import java.util.Objects
-import org.apache.kafka.common.resource.{Resource => JResource}
+import org.apache.kafka.common.resource.{ResourceNameType, Resource => JResource}
 
 object Resource {
   val ClusterResourceName = "kafka-cluster"
-  val ClusterResource = new Resource(Cluster, Resource.ClusterResourceName, Literal)
+  val ClusterResource = Resource(Cluster, Resource.ClusterResourceName, ResourceNameType.LITERAL)
   val ProducerIdResourceName = "producer-id"
   val WildCardResource = "*"
 }
@@ -31,13 +31,17 @@ object Resource {
  * @param resourceType non-null type of resource.
  * @param name non-null name of the resource, for topic this will be topic name , for group it will be group name. For cluster type
  *             it will be a constant string kafka-cluster.
- * @param resourceNameType non-null type of resource name: literal, prefixed, etc.
+ * @param nameType non-null type of resource name: literal, prefixed, etc.
  */
-case class Resource(resourceType: ResourceType, name: String, resourceNameType: ResourceNameType) {
+case class Resource(resourceType: ResourceType, name: String, nameType: ResourceNameType) {
 
   Objects.requireNonNull(resourceType, "resourceType")
   Objects.requireNonNull(name, "name")
-  Objects.requireNonNull(resourceNameType, "resourceNameType")
+  Objects.requireNonNull(nameType, "nameType")
+
+  if (nameType.isNotSpecific) {
+    throw new IllegalArgumentException("nameType must be a specific type, not: " + nameType)
+  }
 
   /**
     * Create an instance of this class with the provided parameters.
@@ -49,11 +53,11 @@ case class Resource(resourceType: ResourceType, name: String, resourceNameType:
     */
   @deprecated("Use Resource(ResourceType, String, ResourceNameType")
   def this(resourceType: ResourceType, name: String) {
-    this(resourceType, name, Literal)
+    this(resourceType, name, ResourceNameType.LITERAL)
   }
 
   def toJava: JResource = {
-    new JResource(resourceType.toJava, name, resourceNameType.toJava)
+    new JResource(resourceType.toJava, name, nameType)
   }
 }