You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2018/06/07 23:37:27 UTC
[kafka] branch 2.0 updated: KAFKA-7011 - Remove ResourceNameType
field from Java Resource class. (#5160)
This is an automated email from the ASF dual-hosted git repository.
junrao pushed a commit to branch 2.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.0 by this push:
new 38bc31a KAFKA-7011 - Remove ResourceNameType field from Java Resource class. (#5160)
38bc31a is described below
commit 38bc31a77b897a3f248878e72d731c4767e1f15b
Author: Andy Coates <80...@users.noreply.github.com>
AuthorDate: Fri Jun 8 00:35:30 2018 +0100
KAFKA-7011 - Remove ResourceNameType field from Java Resource class. (#5160)
The initial PR for KIP-290 #5117 added a `ResourceNameType` field to the Java and Scala `Resource` classes to introduce the concept of Prefixed ACLS. This does not make a lot of sense as these classes are meant to represent cluster resources, which would not have a concept of 'name type'. This work has not been released yet, so we have time to change it.
This PR looks to refactor the code to remove the name type field from the Java `Resource` class. (The Scala one will age out once KIP-290 is done, and removing it would involve changes to the `Authorizer` interface, so this class was not touched).
This is achieved by replacing the use of `Resource` with `ResourcePattern` and `ResourceFilter` with `ResourceFilterPattern`. A `ResourcePattern` is a combination of resource type, name and name type, where each field needs to be defined. A `ResourcePatternFilter` is used to select patterns during describe and delete operations.
The adminClient uses `AclBinding` and `AclBindingFilter`. These types have been switched over to use the new pattern types.
The AclCommands class, used by Kafka-acls.sh, has been converted to use the new pattern types.
The result is that the original `Resource` and `ResourceFilter` classes are not really used anywhere, except deprecated methods. However, the `Resource` class will be used if/when KIP-50 is done.
Reviewers: Colin Patrick McCabe <co...@cmccabe.xyz>, Jun Rao <ju...@gmail.com>
---
.../org/apache/kafka/common/acl/AclBinding.java | 51 ++++---
.../apache/kafka/common/acl/AclBindingFilter.java | 55 ++++---
.../kafka/common/requests/CreateAclsRequest.java | 19 ++-
.../kafka/common/requests/DeleteAclsRequest.java | 15 +-
.../kafka/common/requests/DeleteAclsResponse.java | 18 ++-
.../kafka/common/requests/DescribeAclsRequest.java | 15 +-
.../common/requests/DescribeAclsResponse.java | 32 ++--
.../apache/kafka/common/requests/RequestUtils.java | 30 ++--
.../org/apache/kafka/common/resource/Resource.java | 57 ++-----
.../kafka/common/resource/ResourceFilter.java | 114 ++------------
.../{Resource.java => ResourcePattern.java} | 72 ++++-----
.../common/resource/ResourcePatternFilter.java | 169 +++++++++++++++++++++
.../kafka/clients/admin/KafkaAdminClientTest.java | 12 +-
.../apache/kafka/common/acl/AclBindingTest.java | 20 +--
.../common/acl/ResourcePatternFilterTest.java | 164 ++++++++++++++++++++
.../kafka/common/acl/ResourcePatternTest.java | 40 +++++
.../common/requests/CreateAclsRequestTest.java | 16 +-
.../common/requests/DeleteAclsRequestTest.java | 16 +-
.../common/requests/DeleteAclsResponseTest.java | 18 ++-
.../common/requests/DescribeAclsRequestTest.java | 16 +-
.../common/requests/DescribeAclsResponseTest.java | 16 +-
.../kafka/common/requests/RequestResponseTest.java | 20 +--
.../kafka/common/resource/ResourceFilterTest.java | 123 +++------------
core/src/main/scala/kafka/admin/AclCommand.scala | 48 +++---
.../main/scala/kafka/security/SecurityUtils.scala | 12 +-
core/src/main/scala/kafka/security/auth/Acl.scala | 3 +-
.../main/scala/kafka/security/auth/Resource.scala | 35 +++--
.../kafka/security/auth/SimpleAclAuthorizer.scala | 9 +-
core/src/main/scala/kafka/server/KafkaApis.scala | 6 +-
core/src/main/scala/kafka/zk/KafkaZkClient.scala | 2 +-
core/src/main/scala/kafka/zk/ZkData.scala | 2 +-
.../kafka/api/AdminClientIntegrationTest.scala | 4 +-
.../kafka/api/AuthorizerIntegrationTest.scala | 8 +-
.../api/SaslSslAdminClientIntegrationTest.scala | 62 ++++----
.../scala/kafka/security/auth/ResourceTest.scala | 61 ++++++++
.../scala/unit/kafka/server/RequestQuotaTest.scala | 7 +-
36 files changed, 849 insertions(+), 518 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java b/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java
index d264ef1..feba875 100644
--- a/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java
+++ b/clients/src/main/java/org/apache/kafka/common/acl/AclBinding.java
@@ -19,48 +19,60 @@ package org.apache.kafka.common.acl;
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 class AclBinding {
* 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);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java b/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java
index 5841b5a..9d13fa7 100644
--- a/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java
+++ b/clients/src/main/java/org/apache/kafka/common/acl/AclBindingFilter.java
@@ -19,6 +19,8 @@ package org.apache.kafka.common.acl;
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,43 +31,53 @@ import java.util.Objects;
*/
@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;
@@ -73,15 +85,16 @@ public class AclBindingFilter {
@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);
}
/**
@@ -89,14 +102,14 @@ public class AclBindingFilter {
* 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 class AclBindingFilter {
* 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);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java
index 00f65c6..6782f78 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java
@@ -19,13 +19,13 @@ package org.apache.kafka.common.requests;
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,9 +81,9 @@ public class CreateAclsRequest extends AbstractRequest {
}
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() {
@@ -91,7 +91,7 @@ public class CreateAclsRequest extends AbstractRequest {
}
void setStructFields(Struct struct) {
- RequestUtils.resourceSetStructFields(acl.resource(), struct);
+ RequestUtils.resourcePatternSetStructFields(acl.pattern(), struct);
RequestUtils.aceSetStructFields(acl.entry(), struct);
}
@@ -179,12 +179,19 @@ public class CreateAclsRequest extends AbstractRequest {
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");
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java
index d896bb2..9bb15a3 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsRequest.java
@@ -24,8 +24,8 @@ 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.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 class DeleteAclsRequest extends AbstractRequest {
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 @@ public class DeleteAclsRequest extends AbstractRequest {
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 class DeleteAclsRequest extends AbstractRequest {
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");
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
index 1790457..112c6a3 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
@@ -18,6 +18,7 @@ package org.apache.kafka.common.requests;
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;
@@ -25,7 +26,6 @@ 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 class DeleteAclsResponse extends AbstractResponse {
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 @@ public class DeleteAclsResponse extends AbstractResponse {
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 @@ public class DeleteAclsResponse extends AbstractResponse {
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");
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java
index d3a04d0..5096360 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsRequest.java
@@ -17,14 +17,13 @@
package org.apache.kafka.common.requests;
import org.apache.kafka.common.acl.AccessControlEntryFilter;
-import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.ApiKeys;
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 java.nio.ByteBuffer;
import java.util.Collections;
@@ -95,7 +94,7 @@ public class DescribeAclsRequest extends AbstractRequest {
public DescribeAclsRequest(Struct struct, short version) {
super(version);
- ResourceFilter resourceFilter = RequestUtils.resourceFilterFromStructFields(struct);
+ ResourcePatternFilter resourceFilter = RequestUtils.resourcePatternFilterFromStructFields(struct);
AccessControlEntryFilter entryFilter = RequestUtils.aceFilterFromStructFields(struct);
this.filter = new AclBindingFilter(resourceFilter, entryFilter);
}
@@ -103,7 +102,7 @@ public class DescribeAclsRequest extends AbstractRequest {
@Override
protected Struct toStruct() {
Struct struct = new Struct(ApiKeys.DESCRIBE_ACLS.requestSchema(version()));
- RequestUtils.resourceFilterSetStructFields(filter.resourceFilter(), struct);
+ RequestUtils.resourcePatternFilterSetStructFields(filter.patternFilter(), struct);
RequestUtils.aceFilterSetStructFields(filter.entryFilter(), struct);
return struct;
}
@@ -115,7 +114,7 @@ public class DescribeAclsRequest extends AbstractRequest {
case 0:
case 1:
return new DescribeAclsResponse(throttleTimeMs, ApiError.fromThrowable(throwable),
- Collections.<AclBinding>emptySet());
+ Collections.emptySet());
default:
throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
versionId, this.getClass().getSimpleName(), ApiKeys.DESCRIBE_ACLS.latestVersion()));
@@ -131,8 +130,12 @@ public class DescribeAclsRequest extends AbstractRequest {
}
private void validate(AclBindingFilter filter, short version) {
- if (version == 0 && filter.resourceFilter().nameType() != ResourceNameType.LITERAL) {
+ if (version == 0 && filter.patternFilter().nameType() != ResourceNameType.LITERAL) {
throw new UnsupportedVersionException("Version 0 only supports literal resource name types");
}
+
+ if (filter.isUnknown()) {
+ throw new IllegalArgumentException("Filter contain UNKNOWN elements");
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
index b6673d9..66f2895 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
@@ -19,6 +19,7 @@ package org.apache.kafka.common.requests;
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;
@@ -26,7 +27,6 @@ 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 java.nio.ByteBuffer;
@@ -113,11 +113,11 @@ public class DescribeAclsResponse extends AbstractResponse {
this.acls = new ArrayList<>();
for (Object resourceStructObj : struct.getArray(RESOURCES_KEY_NAME)) {
Struct resourceStruct = (Struct) resourceStructObj;
- Resource resource = RequestUtils.resourceFromStructFields(resourceStruct);
+ ResourcePattern pattern = RequestUtils.resourcePatternromStructFields(resourceStruct);
for (Object aclDataStructObj : resourceStruct.getArray(ACLS_KEY_NAME)) {
Struct aclDataStruct = (Struct) aclDataStructObj;
AccessControlEntry entry = RequestUtils.aceFromStructFields(aclDataStruct);
- this.acls.add(new AclBinding(resource, entry));
+ this.acls.add(new AclBinding(pattern, entry));
}
}
}
@@ -130,21 +130,18 @@ public class DescribeAclsResponse extends AbstractResponse {
struct.set(THROTTLE_TIME_MS, throttleTimeMs);
error.write(struct);
- Map<Resource, List<AccessControlEntry>> resourceToData = new HashMap<>();
+ Map<ResourcePattern, List<AccessControlEntry>> resourceToData = new HashMap<>();
for (AclBinding acl : acls) {
- List<AccessControlEntry> entry = resourceToData.get(acl.resource());
- if (entry == null) {
- entry = new ArrayList<>();
- resourceToData.put(acl.resource(), entry);
- }
- entry.add(acl.entry());
+ resourceToData
+ .computeIfAbsent(acl.pattern(), k -> new ArrayList<>())
+ .add(acl.entry());
}
List<Struct> resourceStructs = new ArrayList<>();
- for (Map.Entry<Resource, List<AccessControlEntry>> tuple : resourceToData.entrySet()) {
- Resource resource = tuple.getKey();
+ for (Map.Entry<ResourcePattern, List<AccessControlEntry>> tuple : resourceToData.entrySet()) {
+ ResourcePattern resource = tuple.getKey();
Struct resourceStruct = struct.instance(RESOURCES_KEY_NAME);
- RequestUtils.resourceSetStructFields(resource, resourceStruct);
+ RequestUtils.resourcePatternSetStructFields(resource, resourceStruct);
List<Struct> dataStructs = new ArrayList<>();
for (AccessControlEntry entry : tuple.getValue()) {
Struct dataStruct = resourceStruct.instance(ACLS_KEY_NAME);
@@ -188,12 +185,17 @@ public class DescribeAclsResponse extends AbstractResponse {
private void validate(short version) {
if (version == 0) {
final boolean unsupported = acls.stream()
- .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 = acls.stream().anyMatch(AclBinding::isUnknown);
+ if (unknown) {
+ throw new IllegalArgumentException("Contain UNKNOWN elements");
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java b/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java
index f4f00a8..76b2707 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/RequestUtils.java
@@ -20,11 +20,11 @@ import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.protocol.types.Struct;
-import org.apache.kafka.common.resource.Resource;
-import org.apache.kafka.common.resource.ResourceFilter;
-import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourceType;
import static org.apache.kafka.common.protocol.CommonFields.HOST;
import static org.apache.kafka.common.protocol.CommonFields.HOST_FILTER;
@@ -42,39 +42,39 @@ final class RequestUtils {
private RequestUtils() {}
- static Resource resourceFromStructFields(Struct struct) {
+ static ResourcePattern resourcePatternromStructFields(Struct struct) {
byte resourceType = struct.get(RESOURCE_TYPE);
String name = struct.get(RESOURCE_NAME);
ResourceNameType resourceNameType = ResourceNameType.LITERAL;
if (struct.hasField(RESOURCE_NAME_TYPE)) {
resourceNameType = ResourceNameType.fromCode(struct.get(RESOURCE_NAME_TYPE));
}
- return new Resource(ResourceType.fromCode(resourceType), name, resourceNameType);
+ return new ResourcePattern(ResourceType.fromCode(resourceType), name, resourceNameType);
}
- static void resourceSetStructFields(Resource resource, Struct struct) {
- struct.set(RESOURCE_TYPE, resource.resourceType().code());
- struct.set(RESOURCE_NAME, resource.name());
+ static void resourcePatternSetStructFields(ResourcePattern pattern, Struct struct) {
+ struct.set(RESOURCE_TYPE, pattern.resourceType().code());
+ struct.set(RESOURCE_NAME, pattern.name());
if (struct.hasField(RESOURCE_NAME_TYPE)) {
- struct.set(RESOURCE_NAME_TYPE, resource.nameType().code());
+ struct.set(RESOURCE_NAME_TYPE, pattern.nameType().code());
}
}
- static ResourceFilter resourceFilterFromStructFields(Struct struct) {
+ static ResourcePatternFilter resourcePatternFilterFromStructFields(Struct struct) {
byte resourceType = struct.get(RESOURCE_TYPE);
String name = struct.get(RESOURCE_NAME_FILTER);
ResourceNameType resourceNameType = ResourceNameType.LITERAL;
if (struct.hasField(RESOURCE_NAME_TYPE_FILTER)) {
resourceNameType = ResourceNameType.fromCode(struct.get(RESOURCE_NAME_TYPE_FILTER));
}
- return new ResourceFilter(ResourceType.fromCode(resourceType), name, resourceNameType);
+ return new ResourcePatternFilter(ResourceType.fromCode(resourceType), name, resourceNameType);
}
- static void resourceFilterSetStructFields(ResourceFilter resourceFilter, Struct struct) {
- struct.set(RESOURCE_TYPE, resourceFilter.resourceType().code());
- struct.set(RESOURCE_NAME_FILTER, resourceFilter.name());
+ static void resourcePatternFilterSetStructFields(ResourcePatternFilter patternFilter, Struct struct) {
+ struct.set(RESOURCE_TYPE, patternFilter.resourceType().code());
+ struct.set(RESOURCE_NAME_FILTER, patternFilter.name());
if (struct.hasField(RESOURCE_NAME_TYPE_FILTER)) {
- struct.set(RESOURCE_NAME_TYPE_FILTER, resourceFilter.nameType().code());
+ struct.set(RESOURCE_NAME_TYPE_FILTER, patternFilter.nameType().code());
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/resource/Resource.java b/clients/src/main/java/org/apache/kafka/common/resource/Resource.java
index a4810b2..f41f41a 100644
--- a/clients/src/main/java/org/apache/kafka/common/resource/Resource.java
+++ b/clients/src/main/java/org/apache/kafka/common/resource/Resource.java
@@ -22,20 +22,14 @@ import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Objects;
/**
- * Represents a cluster resource with a tuple of (type, name, nameType).
+ * Represents a cluster resource with a tuple of (type, name).
*
* The API for this class is still evolving and we may break compatibility in minor releases, if necessary.
*/
@InterfaceStability.Evolving
public class Resource {
- /**
- * A special literal resource name that corresponds to 'all resources of a certain type'.
- */
- public static final String WILDCARD_RESOURCE = "*";
-
private final ResourceType resourceType;
private final String name;
- private final ResourceNameType nameType;
/**
* The name of the CLUSTER resource.
@@ -45,32 +39,19 @@ public class Resource {
/**
* A resource representing the whole cluster.
*/
- public final static Resource CLUSTER = new Resource(ResourceType.CLUSTER, CLUSTER_NAME, ResourceNameType.LITERAL);
-
- /**
- * Create an instance of this class with the provided parameters.
- *
- * @param resourceType non-null resource type
- * @param name non-null resource name
- * @param nameType non-null resource name type
- */
- public Resource(ResourceType resourceType, String name, ResourceNameType nameType) {
- this.resourceType = Objects.requireNonNull(resourceType, "resourceType");
- this.name = Objects.requireNonNull(name, "name");
- this.nameType = Objects.requireNonNull(nameType, "nameType");
- }
+ public final static Resource CLUSTER = new Resource(ResourceType.CLUSTER, CLUSTER_NAME);
/**
* Create an instance of this class with the provided parameters.
- * Resource name type would default to ResourceNameType.LITERAL.
*
* @param resourceType non-null resource type
* @param name non-null resource name
- * @deprecated Since 2.0. Use {@link #Resource(ResourceType, String, ResourceNameType)}
*/
- @Deprecated
public Resource(ResourceType resourceType, String name) {
- this(resourceType, name, ResourceNameType.LITERAL);
+ Objects.requireNonNull(resourceType);
+ this.resourceType = resourceType;
+ Objects.requireNonNull(name);
+ this.name = name;
}
/**
@@ -81,13 +62,6 @@ public class Resource {
}
/**
- * Return the resource name type.
- */
- public ResourceNameType nameType() {
- return nameType;
- }
-
- /**
* Return the resource name.
*/
public String name() {
@@ -98,36 +72,31 @@ public class Resource {
* Create a filter which matches only this Resource.
*/
public ResourceFilter toFilter() {
- return new ResourceFilter(resourceType, name, nameType);
+ return new ResourceFilter(resourceType, name);
}
@Override
public String toString() {
- return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ", nameType=" + nameType + ")";
+ return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ")";
}
/**
* Return true if this Resource has any UNKNOWN components.
*/
public boolean isUnknown() {
- return resourceType.isUnknown() || nameType.isUnknown();
+ return resourceType.isUnknown();
}
@Override
public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
+ if (!(o instanceof Resource))
return false;
-
- final Resource resource = (Resource) o;
- return resourceType == resource.resourceType &&
- Objects.equals(name, resource.name) &&
- nameType == resource.nameType;
+ Resource other = (Resource) o;
+ return resourceType.equals(other.resourceType) && Objects.equals(name, other.name);
}
@Override
public int hashCode() {
- return Objects.hash(resourceType, name, nameType);
+ return Objects.hash(resourceType, name);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java
index e197e91..0a4611f 100644
--- a/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java
+++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourceFilter.java
@@ -21,8 +21,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Objects;
-import static org.apache.kafka.common.resource.Resource.WILDCARD_RESOURCE;
-
/**
* A filter which matches Resource objects.
*
@@ -32,54 +30,22 @@ import static org.apache.kafka.common.resource.Resource.WILDCARD_RESOURCE;
public class ResourceFilter {
private final ResourceType resourceType;
private final String name;
- private final ResourceNameType nameType;
/**
* Matches any resource.
*/
- public static final ResourceFilter ANY = new ResourceFilter(ResourceType.ANY, null, ResourceNameType.ANY);
+ public static final ResourceFilter ANY = new ResourceFilter(ResourceType.ANY, null);
/**
- * Create a filter that matches {@link ResourceNameType#LITERAL literal} resources of the
- * supplied {@code resourceType} and {@code name}.
+ * Create an instance of this class with the provided parameters.
*
* @param resourceType non-null resource type
- * @param name resource name or {@code null}.
- * If {@code null}, the filter will ignore the name of resources.
- * @deprecated Since 2.0. Use {@link #ResourceFilter(ResourceType, String, ResourceNameType)}
+ * @param name resource name or null
*/
- @Deprecated
public ResourceFilter(ResourceType resourceType, String name) {
- this(resourceType, name, ResourceNameType.LITERAL);
- }
-
- /**
- * Create a filter that matches resources of the supplied {@code resourceType}, {@code name} and
- * {@code nameType}.
- * <p>
- * If the filter has each three parameters fully supplied, then it will only match a resource that has exactly
- * the same values, e.g. a filter of {@code new ResourceFilter(ResourceType.GROUP, "one", ResourceTypeName.PREFIXED)}
- * will only match the resource {@code new Resource(ResourceType.GROUP, "one", ResourceTypeName.PREFIXED)}.
- * <p>
- * Any of the three parameters can be set to be ignored by the filter:
- * <ul>
- * <li><b>{@code resourceType}</b> can be set to {@link ResourceType#ANY},
- * meaning it will match a resource of any resource type</li>
- * <li><b>{@code name}</b> can be set to {@code null}, meaning it will match a resource of any name.</li>
- * <li><b>{@code nameType}</b> can be set to {@link ResourceNameType#ANY},
- * meaning it will match a resource with any resource name type, including the
- * {@link Resource#WILDCARD_RESOURCE wildcard resource}</li>
- * </ul>
- *
- * @param resourceType non-null resource type to filter by.
- * @param name resource name to filter by, or {@code null}.
- * If {@code null}, the filter will ignore the name of resources.
- * @param nameType non-null resource name type to filter by.
- */
- public ResourceFilter(ResourceType resourceType, String name, ResourceNameType nameType) {
- this.resourceType = Objects.requireNonNull(resourceType, "resourceType");
+ Objects.requireNonNull(resourceType);
+ this.resourceType = resourceType;
this.name = name;
- this.nameType = Objects.requireNonNull(nameType, "nameType");
}
/**
@@ -96,76 +62,40 @@ public class ResourceFilter {
return name;
}
- /**
- * Return the resource name type.
- */
- public ResourceNameType nameType() {
- return nameType;
- }
-
@Override
public String toString() {
- return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ", nameType=" + nameType + ")";
+ return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ")";
}
/**
* Return true if this ResourceFilter has any UNKNOWN components.
*/
public boolean isUnknown() {
- return resourceType.isUnknown() || nameType.isUnknown();
+ return resourceType.isUnknown();
}
@Override
public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
+ if (!(o instanceof ResourceFilter))
return false;
-
- final ResourceFilter that = (ResourceFilter) o;
- return resourceType == that.resourceType &&
- Objects.equals(name, that.name) &&
- nameType == that.nameType;
+ ResourceFilter other = (ResourceFilter) o;
+ return resourceType.equals(other.resourceType) && Objects.equals(name, other.name);
}
@Override
public int hashCode() {
- return Objects.hash(resourceType, name, nameType);
+ return Objects.hash(resourceType, name);
}
/**
* Return true if this filter matches the given Resource.
- * @param other the resource path under which ACLs are stored.
*/
- public boolean matches(final Resource other) {
- throwOnInvalidParams(other);
-
- if (!resourceType().equals(ResourceType.ANY) && !resourceType().equals(other.resourceType())) {
+ public boolean matches(Resource other) {
+ if ((name != null) && (!name.equals(other.name())))
return false;
- }
-
- if (!nameType().equals(ResourceNameType.ANY) && !nameType().equals(other.nameType())) {
+ if ((resourceType != ResourceType.ANY) && (!resourceType.equals(other.resourceType())))
return false;
- }
-
- if (name() == null) {
- return true;
- }
-
- if (nameType().equals(other.nameType())) {
- return other.name().equals(name());
- }
-
- switch (other.nameType()) {
- case LITERAL:
- return other.name().equals(name()) || other.name().equals(WILDCARD_RESOURCE);
-
- case PREFIXED:
- return name().startsWith(other.name());
-
- default:
- throw new IllegalArgumentException("Unsupported ResourceNameType: " + other.nameType());
- }
+ return true;
}
/**
@@ -185,20 +115,6 @@ public class ResourceFilter {
return "Resource type is UNKNOWN.";
if (name == null)
return "Resource name is NULL.";
- if (nameType == ResourceNameType.ANY)
- return "Resource name type is ANY.";
- if (nameType == ResourceNameType.UNKNOWN)
- return "Resource name type is UNKNOWN.";
return null;
}
-
- private static void throwOnInvalidParams(final Resource aclPath) {
- if (aclPath.resourceType().equals(ResourceType.ANY) || aclPath.resourceType().equals(ResourceType.UNKNOWN)) {
- throw new IllegalArgumentException("Only concrete resource types are supported. Got: " + aclPath.resourceType());
- }
-
- if (aclPath.nameType().equals(ResourceNameType.ANY) || aclPath.nameType().equals(ResourceNameType.UNKNOWN)) {
- throw new IllegalArgumentException("Only concrete resource name types are supported. Got: " + aclPath.nameType());
- }
- }
}
diff --git a/clients/src/main/java/org/apache/kafka/common/resource/Resource.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePattern.java
similarity index 59%
copy from clients/src/main/java/org/apache/kafka/common/resource/Resource.java
copy to clients/src/main/java/org/apache/kafka/common/resource/ResourcePattern.java
index a4810b2..c6aee91 100644
--- a/clients/src/main/java/org/apache/kafka/common/resource/Resource.java
+++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePattern.java
@@ -22,12 +22,13 @@ import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Objects;
/**
- * Represents a cluster resource with a tuple of (type, name, nameType).
+ * Represents a pattern that is used by ACLs to match zero or more
+ * {@link org.apache.kafka.common.resource.Resource Resources}.
*
* The API for this class is still evolving and we may break compatibility in minor releases, if necessary.
*/
@InterfaceStability.Evolving
-public class Resource {
+public class ResourcePattern {
/**
* A special literal resource name that corresponds to 'all resources of a certain type'.
*/
@@ -37,77 +38,62 @@ public class Resource {
private final String name;
private final ResourceNameType nameType;
- /**
- * The name of the CLUSTER resource.
- */
- public final static String CLUSTER_NAME = "kafka-cluster";
-
- /**
- * A resource representing the whole cluster.
- */
- public final static Resource CLUSTER = new Resource(ResourceType.CLUSTER, CLUSTER_NAME, ResourceNameType.LITERAL);
-
- /**
- * Create an instance of this class with the provided parameters.
+ /**
+ * Create a pattern using the supplied parameters.
*
- * @param resourceType non-null resource type
- * @param name non-null resource name
- * @param nameType non-null resource name type
+ * @param resourceType non-null, specific, resource type
+ * @param name non-null resource name, which can be the {@link #WILDCARD_RESOURCE}.
+ * @param nameType non-null, specific, resource name type, which controls how the pattern will match resource names.
*/
- public Resource(ResourceType resourceType, String name, ResourceNameType nameType) {
+ public ResourcePattern(ResourceType resourceType, String name, ResourceNameType nameType) {
this.resourceType = Objects.requireNonNull(resourceType, "resourceType");
this.name = Objects.requireNonNull(name, "name");
this.nameType = Objects.requireNonNull(nameType, "nameType");
- }
- /**
- * Create an instance of this class with the provided parameters.
- * Resource name type would default to ResourceNameType.LITERAL.
- *
- * @param resourceType non-null resource type
- * @param name non-null resource name
- * @deprecated Since 2.0. Use {@link #Resource(ResourceType, String, ResourceNameType)}
- */
- @Deprecated
- public Resource(ResourceType resourceType, String name) {
- this(resourceType, name, ResourceNameType.LITERAL);
+ if (resourceType == ResourceType.ANY) {
+ throw new IllegalArgumentException("resourceType must not be ANY");
+ }
+
+ if (nameType == ResourceNameType.ANY) {
+ throw new IllegalArgumentException("nameType must not be ANY");
+ }
}
/**
- * Return the resource type.
+ * @return the specific resource type this pattern matches
*/
public ResourceType resourceType() {
return resourceType;
}
/**
- * Return the resource name type.
+ * @return the resource name.
*/
- public ResourceNameType nameType() {
- return nameType;
+ public String name() {
+ return name;
}
/**
- * Return the resource name.
+ * @return the resource name type.
*/
- public String name() {
- return name;
+ public ResourceNameType nameType() {
+ return nameType;
}
/**
- * Create a filter which matches only this Resource.
+ * @return a filter which matches only this pattern.
*/
- public ResourceFilter toFilter() {
- return new ResourceFilter(resourceType, name, nameType);
+ public ResourcePatternFilter toFilter() {
+ return new ResourcePatternFilter(resourceType, name, nameType);
}
@Override
public String toString() {
- return "(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ", nameType=" + nameType + ")";
+ return "ResourcePattern(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ", nameType=" + nameType + ")";
}
/**
- * Return true if this Resource has any UNKNOWN components.
+ * @return {@code true} if this Resource has any UNKNOWN components.
*/
public boolean isUnknown() {
return resourceType.isUnknown() || nameType.isUnknown();
@@ -120,7 +106,7 @@ public class Resource {
if (o == null || getClass() != o.getClass())
return false;
- final Resource resource = (Resource) o;
+ final ResourcePattern resource = (ResourcePattern) o;
return resourceType == resource.resourceType &&
Objects.equals(name, resource.name) &&
nameType == resource.nameType;
diff --git a/clients/src/main/java/org/apache/kafka/common/resource/ResourcePatternFilter.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePatternFilter.java
new file mode 100644
index 0000000..8b4fdc0
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourcePatternFilter.java
@@ -0,0 +1,169 @@
+/*
+ * 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.resource;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Objects;
+
+import static org.apache.kafka.common.resource.ResourcePattern.WILDCARD_RESOURCE;
+
+/**
+ * Represents a filter that can match {@link ResourcePattern}.
+ * <p>
+ * The API for this class is still evolving and we may break compatibility in minor releases, if necessary.
+ */
+@InterfaceStability.Evolving
+public class ResourcePatternFilter {
+ /**
+ * Matches any resource pattern.
+ */
+ public static final ResourcePatternFilter ANY = new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.ANY);
+
+ private final ResourceType resourceType;
+ private final String name;
+ private final ResourceNameType nameType;
+
+ /**
+ * Create a filter using the supplied parameters.
+ *
+ * @param resourceType non-null resource type.
+ * If {@link ResourceType#ANY}, the filter will ignore the resource type of the pattern.
+ * If any other resource type, the filter will match only patterns with the same type.
+ * @param name resource name or {@code null}.
+ * If {@code null}, the filter will ignore the name of resources.
+ * If {@link ResourcePattern#WILDCARD_RESOURCE}, will match only wildcard patterns.
+ * @param nameType non-null resource name type.
+ * If {@link ResourceNameType#ANY}, the filter will match patterns that would match any
+ * {@code ResourceNameType}, and also any wildcards patterns.
+ * If any other resource name type, the filter will match only patterns with the same type.
+ */
+ public ResourcePatternFilter(ResourceType resourceType, String name, ResourceNameType nameType) {
+ this.resourceType = Objects.requireNonNull(resourceType, "resourceType");
+ this.name = name;
+ this.nameType = Objects.requireNonNull(nameType, "nameType");
+ }
+
+ /**
+ * @return {@code true} if this filter has any UNKNOWN components.
+ */
+ public boolean isUnknown() {
+ return resourceType.isUnknown() || nameType.isUnknown();
+ }
+
+ /**
+ * @return the specific resource type this pattern matches
+ */
+ public ResourceType resourceType() {
+ return resourceType;
+ }
+
+ /**
+ * @return the resource name.
+ */
+ public String name() {
+ return name;
+ }
+
+ /**
+ * @return the resource name type.
+ */
+ public ResourceNameType nameType() {
+ return nameType;
+ }
+
+ /**
+ * @return {@code true} if this filter matches the given pattern.
+ */
+ public boolean matches(ResourcePattern pattern) {
+ if (!resourceType.equals(ResourceType.ANY) && !resourceType.equals(pattern.resourceType())) {
+ return false;
+ }
+
+ if (!nameType.equals(ResourceNameType.ANY) && !nameType.equals(pattern.nameType())) {
+ return false;
+ }
+
+ if (name == null) {
+ return true;
+ }
+
+ if (nameType.equals(pattern.nameType())) {
+ return name.equals(pattern.name());
+ }
+
+ switch (pattern.nameType()) {
+ case LITERAL:
+ return name.equals(pattern.name()) || pattern.name().equals(WILDCARD_RESOURCE);
+
+ case PREFIXED:
+ return name.startsWith(pattern.name());
+
+ default:
+ throw new IllegalArgumentException("Unsupported ResourceNameType: " + pattern.nameType());
+ }
+ }
+
+ /**
+ * @return {@code true} if this filter could only match one pattern.
+ * In other words, if there are no ANY or UNKNOWN fields.
+ */
+ public boolean matchesAtMostOne() {
+ return findIndefiniteField() == null;
+ }
+
+ /**
+ * @return a string describing any ANY or UNKNOWN field, or null if there is no such field.
+ */
+ public String findIndefiniteField() {
+ if (resourceType == ResourceType.ANY)
+ return "Resource type is ANY.";
+ if (resourceType == ResourceType.UNKNOWN)
+ return "Resource type is UNKNOWN.";
+ if (name == null)
+ return "Resource name is NULL.";
+ if (nameType == ResourceNameType.ANY)
+ return "Resource name type is ANY.";
+ if (nameType == ResourceNameType.UNKNOWN)
+ return "Resource name type is UNKNOWN.";
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ return "ResourcePattern(resourceType=" + resourceType + ", name=" + ((name == null) ? "<any>" : name) + ", nameType=" + nameType + ")";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ final ResourcePatternFilter resource = (ResourcePatternFilter) o;
+ return resourceType == resource.resourceType &&
+ Objects.equals(name, resource.name) &&
+ nameType == resource.nameType;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resourceType, name, nameType);
+ }
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 3a300db..5cb6bbc 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -34,6 +34,8 @@ import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
@@ -65,8 +67,6 @@ import org.apache.kafka.common.requests.ListGroupsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.requests.OffsetFetchResponse;
-import org.apache.kafka.common.resource.Resource;
-import org.apache.kafka.common.resource.ResourceFilter;
import org.apache.kafka.common.resource.ResourceNameType;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.utils.MockTime;
@@ -537,13 +537,13 @@ public class KafkaAdminClientTest {
}
}
- private static final AclBinding ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
+ private static final AclBinding ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW));
- private static final AclBinding ACL2 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL),
+ private static final AclBinding ACL2 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.DENY));
- private static final AclBindingFilter FILTER1 = new AclBindingFilter(new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
+ private static final AclBindingFilter FILTER1 = new AclBindingFilter(new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:ANONYMOUS", null, AclOperation.ANY, AclPermissionType.ANY));
- private static final AclBindingFilter FILTER2 = new AclBindingFilter(new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
+ private static final AclBindingFilter FILTER2 = new AclBindingFilter(new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:bob", null, AclOperation.ANY, AclPermissionType.ANY));
@Test
diff --git a/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java b/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java
index 6110c48..a35faca 100644
--- a/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/acl/AclBindingTest.java
@@ -16,9 +16,9 @@
*/
package org.apache.kafka.common.acl;
-import org.apache.kafka.common.resource.Resource;
-import org.apache.kafka.common.resource.ResourceFilter;
import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import org.junit.Test;
@@ -28,38 +28,38 @@ import static org.junit.Assert.assertTrue;
public class AclBindingTest {
private static final AclBinding ACL1 = new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "", AclOperation.ALL, AclPermissionType.ALLOW));
private static final AclBinding ACL2 = new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "", AclOperation.READ, AclPermissionType.ALLOW));
private static final AclBinding ACL3 = new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
private static final AclBinding UNKNOWN_ACL = new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.UNKNOWN, AclPermissionType.DENY));
private static final AclBindingFilter ANY_ANONYMOUS = new AclBindingFilter(
- ResourceFilter.ANY,
+ ResourcePatternFilter.ANY,
new AccessControlEntryFilter("User:ANONYMOUS", null, AclOperation.ANY, AclPermissionType.ANY));
private static final AclBindingFilter ANY_DENY = new AclBindingFilter(
- ResourceFilter.ANY,
+ ResourcePatternFilter.ANY,
new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.DENY));
private static final AclBindingFilter ANY_MYTOPIC = new AclBindingFilter(
- new ResourceFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePatternFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY));
@Test
public void testMatching() throws Exception {
assertTrue(ACL1.equals(ACL1));
final AclBinding acl1Copy = new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "", AclOperation.ALL, AclPermissionType.ALLOW));
assertTrue(ACL1.equals(acl1Copy));
assertTrue(acl1Copy.equals(ACL1));
diff --git a/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternFilterTest.java b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternFilterTest.java
new file mode 100644
index 0000000..87b25fc
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternFilterTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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.acl;
+
+import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
+import org.junit.Test;
+
+import static org.apache.kafka.common.resource.ResourceNameType.LITERAL;
+import static org.apache.kafka.common.resource.ResourceNameType.PREFIXED;
+import static org.apache.kafka.common.resource.ResourceType.ANY;
+import static org.apache.kafka.common.resource.ResourceType.GROUP;
+import static org.apache.kafka.common.resource.ResourceType.TOPIC;
+import static org.apache.kafka.common.resource.ResourceType.UNKNOWN;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ResourcePatternFilterTest {
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowIfResourceTypeIsAny() {
+ new ResourcePatternFilter(ANY, null, ResourceNameType.ANY)
+ .matches(new ResourcePattern(ANY, "Name", PREFIXED));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowIfResourceNameTypeIsAny() {
+ new ResourcePatternFilter(ANY, null, ResourceNameType.ANY)
+ .matches(new ResourcePattern(GROUP, "Name", ResourceNameType.ANY));
+ }
+
+ @Test
+ public void shouldBeUnknownIfResourceTypeUnknown() {
+ assertTrue(new ResourcePatternFilter(UNKNOWN, null, ResourceNameType.LITERAL).isUnknown());
+ }
+
+ @Test
+ public void shouldBeUnknownIfResourceNameTypeUnknown() {
+ assertTrue(new ResourcePatternFilter(GROUP, null, ResourceNameType.UNKNOWN).isUnknown());
+ }
+
+ @Test
+ public void shouldNotMatchIfDifferentResourceType() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Name", LITERAL)
+ .matches(new ResourcePattern(GROUP, "Name", LITERAL)));
+ }
+
+ @Test
+ public void shouldNotMatchIfDifferentName() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Different", PREFIXED)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldNotMatchIfDifferentNameCase() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "NAME", LITERAL)
+ .matches(new ResourcePattern(TOPIC, "Name", LITERAL)));
+ }
+
+ @Test
+ public void shouldNotMatchIfDifferentNameType() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Name", LITERAL)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldMatchWhereResourceTypeIsAny() {
+ assertTrue(new ResourcePatternFilter(ANY, "Name", PREFIXED)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldMatchWhereResourceNameIsAny() {
+ assertTrue(new ResourcePatternFilter(TOPIC, null, PREFIXED)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldMatchWhereResourceNameTypeIsAny() {
+ assertTrue(new ResourcePatternFilter(TOPIC, null, ResourceNameType.ANY)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldMatchLiteralIfExactMatch() {
+ assertTrue(new ResourcePatternFilter(TOPIC, "Name", LITERAL)
+ .matches(new ResourcePattern(TOPIC, "Name", LITERAL)));
+ }
+
+ @Test
+ public void shouldMatchLiteralIfNameMatchesAndFilterIsOnAnyNameType() {
+ assertTrue(new ResourcePatternFilter(TOPIC, "Name", ResourceNameType.ANY)
+ .matches(new ResourcePattern(TOPIC, "Name", LITERAL)));
+ }
+
+ @Test
+ public void shouldNotMatchLiteralIfNamePrefixed() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Name-something", ResourceNameType.ANY)
+ .matches(new ResourcePattern(TOPIC, "Name", LITERAL)));
+ }
+
+ @Test
+ public void shouldMatchLiteralWildcardIfExactMatch() {
+ assertTrue(new ResourcePatternFilter(TOPIC, "*", LITERAL)
+ .matches(new ResourcePattern(TOPIC, "*", LITERAL)));
+ }
+
+ @Test
+ public void shouldNotMatchLiteralWildcardAgainstOtherName() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Name", LITERAL)
+ .matches(new ResourcePattern(TOPIC, "*", LITERAL)));
+ }
+
+ @Test
+ public void shouldNotMatchLiteralWildcardTheWayAround() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "*", LITERAL)
+ .matches(new ResourcePattern(TOPIC, "Name", LITERAL)));
+ }
+
+ @Test
+ public void shouldMatchLiteralWildcardIfFilterHasNameTypeOfAny() {
+ assertTrue(new ResourcePatternFilter(TOPIC, "Name", ResourceNameType.ANY)
+ .matches(new ResourcePattern(TOPIC, "*", LITERAL)));
+ }
+
+ @Test
+ public void shouldMatchPrefixedIfExactMatch() {
+ assertTrue(new ResourcePatternFilter(TOPIC, "Name", PREFIXED)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldNotMatchIfBothPrefixedAndFilterIsPrefixOfResource() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Name", PREFIXED)
+ .matches(new ResourcePattern(TOPIC, "Name-something", PREFIXED)));
+ }
+
+ @Test
+ public void shouldNotMatchIfBothPrefixedAndResourceIsPrefixOfFilter() {
+ assertFalse(new ResourcePatternFilter(TOPIC, "Name-something", PREFIXED)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+
+ @Test
+ public void shouldMatchPrefixedIfNamePrefixedAnyFilterTypeIsAny() {
+ assertTrue(new ResourcePatternFilter(TOPIC, "Name-something", ResourceNameType.ANY)
+ .matches(new ResourcePattern(TOPIC, "Name", PREFIXED)));
+ }
+}
\ No newline at end of file
diff --git a/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternTest.java b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternTest.java
new file mode 100644
index 0000000..d76e213
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/acl/ResourcePatternTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.acl;
+
+import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.junit.Test;
+
+public class ResourcePatternTest {
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowIfResourceTypeIsAny() {
+ new ResourcePattern(ResourceType.ANY, "name", ResourceNameType.LITERAL);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowIfResourceNameTypeIsAny() {
+ new ResourcePattern(ResourceType.TOPIC, "name", ResourceNameType.ANY);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void shouldThrowIfResourceNameIsNull() {
+ new ResourcePattern(ResourceType.TOPIC, null, ResourceNameType.ANY);
+ }
+}
\ No newline at end of file
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java
index 748914b..1f3c15c 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/CreateAclsRequestTest.java
@@ -24,8 +24,8 @@ import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation;
-import org.apache.kafka.common.resource.Resource;
import org.apache.kafka.common.resource.ResourceNameType;
+import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
import org.junit.Test;
@@ -39,13 +39,16 @@ public class CreateAclsRequestTest {
private static final short V0 = 0;
private static final short V1 = 1;
- private static final AclBinding LITERAL_ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
+ private static final AclBinding LITERAL_ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
- private static final AclBinding LITERAL_ACL2 = new AclBinding(new Resource(ResourceType.GROUP, "group", ResourceNameType.LITERAL),
+ private static final AclBinding LITERAL_ACL2 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "group", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW));
- private static final AclBinding PREFIXED_ACL1 = new AclBinding(new Resource(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBinding PREFIXED_ACL1 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
+
+ private static final AclBinding UNKNOWN_ACL1 = new AclBinding(new ResourcePattern(ResourceType.UNKNOWN, "unknown", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
@Test(expected = UnsupportedVersionException.class)
@@ -53,6 +56,11 @@ public class CreateAclsRequestTest {
new CreateAclsRequest(V0, aclCreations(PREFIXED_ACL1));
}
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowOnIfUnknown() {
+ new CreateAclsRequest(V0, aclCreations(UNKNOWN_ACL1));
+ }
+
@Test
public void shouldRoundTripV0() {
final CreateAclsRequest original = new CreateAclsRequest(V0, aclCreations(LITERAL_ACL1, LITERAL_ACL2));
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java
index 7761337..42b7dcc 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsRequestTest.java
@@ -21,9 +21,9 @@ import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.ResourceType;
import org.junit.Test;
@@ -37,13 +37,16 @@ public class DeleteAclsRequestTest {
private static final short V0 = 0;
private static final short V1 = 1;
- private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
+ private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
- private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
- private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
+
+ private static final AclBindingFilter UNKNOWN_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.UNKNOWN, "prefix", ResourceNameType.PREFIXED),
new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
@Test(expected = UnsupportedVersionException.class)
@@ -51,6 +54,11 @@ public class DeleteAclsRequestTest {
new DeleteAclsRequest(V0, aclFilters(PREFIXED_FILTER));
}
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowOnUnknownElements() {
+ new DeleteAclsRequest(V1, aclFilters(UNKNOWN_FILTER));
+ }
+
@Test
public void shouldRoundTripV0() {
final DeleteAclsRequest original = new DeleteAclsRequest(V0, aclFilters(LITERAL_FILTER));
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java
index f8e9148..b6d9f92 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/DeleteAclsResponseTest.java
@@ -21,11 +21,11 @@ import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse;
-import org.apache.kafka.common.resource.Resource;
import org.apache.kafka.common.resource.ResourceNameType;
import org.apache.kafka.common.resource.ResourceType;
import org.junit.Test;
@@ -41,24 +41,34 @@ public class DeleteAclsResponseTest {
private static final short V0 = 0;
private static final short V1 = 1;
- private static final AclBinding LITERAL_ACL1 = new AclBinding(new org.apache.kafka.common.resource.Resource(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
+ private static final AclBinding LITERAL_ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
- private static final AclBinding LITERAL_ACL2 = new AclBinding(new org.apache.kafka.common.resource.Resource(ResourceType.GROUP, "group", ResourceNameType.LITERAL),
+ private static final AclBinding LITERAL_ACL2 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "group", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW));
- private static final AclBinding PREFIXED_ACL1 = new AclBinding(new Resource(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBinding PREFIXED_ACL1 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
+ private static final AclBinding UNKNOWN_ACL = new AclBinding(new ResourcePattern(ResourceType.UNKNOWN, "group", ResourceNameType.LITERAL),
+ new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW));
+
private static final AclFilterResponse LITERAL_RESPONSE = new AclFilterResponse(aclDeletions(LITERAL_ACL1, LITERAL_ACL2));
private static final AclFilterResponse PREFIXED_RESPONSE = new AclFilterResponse(aclDeletions(LITERAL_ACL1, PREFIXED_ACL1));
+ private static final AclFilterResponse UNKNOWN_RESPONSE = new AclFilterResponse(aclDeletions(UNKNOWN_ACL));
+
@Test(expected = UnsupportedVersionException.class)
public void shouldThrowOnV0IfNotLiteral() {
new DeleteAclsResponse(10, aclResponses(PREFIXED_RESPONSE)).toStruct(V0);
}
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowOnIfUnknown() {
+ new DeleteAclsResponse(10, aclResponses(UNKNOWN_RESPONSE)).toStruct(V1);
+ }
+
@Test
public void shouldRoundTripV0() {
final DeleteAclsResponse original = new DeleteAclsResponse(10, aclResponses(LITERAL_RESPONSE));
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java
index 543cf37..3e4e531 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsRequestTest.java
@@ -21,9 +21,9 @@ import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.ResourceType;
import org.junit.Test;
@@ -34,20 +34,28 @@ public class DescribeAclsRequestTest {
private static final short V0 = 0;
private static final short V1 = 1;
- private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
+ private static final AclBindingFilter LITERAL_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
- private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBindingFilter PREFIXED_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
- private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBindingFilter ANY_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
new AccessControlEntryFilter("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
+ private static final AclBindingFilter UNKNOWN_FILTER = new AclBindingFilter(new ResourcePatternFilter(ResourceType.UNKNOWN, "foo", ResourceNameType.LITERAL),
+ new AccessControlEntryFilter("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
+
@Test(expected = UnsupportedVersionException.class)
public void shouldThrowOnV0IfNotLiteral() {
new DescribeAclsRequest(PREFIXED_FILTER, V0);
}
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowIfUnknown() {
+ new DescribeAclsRequest(UNKNOWN_FILTER, V0);
+ }
+
@Test
public void shouldRoundTripV0() {
final DescribeAclsRequest original = new DescribeAclsRequest(LITERAL_FILTER, V0);
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java
index 81cf518..2d3ac8f 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/DescribeAclsResponseTest.java
@@ -23,8 +23,8 @@ import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.errors.UnsupportedVersionException;
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.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
import org.junit.Test;
@@ -39,20 +39,28 @@ public class DescribeAclsResponseTest {
private static final short V0 = 0;
private static final short V1 = 1;
- private static final AclBinding LITERAL_ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
+ private static final AclBinding LITERAL_ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foo", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
- private static final AclBinding LITERAL_ACL2 = new AclBinding(new Resource(ResourceType.GROUP, "group", ResourceNameType.LITERAL),
+ private static final AclBinding LITERAL_ACL2 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "group", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "127.0.0.1", AclOperation.WRITE, AclPermissionType.ALLOW));
- private static final AclBinding PREFIXED_ACL1 = new AclBinding(new Resource(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
+ private static final AclBinding PREFIXED_ACL1 = new AclBinding(new ResourcePattern(ResourceType.GROUP, "prefix", ResourceNameType.PREFIXED),
new AccessControlEntry("User:*", "127.0.0.1", AclOperation.CREATE, AclPermissionType.ALLOW));
+ private static final AclBinding UNKNOWN_ACL = new AclBinding(new ResourcePattern(ResourceType.UNKNOWN, "foo", ResourceNameType.LITERAL),
+ new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.DENY));
+
@Test(expected = UnsupportedVersionException.class)
public void shouldThrowOnV0IfNotLiteral() {
new DescribeAclsResponse(10, ApiError.NONE, aclBindings(PREFIXED_ACL1)).toStruct(V0);
}
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldThrowIfUnknown() {
+ new DescribeAclsResponse(10, ApiError.NONE, aclBindings(UNKNOWN_ACL)).toStruct(V0);
+ }
+
@Test
public void shouldRoundTripV0() {
final DescribeAclsResponse original = new DescribeAclsResponse(10, ApiError.NONE, aclBindings(LITERAL_ACL1, LITERAL_ACL2));
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index e0c72a2..f537f48 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -25,6 +25,8 @@ import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.errors.InvalidReplicaAssignmentException;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.NotCoordinatorException;
@@ -45,8 +47,6 @@ import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation;
import org.apache.kafka.common.requests.CreateAclsResponse.AclCreationResponse;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse;
-import org.apache.kafka.common.resource.Resource;
-import org.apache.kafka.common.resource.ResourceFilter;
import org.apache.kafka.common.resource.ResourceNameType;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
@@ -1087,23 +1087,23 @@ public class RequestResponseTest {
private DescribeAclsRequest createListAclsRequest() {
return new DescribeAclsRequest.Builder(new AclBindingFilter(
- new ResourceFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePatternFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY))).build();
}
private DescribeAclsResponse createDescribeAclsResponse() {
return new DescribeAclsResponse(0, ApiError.NONE, Collections.singleton(new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.ALLOW))));
}
private CreateAclsRequest createCreateAclsRequest() {
List<AclCreation> creations = new ArrayList<>();
creations.add(new AclCreation(new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "127.0.0.1", AclOperation.READ, AclPermissionType.ALLOW))));
creations.add(new AclCreation(new AclBinding(
- new Resource(ResourceType.GROUP, "mygroup", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.GROUP, "mygroup", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.DENY))));
return new CreateAclsRequest.Builder(creations).build();
}
@@ -1116,10 +1116,10 @@ public class RequestResponseTest {
private DeleteAclsRequest createDeleteAclsRequest() {
List<AclBindingFilter> filters = new ArrayList<>();
filters.add(new AclBindingFilter(
- new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
+ new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:ANONYMOUS", null, AclOperation.ANY, AclPermissionType.ANY)));
filters.add(new AclBindingFilter(
- new ResourceFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
+ new ResourcePatternFilter(ResourceType.ANY, null, ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:bob", null, AclOperation.ANY, AclPermissionType.ANY)));
return new DeleteAclsRequest.Builder(filters).build();
}
@@ -1128,10 +1128,10 @@ public class RequestResponseTest {
List<AclFilterResponse> responses = new ArrayList<>();
responses.add(new AclFilterResponse(Utils.mkSet(
new AclDeletionResult(new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))),
new AclDeletionResult(new AclBinding(
- new Resource(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL),
+ new ResourcePattern(ResourceType.TOPIC, "mytopic4", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.DENY))))));
responses.add(new AclFilterResponse(new ApiError(Errors.SECURITY_DISABLED, "No security"),
Collections.<AclDeletionResult>emptySet()));
diff --git a/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java b/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java
index 9b2d6d4..4399744 100644
--- a/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/resource/ResourceFilterTest.java
@@ -19,145 +19,64 @@ package org.apache.kafka.common.resource;
import org.junit.Test;
-import static org.apache.kafka.common.resource.ResourceNameType.LITERAL;
-import static org.apache.kafka.common.resource.ResourceNameType.PREFIXED;
import static org.apache.kafka.common.resource.ResourceType.ANY;
import static org.apache.kafka.common.resource.ResourceType.GROUP;
import static org.apache.kafka.common.resource.ResourceType.TOPIC;
-import static org.apache.kafka.common.resource.ResourceType.UNKNOWN;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ResourceFilterTest {
- @Test(expected = IllegalArgumentException.class)
- public void shouldThrowIfResourceTypeIsAny() {
- new ResourceFilter(ANY, null, ResourceNameType.ANY)
- .matches(new Resource(ANY, "Name", PREFIXED));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void shouldThrowIfResourceTypeIsUnknown() {
- new ResourceFilter(ANY, null, ResourceNameType.ANY)
- .matches(new Resource(UNKNOWN, "Name", LITERAL));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void shouldThrowIfResourceNameTypeIsAny() {
- new ResourceFilter(ANY, null, ResourceNameType.ANY)
- .matches(new Resource(GROUP, "Name", ResourceNameType.ANY));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void shouldThrowIfAclResourceNameTypeIsUnknown() {
- new ResourceFilter(ANY, null, ResourceNameType.ANY)
- .matches(new Resource(GROUP, "Name", ResourceNameType.UNKNOWN));
- }
-
@Test
public void shouldNotMatchIfDifferentResourceType() {
- assertFalse(new ResourceFilter(TOPIC, "Name", LITERAL)
- .matches(new Resource(GROUP, "Name", LITERAL)));
+ assertFalse(new ResourceFilter(TOPIC, "Name")
+ .matches(new Resource(GROUP, "Name")));
}
@Test
public void shouldNotMatchIfDifferentName() {
- assertFalse(new ResourceFilter(TOPIC, "Different", PREFIXED)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
+ assertFalse(new ResourceFilter(TOPIC, "Different")
+ .matches(new Resource(TOPIC, "Name")));
}
@Test
public void shouldNotMatchIfDifferentNameCase() {
- assertFalse(new ResourceFilter(TOPIC, "NAME", LITERAL)
- .matches(new Resource(TOPIC, "Name", LITERAL)));
- }
-
- @Test
- public void shouldNotMatchIfDifferentNameType() {
- assertFalse(new ResourceFilter(TOPIC, "Name", LITERAL)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
+ assertFalse(new ResourceFilter(TOPIC, "NAME")
+ .matches(new Resource(TOPIC, "Name")));
}
@Test
public void shouldMatchWhereResourceTypeIsAny() {
- assertTrue(new ResourceFilter(ANY, "Name", PREFIXED)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
+ assertTrue(new ResourceFilter(ANY, "Name")
+ .matches(new Resource(TOPIC, "Name")));
}
@Test
public void shouldMatchWhereResourceNameIsAny() {
- assertTrue(new ResourceFilter(TOPIC, null, PREFIXED)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
- }
-
- @Test
- public void shouldMatchWhereResourceNameTypeIsAny() {
- assertTrue(new ResourceFilter(TOPIC, null, ResourceNameType.ANY)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
+ assertTrue(new ResourceFilter(TOPIC, null)
+ .matches(new Resource(TOPIC, "Name")));
}
@Test
- public void shouldMatchLiteralIfExactMatch() {
- assertTrue(new ResourceFilter(TOPIC, "Name", LITERAL)
- .matches(new Resource(TOPIC, "Name", LITERAL)));
+ public void shouldMatchIfExactMatch() {
+ assertTrue(new ResourceFilter(TOPIC, "Name")
+ .matches(new Resource(TOPIC, "Name")));
}
@Test
- public void shouldMatchLiteralIfNameMatchesAndFilterIsOnAnyNameType() {
- assertTrue(new ResourceFilter(TOPIC, "Name", ResourceNameType.ANY)
- .matches(new Resource(TOPIC, "Name", LITERAL)));
+ public void shouldMatchWildcardIfExactMatch() {
+ assertTrue(new ResourceFilter(TOPIC, "*")
+ .matches(new Resource(TOPIC, "*")));
}
@Test
- public void shouldNotMatchLiteralIfNamePrefixed() {
- assertFalse(new ResourceFilter(TOPIC, "Name-something", ResourceNameType.ANY)
- .matches(new Resource(TOPIC, "Name", LITERAL)));
- }
-
- @Test
- public void shouldMatchLiteralWildcardIfExactMatch() {
- assertTrue(new ResourceFilter(TOPIC, "*", LITERAL)
- .matches(new Resource(TOPIC, "*", LITERAL)));
- }
-
- @Test
- public void shouldNotMatchLiteralWildcardAgainstOtherName() {
- assertFalse(new ResourceFilter(TOPIC, "Name", LITERAL)
- .matches(new Resource(TOPIC, "*", LITERAL)));
+ public void shouldNotMatchWildcardAgainstOtherName() {
+ assertFalse(new ResourceFilter(TOPIC, "Name")
+ .matches(new Resource(TOPIC, "*")));
}
@Test
public void shouldNotMatchLiteralWildcardTheWayAround() {
- assertFalse(new ResourceFilter(TOPIC, "*", LITERAL)
- .matches(new Resource(TOPIC, "Name", LITERAL)));
- }
-
- @Test
- public void shouldMatchLiteralWildcardIfFilterHasNameTypeOfAny() {
- assertTrue(new ResourceFilter(TOPIC, "Name", ResourceNameType.ANY)
- .matches(new Resource(TOPIC, "*", LITERAL)));
- }
-
- @Test
- public void shouldMatchPrefixedIfExactMatch() {
- assertTrue(new ResourceFilter(TOPIC, "Name", PREFIXED)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
- }
-
- @Test
- public void shouldNotMatchIfBothPrefixedAndFilterIsPrefixOfResource() {
- assertFalse(new ResourceFilter(TOPIC, "Name", PREFIXED)
- .matches(new Resource(TOPIC, "Name-something", PREFIXED)));
- }
-
- @Test
- public void shouldNotMatchIfBothPrefixedAndResourceIsPrefixOfFilter() {
- assertFalse(new ResourceFilter(TOPIC, "Name-something", PREFIXED)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
- }
-
- @Test
- public void shouldMatchPrefixedIfNamePrefixedAnyFilterTypeIsAny() {
- assertTrue(new ResourceFilter(TOPIC, "Name-something", ResourceNameType.ANY)
- .matches(new Resource(TOPIC, "Name", PREFIXED)));
+ assertFalse(new ResourceFilter(TOPIC, "*")
+ .matches(new Resource(TOPIC, "Name")));
}
}
\ No newline at end of file
diff --git a/core/src/main/scala/kafka/admin/AclCommand.scala b/core/src/main/scala/kafka/admin/AclCommand.scala
index 25b630d..d55e886 100644
--- a/core/src/main/scala/kafka/admin/AclCommand.scala
+++ b/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.{ResourcePatternFilter, ResourceNameType => JResourceNameType, 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 ResourcePatternFilter(JResourceType.CLUSTER, JResource.CLUSTER_NAME, JResourceNameType.LITERAL)
private val Newline = scala.util.Properties.lineSeparator
@@ -126,7 +126,7 @@ object AclCommand extends Logging {
}
}
- private def removeAcls(authorizer: Authorizer, acls: Set[Acl], filter: ResourceFilter) {
+ private def removeAcls(authorizer: Authorizer, acls: Set[Acl], filter: ResourcePatternFilter) {
getAcls(authorizer, filter)
.keys
.foreach(resource =>
@@ -148,12 +148,12 @@ object AclCommand extends Logging {
}
}
- private def getAcls(authorizer: Authorizer, filter: ResourceFilter): Map[Resource, Set[Acl]] =
+ private def getAcls(authorizer: Authorizer, filter: ResourcePatternFilter): Map[Resource, Set[Acl]] =
authorizer.getAcls()
- .filter { case (resource, acl) => filter.matches(resource.toJava) }
+ .filter { case (resource, acl) => filter.matches(resource.toPattern) }
- private def getResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = {
- var resourceToAcls = Map.empty[ResourceFilter, Set[Acl]]
+ private def getResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = {
+ var resourceToAcls = Map.empty[ResourcePatternFilter, Set[Acl]]
//if none of the --producer or --consumer options are specified , just construct ACLs from CLI options.
if (!opts.options.has(opts.producerOpt) && !opts.options.has(opts.consumerOpt)) {
@@ -172,11 +172,11 @@ object AclCommand extends Logging {
resourceToAcls
}
- private def getProducerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = {
+ private def getProducerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = {
val filters = getResourceFilter(opts)
- val topics: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.TOPIC)
- val transactionalIds: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.TRANSACTIONAL_ID)
+ val topics: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TOPIC)
+ val transactionalIds: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TRANSACTIONAL_ID)
val enableIdempotence = opts.options.has(opts.idempotentOpt)
val topicAcls = getAcl(opts, Set(Write, Describe, Create))
@@ -185,27 +185,27 @@ object AclCommand extends Logging {
//Write, Describe, Create permission on topics, Write, Describe on transactionalIds
topics.map(_ -> topicAcls).toMap ++
transactionalIds.map(_ -> transactionalIdAcls).toMap ++
- (if (enableIdempotence)
+ (if (enableIdempotence)
Map(ClusterResourceFilter -> getAcl(opts, Set(IdempotentWrite)))
else
Map.empty)
}
- private def getConsumerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = {
+ private def getConsumerResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = {
val filters = getResourceFilter(opts)
- val topics: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.TOPIC)
- val groups: Set[ResourceFilter] = filters.filter(_.resourceType == JResourceType.GROUP)
+ val topics: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.TOPIC)
+ val groups: Set[ResourcePatternFilter] = filters.filter(_.resourceType == JResourceType.GROUP)
//Read, Describe on topic, Read on consumerGroup
val acls = getAcl(opts, Set(Read, Describe))
- topics.map(_ -> acls).toMap[ResourceFilter, Set[Acl]] ++
- groups.map(_ -> getAcl(opts, Set(Read))).toMap[ResourceFilter, Set[Acl]]
+ topics.map(_ -> acls).toMap[ResourcePatternFilter, Set[Acl]] ++
+ groups.map(_ -> getAcl(opts, Set(Read))).toMap[ResourcePatternFilter, Set[Acl]]
}
- private def getCliResourceFilterToAcls(opts: AclCommandOptions): Map[ResourceFilter, Set[Acl]] = {
+ private def getCliResourceFilterToAcls(opts: AclCommandOptions): Map[ResourcePatternFilter, Set[Acl]] = {
val acls = getAcl(opts)
val filters = getResourceFilter(opts)
filters.map(_ -> acls).toMap
@@ -261,25 +261,25 @@ object AclCommand extends Logging {
Set.empty[KafkaPrincipal]
}
- private def getResourceFilter(opts: AclCommandOptions, dieIfNoResourceFound: Boolean = true): Set[ResourceFilter] = {
+ private def getResourceFilter(opts: AclCommandOptions, dieIfNoResourceFound: Boolean = true): Set[ResourcePatternFilter] = {
val resourceNameType: JResourceNameType = opts.options.valueOf(opts.resourceNameType)
- var resourceFilters = Set.empty[ResourceFilter]
+ var resourceFilters = Set.empty[ResourcePatternFilter]
if (opts.options.has(opts.topicOpt))
- opts.options.valuesOf(opts.topicOpt).asScala.foreach(topic => resourceFilters += new ResourceFilter(JResourceType.TOPIC, topic.trim, resourceNameType))
+ opts.options.valuesOf(opts.topicOpt).asScala.foreach(topic => resourceFilters += new ResourcePatternFilter(JResourceType.TOPIC, topic.trim, resourceNameType))
if (resourceNameType == JResourceNameType.LITERAL && (opts.options.has(opts.clusterOpt) || opts.options.has(opts.idempotentOpt)))
resourceFilters += ClusterResourceFilter
if (opts.options.has(opts.groupOpt))
- opts.options.valuesOf(opts.groupOpt).asScala.foreach(group => resourceFilters += new ResourceFilter(JResourceType.GROUP, group.trim, resourceNameType))
+ opts.options.valuesOf(opts.groupOpt).asScala.foreach(group => resourceFilters += new ResourcePatternFilter(JResourceType.GROUP, group.trim, resourceNameType))
if (opts.options.has(opts.transactionalIdOpt))
opts.options.valuesOf(opts.transactionalIdOpt).asScala.foreach(transactionalId =>
- resourceFilters += new ResourceFilter(JResourceType.TRANSACTIONAL_ID, transactionalId, resourceNameType))
+ resourceFilters += new ResourcePatternFilter(JResourceType.TRANSACTIONAL_ID, transactionalId, resourceNameType))
if (opts.options.has(opts.delegationTokenOpt))
- opts.options.valuesOf(opts.delegationTokenOpt).asScala.foreach(token => resourceFilters += new ResourceFilter(JResourceType.DELEGATION_TOKEN, token.trim, resourceNameType))
+ opts.options.valuesOf(opts.delegationTokenOpt).asScala.foreach(token => resourceFilters += new ResourcePatternFilter(JResourceType.DELEGATION_TOKEN, token.trim, resourceNameType))
if (resourceFilters.isEmpty && dieIfNoResourceFound)
CommandLineUtils.printUsageAndDie(opts.parser, "You must provide at least one resource: --topic <topic> or --cluster or --group <group> or --delegation-token <Delegation Token ID>")
@@ -294,7 +294,7 @@ object AclCommand extends Logging {
Console.readLine().equalsIgnoreCase("y")
}
- private def validateOperation(opts: AclCommandOptions, resourceToAcls: Map[ResourceFilter, Set[Acl]]): Unit = {
+ private def validateOperation(opts: AclCommandOptions, resourceToAcls: Map[ResourcePatternFilter, Set[Acl]]): Unit = {
for ((resource, acls) <- resourceToAcls) {
val validOps = ResourceTypeToValidOperations(resource.resourceType)
if ((acls.map(_.operation) -- validOps).nonEmpty)
diff --git a/core/src/main/scala/kafka/security/SecurityUtils.scala b/core/src/main/scala/kafka/security/SecurityUtils.scala
index 8442ba0..7489a3e 100644
--- a/core/src/main/scala/kafka/security/SecurityUtils.scala
+++ b/core/src/main/scala/kafka/security/SecurityUtils.scala
@@ -21,7 +21,7 @@ import kafka.security.auth.{Acl, Operation, PermissionType, Resource, ResourceNa
import org.apache.kafka.common.acl.{AccessControlEntry, AclBinding, AclBindingFilter}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.ApiError
-import org.apache.kafka.common.resource.{Resource => AdminResource}
+import org.apache.kafka.common.resource.ResourcePattern
import org.apache.kafka.common.security.auth.KafkaPrincipal
import scala.util.{Failure, Success, Try}
@@ -31,12 +31,12 @@ 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))
+ resourceType <- Try(ResourceType.fromJava(filter.patternFilter.resourceType))
+ resourceNameType <- Try(ResourceNameType.fromJava(filter.patternFilter.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.patternFilter.name, resourceNameType)
acl = Acl(principal, permissionType, filter.entryFilter.host, operation)
} yield (resource, acl)) match {
case Failure(throwable) => Left(new ApiError(Errors.INVALID_REQUEST, throwable.getMessage))
@@ -45,10 +45,10 @@ object SecurityUtils {
}
def convertToAclBinding(resource: Resource, acl: Acl): AclBinding = {
- val adminResource = new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava)
+ val resourcePattern = new ResourcePattern(resource.resourceType.toJava, resource.name, resource.nameType.toJava)
val entry = new AccessControlEntry(acl.principal.toString, acl.host.toString,
acl.operation.toJava, acl.permissionType.toJava)
- new AclBinding(adminResource, entry)
+ new AclBinding(resourcePattern, entry)
}
}
diff --git a/core/src/main/scala/kafka/security/auth/Acl.scala b/core/src/main/scala/kafka/security/auth/Acl.scala
index 7fa1638..bdd8584 100644
--- a/core/src/main/scala/kafka/security/auth/Acl.scala
+++ b/core/src/main/scala/kafka/security/auth/Acl.scala
@@ -18,6 +18,7 @@
package kafka.security.auth
import kafka.utils.Json
+import org.apache.kafka.common.resource.ResourcePattern
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.SecurityUtils
@@ -26,7 +27,7 @@ import scala.collection.JavaConverters._
object Acl {
val WildCardPrincipal: KafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "*")
val WildCardHost: String = "*"
- val WildCardResource: String = org.apache.kafka.common.resource.Resource.WILDCARD_RESOURCE
+ val WildCardResource: String = ResourcePattern.WILDCARD_RESOURCE
val AllowAllAcl = new Acl(WildCardPrincipal, Allow, WildCardHost, All)
val PrincipalKey = "principal"
val PermissionTypeKey = "permissionType"
diff --git a/core/src/main/scala/kafka/security/auth/Resource.scala b/core/src/main/scala/kafka/security/auth/Resource.scala
index fa63fcc..c9b5727 100644
--- a/core/src/main/scala/kafka/security/auth/Resource.scala
+++ b/core/src/main/scala/kafka/security/auth/Resource.scala
@@ -16,14 +16,29 @@
*/
package kafka.security.auth
-import java.util.Objects
-import org.apache.kafka.common.resource.{Resource => JResource}
+import org.apache.kafka.common.resource.ResourcePattern
object Resource {
+ val Separator = ":"
val ClusterResourceName = "kafka-cluster"
val ClusterResource = new Resource(Cluster, Resource.ClusterResourceName, Literal)
val ProducerIdResourceName = "producer-id"
val WildCardResource = "*"
+
+ def fromString(str: String): Resource = {
+ ResourceNameType.values.find(nameType => str.startsWith(nameType.name)) match {
+ case Some(nameType) =>
+ str.split(Separator, 3) match {
+ case Array(_, resourceType, name, _*) => new Resource(ResourceType.fromString(resourceType), name, nameType)
+ case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str)
+ }
+ case _ =>
+ str.split(Separator, 2) match {
+ case Array(resourceType, name, _*) => new Resource(ResourceType.fromString(resourceType), name, Literal)
+ case _ => throw new IllegalArgumentException("expected a string in format ResourceType:ResourceName but got " + str)
+ }
+ }
+ }
}
/**
@@ -31,13 +46,9 @@ 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) {
-
- Objects.requireNonNull(resourceType, "resourceType")
- Objects.requireNonNull(name, "name")
- Objects.requireNonNull(resourceNameType, "resourceNameType")
+case class Resource(resourceType: ResourceType, name: String, nameType: ResourceNameType) {
/**
* Create an instance of this class with the provided parameters.
@@ -52,8 +63,12 @@ case class Resource(resourceType: ResourceType, name: String, resourceNameType:
this(resourceType, name, Literal)
}
- def toJava: JResource = {
- new JResource(resourceType.toJava, name, resourceNameType.toJava)
+ def toPattern: ResourcePattern = {
+ new ResourcePattern(resourceType.toJava, name, nameType.toJava)
+ }
+
+ override def toString: String = {
+ nameType + Resource.Separator + resourceType.name + Resource.Separator + name
}
}
diff --git a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
index c828970..0cb2fae 100644
--- a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
+++ b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
@@ -27,7 +27,6 @@ import kafka.server.KafkaConfig
import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
import kafka.utils._
import kafka.zk.{AclChangeNotificationSequenceZNode, KafkaZkClient, ZkAclStore}
-import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType => JResourceNameType}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{SecurityUtils, Time}
@@ -102,8 +101,8 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
}
override def authorize(session: Session, operation: Operation, resource: Resource): Boolean = {
- if (resource.resourceNameType != Literal) {
- throw new IllegalArgumentException("Only literal resources are supported. Got: " + resource.resourceNameType)
+ if (resource.nameType != Literal) {
+ throw new IllegalArgumentException("Only literal resources are supported. Got: " + resource.nameType)
}
val principal = session.principal
@@ -203,8 +202,6 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
}
def getMatchingAcls(resourceType: ResourceType, resourceName: String): Set[Acl] = {
- val filter = new ResourceFilter(resourceType.toJava, resourceName, JResourceNameType.ANY)
-
inReadLock(lock) {
val wildcard = aclCache.get(Resource(resourceType, Acl.WildCardResource, Literal))
.map(_.acls)
@@ -368,7 +365,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
if (rt != 0)
rt
else {
- val rnt = a.resourceNameType compare b.resourceNameType
+ val rnt = a.nameType compare b.nameType
if (rnt != 0)
rnt
else
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index f4b8689..6d9e3d1 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -52,7 +52,6 @@ import org.apache.kafka.common.requests.DeleteAclsResponse.{AclDeletionResult, A
import org.apache.kafka.common.requests.DescribeLogDirsResponse.LogDirInfo
import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _}
-import org.apache.kafka.common.resource.{Resource => AdminResource}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation}
import org.apache.kafka.common.utils.{Time, Utils}
@@ -63,6 +62,7 @@ import scala.collection._
import scala.collection.mutable.ArrayBuffer
import scala.util.{Failure, Success, Try}
import org.apache.kafka.common.requests.CreateTopicsRequest.TopicDetails
+import org.apache.kafka.common.resource.ResourcePattern
/**
* Logic to handle the various Kafka requests
@@ -1920,7 +1920,7 @@ class KafkaApis(val requestChannel: RequestChannel,
val filter = describeAclsRequest.filter()
val returnedAcls = auth.getAcls.toSeq.flatMap { case (resource, acls) =>
acls.flatMap { acl =>
- val fixture = new AclBinding(new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava),
+ val fixture = new AclBinding(new ResourcePattern(resource.resourceType.toJava, resource.name, resource.nameType.toJava),
new AccessControlEntry(acl.principal.toString, acl.host.toString, acl.operation.toJava, acl.permissionType.toJava))
if (filter.matches(fixture)) Some(fixture)
else None
@@ -1994,7 +1994,7 @@ class KafkaApis(val requestChannel: RequestChannel,
val filtersWithIndex = filters.zipWithIndex
for ((resource, acls) <- aclMap; acl <- acls) {
val binding = new AclBinding(
- new AdminResource(resource.resourceType.toJava, resource.name, resource.resourceNameType.toJava),
+ new ResourcePattern(resource.resourceType.toJava, resource.name, resource.nameType.toJava),
new AccessControlEntry(acl.principal.toString, acl.host.toString, acl.operation.toJava,
acl.permissionType.toJava))
diff --git a/core/src/main/scala/kafka/zk/KafkaZkClient.scala b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
index 90f71a1..20e4b83 100644
--- a/core/src/main/scala/kafka/zk/KafkaZkClient.scala
+++ b/core/src/main/scala/kafka/zk/KafkaZkClient.scala
@@ -1008,7 +1008,7 @@ class KafkaZkClient private (zooKeeperClient: ZooKeeperClient, isSecure: Boolean
* @param resource resource name
*/
def createAclChangeNotification(resource: Resource): Unit = {
- val store = ZkAclStore(resource.resourceNameType)
+ val store = ZkAclStore(resource.nameType)
val path = store.changeSequenceZNode.createPath
val createRequest = CreateRequest(path, AclChangeNotificationSequenceZNode.encode(resource), acls(path), CreateMode.PERSISTENT_SEQUENTIAL)
val createResponse = retryRequestUntilConnected(createRequest)
diff --git a/core/src/main/scala/kafka/zk/ZkData.scala b/core/src/main/scala/kafka/zk/ZkData.scala
index 0524b45..6121035 100644
--- a/core/src/main/scala/kafka/zk/ZkData.scala
+++ b/core/src/main/scala/kafka/zk/ZkData.scala
@@ -487,7 +487,7 @@ object ZkAclStore {
}
object ResourceZNode {
- def path(resource: Resource): String = ZkAclStore(resource.resourceNameType).path(resource.resourceType, resource.name)
+ def path(resource: Resource): String = ZkAclStore(resource.nameType).path(resource.resourceType, resource.name)
def encode(acls: Set[Acl]): Array[Byte] = Json.encodeAsBytes(Acl.toJsonCompatibleMap(acls).asJava)
def decode(bytes: Array[Byte], stat: Stat): VersionedAcls = VersionedAcls(Acl.fromBytes(bytes), stat.getVersion)
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index 331a449..986fa4a 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -40,7 +40,7 @@ import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.errors._
import org.junit.{After, Before, Rule, Test}
import org.apache.kafka.common.requests.{DeleteRecordsRequest, MetadataResponse}
-import org.apache.kafka.common.resource.{Resource, ResourceNameType, ResourceType}
+import org.apache.kafka.common.resource.{ResourceNameType, ResourcePattern, ResourceType}
import org.junit.rules.Timeout
import org.junit.Assert._
@@ -933,7 +933,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
checkInvalidAlterConfigs(zkClient, servers, client)
}
- val ACL1 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
+ val ACL1 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
/**
diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
index b40dab7..a3b3233 100644
--- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
@@ -40,9 +40,9 @@ import org.apache.kafka.common.record.{CompressionType, MemoryRecords, Records,
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation
import org.apache.kafka.common.requests.CreateTopicsRequest.TopicDetails
import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _}
-import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType, Resource => AdminResource, ResourceType => AdminResourceType}
+import org.apache.kafka.common.resource.{ResourcePattern, ResourcePatternFilter, ResourceType => AdminResourceType}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
-import org.apache.kafka.common.{KafkaException, Node, TopicPartition, requests}
+import org.apache.kafka.common.{KafkaException, Node, TopicPartition, acl, requests, resource}
import org.junit.Assert._
import org.junit.{After, Assert, Before, Test}
@@ -378,12 +378,12 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
private def createAclsRequest = new CreateAclsRequest.Builder(
Collections.singletonList(new AclCreation(new AclBinding(
- new AdminResource(AdminResourceType.TOPIC, "mytopic"),
+ new ResourcePattern(AdminResourceType.TOPIC, "mytopic", resource.ResourceNameType.LITERAL),
new AccessControlEntry(userPrincipal.toString, "*", AclOperation.WRITE, AclPermissionType.DENY))))).build()
private def deleteAclsRequest = new DeleteAclsRequest.Builder(
Collections.singletonList(new AclBindingFilter(
- new ResourceFilter(AdminResourceType.TOPIC, null, ResourceNameType.LITERAL),
+ new ResourcePatternFilter(AdminResourceType.TOPIC, null, resource.ResourceNameType.LITERAL),
new AccessControlEntryFilter(userPrincipal.toString, "*", AclOperation.ANY, AclPermissionType.DENY)))).build()
private def alterReplicaLogDirsRequest = new AlterReplicaLogDirsRequest.Builder(Collections.singletonMap(tp, logDir)).build()
diff --git a/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala
index b3572c0..46bf722 100644
--- a/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslSslAdminClientIntegrationTest.scala
@@ -15,13 +15,13 @@ package kafka.api
import java.io.File
import java.util
-import kafka.security.auth.{All, Allow, Alter, AlterConfigs, Authorizer, ClusterAction, Create, Delete, Deny, Describe, Group, Literal, Operation, PermissionType, SimpleAclAuthorizer, Topic, Prefixed, Acl => AuthAcl, Resource => AuthResource}
+import kafka.security.auth.{All, Allow, Alter, AlterConfigs, Authorizer, ClusterAction, Create, Delete, Deny, Describe, Group, Operation, PermissionType, SimpleAclAuthorizer, Topic, Acl => AuthAcl, Resource => AuthResource}
import kafka.server.KafkaConfig
import kafka.utils.{CoreUtils, JaasTestUtils, TestUtils}
import org.apache.kafka.clients.admin.{AdminClient, CreateAclsOptions, DeleteAclsOptions}
import org.apache.kafka.common.acl._
import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidRequestException}
-import org.apache.kafka.common.resource.{Resource, ResourceFilter, ResourceNameType, ResourceType}
+import org.apache.kafka.common.resource.{ResourceNameType, ResourcePattern, ResourcePatternFilter, ResourceType}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.junit.Assert.assertEquals
import org.junit.{After, Assert, Before, Test}
@@ -89,19 +89,19 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
closeSasl()
}
- val anyAcl = new AclBinding(new Resource(ResourceType.TOPIC, "*", ResourceNameType.LITERAL),
+ val anyAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "*", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
- val acl2 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL),
+ val acl2 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.ALLOW))
- val acl3 = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
+ val acl3 = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW))
- val fooAcl = new AclBinding(new Resource(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL),
+ val fooAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW))
- val prefixAcl = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED),
+ val prefixAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW))
- val transactionalIdAcl = new AclBinding(new Resource(ResourceType.TRANSACTIONAL_ID, "transactional_id", ResourceNameType.LITERAL),
+ val transactionalIdAcl = new AclBinding(new ResourcePattern(ResourceType.TRANSACTIONAL_ID, "transactional_id", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.ALLOW))
- val groupAcl = new AclBinding(new Resource(ResourceType.GROUP, "*", ResourceNameType.LITERAL),
+ val groupAcl = new AclBinding(new ResourcePattern(ResourceType.GROUP, "*", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
@Test
@@ -111,7 +111,7 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
val results = client.createAcls(List(acl2, acl3).asJava)
assertEquals(Set(acl2, acl3), results.values.keySet().asScala)
results.values.values().asScala.foreach(value => value.get)
- val aclUnknown = new AclBinding(new Resource(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
+ val aclUnknown = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "mytopic3", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.UNKNOWN, AclPermissionType.ALLOW))
val results2 = client.createAcls(List(aclUnknown).asJava)
assertEquals(Set(aclUnknown), results2.values.keySet().asScala)
@@ -132,9 +132,9 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
waitForDescribeAcls(client, acl2.toFilter, Set(acl2))
waitForDescribeAcls(client, transactionalIdAcl.toFilter, Set(transactionalIdAcl))
- val filterA = new AclBindingFilter(new ResourceFilter(ResourceType.GROUP, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val filterB = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val filterC = new AclBindingFilter(new ResourceFilter(ResourceType.TRANSACTIONAL_ID, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val filterA = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val filterB = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val filterC = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TRANSACTIONAL_ID, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
waitForDescribeAcls(client, filterA, Set(groupAcl))
waitForDescribeAcls(client, filterC, Set(transactionalIdAcl))
@@ -154,13 +154,13 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
client = AdminClient.create(createConfig())
ensureAcls(Set(anyAcl, acl2, fooAcl, prefixAcl))
- val allTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY)
- val allLiteralTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val allPrefixedTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY)
- val literalMyTopic2Acls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val prefixedMyTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY)
- val allMyTopic2Acls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.ANY), AccessControlEntryFilter.ANY)
- val allFooTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foobar", ResourceNameType.ANY), AccessControlEntryFilter.ANY)
+ val allTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY)
+ val allLiteralTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val allPrefixedTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY)
+ val literalMyTopic2Acls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val prefixedMyTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic", ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY)
+ val allMyTopic2Acls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.ANY), AccessControlEntryFilter.ANY)
+ val allFooTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foobar", ResourceNameType.ANY), AccessControlEntryFilter.ANY)
assertEquals(Set(anyAcl), getAcls(anyAcl.toFilter))
assertEquals(Set(prefixAcl), getAcls(prefixAcl.toFilter))
@@ -181,9 +181,9 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
client = AdminClient.create(createConfig())
ensureAcls(Set(anyAcl, acl2, fooAcl, prefixAcl))
- val allTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY)
- val allLiteralTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val allPrefixedTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY)
+ val allTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY)
+ val allLiteralTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val allPrefixedTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.PREFIXED), AccessControlEntryFilter.ANY)
// Delete only ACLs on literal 'mytopic2' topic
var deleted = client.deleteAcls(List(acl2.toFilter).asJava).all().get().asScala.toSet
@@ -231,11 +231,11 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
client = AdminClient.create(createConfig())
ensureAcls(Set(anyAcl, acl2, fooAcl, prefixAcl)) // <-- prefixed exists, but should never be returned.
- val allTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY)
- val legacyAllTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val legacyMyTopic2Acls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val legacyAnyTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
- val legacyFooTopicAcls = new AclBindingFilter(new ResourceFilter(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val allTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.ANY), AccessControlEntryFilter.ANY)
+ val legacyAllTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val legacyMyTopic2Acls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "mytopic2", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val legacyAnyTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "*", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
+ val legacyFooTopicAcls = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "foobar", ResourceNameType.LITERAL), AccessControlEntryFilter.ANY)
assertEquals(Set(anyAcl, acl2, fooAcl), getAcls(legacyAllTopicAcls))
assertEquals(Set(acl2), getAcls(legacyMyTopic2Acls))
@@ -266,9 +266,9 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
@Test
def testAttemptToCreateInvalidAcls(): Unit = {
client = AdminClient.create(createConfig())
- val clusterAcl = new AclBinding(new Resource(ResourceType.CLUSTER, "foobar", ResourceNameType.LITERAL),
+ val clusterAcl = new AclBinding(new ResourcePattern(ResourceType.CLUSTER, "foobar", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW))
- val emptyResourceNameAcl = new AclBinding(new Resource(ResourceType.TOPIC, "", ResourceNameType.LITERAL),
+ val emptyResourceNameAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.READ, AclPermissionType.ALLOW))
val results = client.createAcls(List(clusterAcl, emptyResourceNameAcl).asJava, new CreateAclsOptions())
assertEquals(Set(clusterAcl, emptyResourceNameAcl), results.values.keySet().asScala)
@@ -336,7 +336,7 @@ class SaslSslAdminClientIntegrationTest extends AdminClientIntegrationTest with
private def testAclGet(expectAuth: Boolean): Unit = {
TestUtils.waitUntilTrue(() => {
- val userAcl = new AclBinding(new Resource(ResourceType.TOPIC, "*", ResourceNameType.LITERAL),
+ val userAcl = new AclBinding(new ResourcePattern(ResourceType.TOPIC, "*", ResourceNameType.LITERAL),
new AccessControlEntry("User:*", "*", AclOperation.ALL, AclPermissionType.ALLOW))
val results = client.describeAcls(userAcl.toFilter)
if (expectAuth) {
diff --git a/core/src/test/scala/kafka/security/auth/ResourceTest.scala b/core/src/test/scala/kafka/security/auth/ResourceTest.scala
new file mode 100644
index 0000000..66049b4
--- /dev/null
+++ b/core/src/test/scala/kafka/security/auth/ResourceTest.scala
@@ -0,0 +1,61 @@
+/*
+ * 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 kafka.security.auth
+
+import kafka.common.KafkaException
+import org.junit.Test
+import org.junit.Assert._
+
+class ResourceTest {
+ @Test(expected = classOf[KafkaException])
+ def shouldThrowTwoPartStringWithUnknownResourceType(): Unit = {
+ Resource.fromString("Unknown:fred")
+ }
+
+ @Test
+ def shouldParseOldTwoPartString(): Unit = {
+ assertEquals(Resource(Group, "fred", Literal), Resource.fromString("Group:fred"))
+ assertEquals(Resource(Topic, "t", Literal), Resource.fromString("Topic:t"))
+ }
+
+ @Test
+ def shouldParseOldTwoPartWithEmbeddedSeparators(): Unit = {
+ assertEquals(Resource(Group, ":This:is:a:weird:group:name:", Literal), Resource.fromString("Group::This:is:a:weird:group:name:"))
+ }
+
+ @Test
+ def shouldParseThreePartString(): Unit = {
+ assertEquals(Resource(Group, "fred", Prefixed), Resource.fromString("Prefixed:Group:fred"))
+ assertEquals(Resource(Topic, "t", Literal), Resource.fromString("Literal:Topic:t"))
+ }
+
+ @Test
+ def shouldParseThreePartWithEmbeddedSeparators(): Unit = {
+ assertEquals(Resource(Group, ":This:is:a:weird:group:name:", Prefixed), Resource.fromString("Prefixed:Group::This:is:a:weird:group:name:"))
+ assertEquals(Resource(Group, ":This:is:a:weird:group:name:", Literal), Resource.fromString("Literal:Group::This:is:a:weird:group:name:"))
+ }
+
+ @Test
+ def shouldRoundTripViaString(): Unit = {
+ val expected = Resource(Group, "fred", Prefixed)
+
+ val actual = Resource.fromString(expected.toString)
+
+ assertEquals(expected, actual)
+ }
+}
\ No newline at end of file
diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
index 59f543b..f56b3b4 100644
--- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
+++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
@@ -24,12 +24,13 @@ import kafka.security.auth._
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.NewPartitions
import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter, AclOperation, AclPermissionType}
-import org.apache.kafka.common.resource.{ResourceFilter, ResourceNameType, Resource => AdminResource, ResourceType => AdminResourceType}
+import org.apache.kafka.common.resource.{ResourcePattern, ResourcePatternFilter, ResourceType => AdminResourceType}
import org.apache.kafka.common.{Node, TopicPartition}
import org.apache.kafka.common.metrics.{KafkaMetric, Quota, Sensor}
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.common.protocol.types.Struct
+import org.apache.kafka.common.resource.ResourceNameType
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation
import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _}
@@ -312,12 +313,12 @@ class RequestQuotaTest extends BaseRequestTest {
case ApiKeys.CREATE_ACLS =>
new CreateAclsRequest.Builder(Collections.singletonList(new AclCreation(new AclBinding(
- new AdminResource(AdminResourceType.TOPIC, "mytopic"),
+ new ResourcePattern(AdminResourceType.TOPIC, "mytopic", ResourceNameType.LITERAL),
new AccessControlEntry("User:ANONYMOUS", "*", AclOperation.WRITE, AclPermissionType.DENY)))))
case ApiKeys.DELETE_ACLS =>
new DeleteAclsRequest.Builder(Collections.singletonList(new AclBindingFilter(
- new ResourceFilter(AdminResourceType.TOPIC, null, ResourceNameType.LITERAL),
+ new ResourcePatternFilter(AdminResourceType.TOPIC, null, ResourceNameType.LITERAL),
new AccessControlEntryFilter("User:ANONYMOUS", "*", AclOperation.ANY, AclPermissionType.DENY))))
case ApiKeys.DESCRIBE_CONFIGS =>
--
To stop receiving notification emails like this one, please contact
junrao@apache.org.