Skip to content

KAFKA-9027, KAFKA-9028: Convert create/delete acls requests/response to use generated protocol #7725

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 18 commits into from
Feb 3, 2020
Merged

KAFKA-9027, KAFKA-9028: Convert create/delete acls requests/response to use generated protocol #7725

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
59c9705
KAFKA-9027: Convert create/delete acls requests/response to use gener…
ijuma Nov 21, 2019
43d5c1d
Remove unnecessary classes in CreateAclsRequest and CreateAclsResponse
ijuma Nov 21, 2019
f228b05
WIP: Simplify
ijuma Nov 24, 2019
453651a
Fix failing test cases
hachikuji Dec 5, 2019
b0f13b0
Remove DeleteAclsRequest.Builder overload that takes `AclBindingFilte…
ijuma Dec 9, 2019
5ff2463
Remove unused methods
ijuma Jan 30, 2020
7e5abcf
Remove unused methods
ijuma Jan 31, 2020
f8b789c
Address review comments for DeleteAclsRequest
ijuma Jan 31, 2020
23b6ecd
Tweak DescribeAclsRequest for consistency
ijuma Jan 31, 2020
5f955ba
Simplify DescribeAclsRequest
ijuma Jan 31, 2020
ede4c80
Minor improvemens
ijuma Jan 31, 2020
90f4470
Remove unnecessary classes in DeleteAclsResponse
ijuma Jan 31, 2020
f288303
Avoid unnecessary allocation in CreateAclsRequest
ijuma Jan 31, 2020
e9a29c7
Minor efficiency improvement
ijuma Jan 31, 2020
d3477a0
Fix KafkaAdminClient.createAcls
ijuma Jan 31, 2020
f66f569
CreateAcls renames for consistency
ijuma Jan 31, 2020
4f90436
Fix KafkaAdminClient.deleteAcls
ijuma Jan 31, 2020
44bf6d1
Fix bug in DeleteAclsRequest
ijuma Jan 31, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 43 additions & 28 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,9 @@
import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData;
import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData.ReassignableTopic;
import org.apache.kafka.common.message.CreateAclsRequestData;
import org.apache.kafka.common.message.CreateAclsRequestData.AclCreation;
import org.apache.kafka.common.message.CreateAclsResponseData.AclCreationResult;
import org.apache.kafka.common.message.CreateDelegationTokenRequestData;
import org.apache.kafka.common.message.CreateDelegationTokenRequestData.CreatableRenewers;
import org.apache.kafka.common.message.CreateDelegationTokenResponseData;
Expand All @@ -79,6 +82,11 @@
import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicCollection;
import org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicConfigs;
import org.apache.kafka.common.message.CreateTopicsResponseData.CreatableTopicResult;
import org.apache.kafka.common.message.DeleteAclsRequestData;
import org.apache.kafka.common.message.DeleteAclsRequestData.DeleteAclsFilter;
import org.apache.kafka.common.message.DeleteAclsResponseData;
import org.apache.kafka.common.message.DeleteAclsResponseData.DeleteAclsFilterResult;
import org.apache.kafka.common.message.DeleteAclsResponseData.DeleteAclsMatchingAcl;
import org.apache.kafka.common.message.DeleteGroupsRequestData;
import org.apache.kafka.common.message.DeleteTopicsRequestData;
import org.apache.kafka.common.message.DeleteTopicsResponseData.DeletableTopicResult;
Expand Down Expand Up @@ -125,9 +133,7 @@
import org.apache.kafka.common.requests.AlterReplicaLogDirsResponse;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.requests.CreateAclsRequest;
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation;
import org.apache.kafka.common.requests.CreateAclsResponse;
import org.apache.kafka.common.requests.CreateAclsResponse.AclCreationResponse;
import org.apache.kafka.common.requests.CreateDelegationTokenRequest;
import org.apache.kafka.common.requests.CreateDelegationTokenResponse;
import org.apache.kafka.common.requests.CreatePartitionsRequest;
Expand All @@ -136,8 +142,6 @@
import org.apache.kafka.common.requests.CreateTopicsResponse;
import org.apache.kafka.common.requests.DeleteAclsRequest;
import org.apache.kafka.common.requests.DeleteAclsResponse;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclDeletionResult;
import org.apache.kafka.common.requests.DeleteAclsResponse.AclFilterResponse;
import org.apache.kafka.common.requests.DeleteGroupsRequest;
import org.apache.kafka.common.requests.DeleteGroupsResponse;
import org.apache.kafka.common.requests.DeleteRecordsRequest;
Expand Down Expand Up @@ -1735,44 +1739,48 @@ public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOption
final long now = time.milliseconds();
final Map<AclBinding, KafkaFutureImpl<Void>> futures = new HashMap<>();
final List<AclCreation> aclCreations = new ArrayList<>();
final List<AclBinding> aclBindingsSent = new ArrayList<>();
for (AclBinding acl : acls) {
if (futures.get(acl) == null) {
KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();
futures.put(acl, future);
String indefinite = acl.toFilter().findIndefiniteField();
if (indefinite == null) {
aclCreations.add(new AclCreation(acl));
aclCreations.add(CreateAclsRequest.aclCreation(acl));
aclBindingsSent.add(acl);
} else {
future.completeExceptionally(new InvalidRequestException("Invalid ACL creation: " +
indefinite));
}
}
}
final CreateAclsRequestData data = new CreateAclsRequestData().setCreations(aclCreations);
runnable.call(new Call("createAcls", calcDeadlineMs(now, options.timeoutMs()),
new LeastLoadedNodeProvider()) {

@Override
CreateAclsRequest.Builder createRequest(int timeoutMs) {
return new CreateAclsRequest.Builder(aclCreations);
return new CreateAclsRequest.Builder(data);
}

@Override
void handleResponse(AbstractResponse abstractResponse) {
CreateAclsResponse response = (CreateAclsResponse) abstractResponse;
List<AclCreationResponse> responses = response.aclCreationResponses();
Iterator<AclCreationResponse> iter = responses.iterator();
for (AclCreation aclCreation : aclCreations) {
KafkaFutureImpl<Void> future = futures.get(aclCreation.acl());
List<AclCreationResult> responses = response.results();
Iterator<AclCreationResult> iter = responses.iterator();
for (AclBinding aclBinding : aclBindingsSent) {
KafkaFutureImpl<Void> future = futures.get(aclBinding);
if (!iter.hasNext()) {
future.completeExceptionally(new UnknownServerException(
"The broker reported no creation result for the given ACL."));
"The broker reported no creation result for the given ACL: " + aclBinding));
} else {
AclCreationResponse creation = iter.next();
if (creation.error().isFailure()) {
future.completeExceptionally(creation.error().exception());
} else {
AclCreationResult creation = iter.next();
Errors error = Errors.forCode(creation.errorCode());
ApiError apiError = new ApiError(error, creation.errorMessage());
if (apiError.isFailure())
future.completeExceptionally(apiError.exception());
else
future.complete(null);
}
}
}
}
Expand All @@ -1789,39 +1797,46 @@ void handleFailure(Throwable throwable) {
public DeleteAclsResult deleteAcls(Collection<AclBindingFilter> filters, DeleteAclsOptions options) {
final long now = time.milliseconds();
final Map<AclBindingFilter, KafkaFutureImpl<FilterResults>> futures = new HashMap<>();
final List<AclBindingFilter> filterList = new ArrayList<>();
final List<AclBindingFilter> aclBindingFiltersSent = new ArrayList<>();
final List<DeleteAclsFilter> deleteAclsFilters = new ArrayList<>();
for (AclBindingFilter filter : filters) {
if (futures.get(filter) == null) {
filterList.add(filter);
aclBindingFiltersSent.add(filter);
deleteAclsFilters.add(DeleteAclsRequest.deleteAclsFilter(filter));
futures.put(filter, new KafkaFutureImpl<>());
}
}
final DeleteAclsRequestData data = new DeleteAclsRequestData().setFilters(deleteAclsFilters);
runnable.call(new Call("deleteAcls", calcDeadlineMs(now, options.timeoutMs()),
new LeastLoadedNodeProvider()) {

@Override
DeleteAclsRequest.Builder createRequest(int timeoutMs) {
return new DeleteAclsRequest.Builder(filterList);
return new DeleteAclsRequest.Builder(data);
}

@Override
void handleResponse(AbstractResponse abstractResponse) {
DeleteAclsResponse response = (DeleteAclsResponse) abstractResponse;
List<AclFilterResponse> responses = response.responses();
Iterator<AclFilterResponse> iter = responses.iterator();
for (AclBindingFilter filter : filterList) {
KafkaFutureImpl<FilterResults> future = futures.get(filter);
List<DeleteAclsResponseData.DeleteAclsFilterResult> results = response.filterResults();
Iterator<DeleteAclsResponseData.DeleteAclsFilterResult> iter = results.iterator();
for (AclBindingFilter bindingFilter : aclBindingFiltersSent) {
KafkaFutureImpl<FilterResults> future = futures.get(bindingFilter);
if (!iter.hasNext()) {
future.completeExceptionally(new UnknownServerException(
"The broker reported no deletion result for the given filter."));
} else {
AclFilterResponse deletion = iter.next();
if (deletion.error().isFailure()) {
future.completeExceptionally(deletion.error().exception());
DeleteAclsFilterResult filterResult = iter.next();
ApiError error = new ApiError(Errors.forCode(filterResult.errorCode()), filterResult.errorMessage());
if (error.isFailure()) {
future.completeExceptionally(error.exception());
} else {
List<FilterResult> filterResults = new ArrayList<>();
for (AclDeletionResult deletionResult : deletion.deletions()) {
filterResults.add(new FilterResult(deletionResult.acl(), deletionResult.error().exception()));
for (DeleteAclsMatchingAcl matchingAcl : filterResult.matchingAcls()) {
ApiError aclError = new ApiError(Errors.forCode(matchingAcl.errorCode()),
matchingAcl.errorMessage());
AclBinding aclBinding = DeleteAclsResponse.aclBinding(matchingAcl);
filterResults.add(new FilterResult(aclBinding, aclError.exception()));
}
future.complete(new FilterResults(filterResults));
}
Expand Down
12 changes: 6 additions & 6 deletions clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,16 @@
import org.apache.kafka.common.message.ApiVersionsResponseData;
import org.apache.kafka.common.message.ControlledShutdownRequestData;
import org.apache.kafka.common.message.ControlledShutdownResponseData;
import org.apache.kafka.common.message.CreateAclsRequestData;
import org.apache.kafka.common.message.CreateAclsResponseData;
import org.apache.kafka.common.message.CreateDelegationTokenRequestData;
import org.apache.kafka.common.message.CreateDelegationTokenResponseData;
import org.apache.kafka.common.message.CreatePartitionsRequestData;
import org.apache.kafka.common.message.CreatePartitionsResponseData;
import org.apache.kafka.common.message.CreateTopicsRequestData;
import org.apache.kafka.common.message.CreateTopicsResponseData;
import org.apache.kafka.common.message.DeleteAclsRequestData;
import org.apache.kafka.common.message.DeleteAclsResponseData;
import org.apache.kafka.common.message.DeleteGroupsRequestData;
import org.apache.kafka.common.message.DeleteGroupsResponseData;
import org.apache.kafka.common.message.DeleteTopicsRequestData;
Expand Down Expand Up @@ -98,10 +102,6 @@
import org.apache.kafka.common.requests.AlterConfigsResponse;
import org.apache.kafka.common.requests.AlterReplicaLogDirsRequest;
import org.apache.kafka.common.requests.AlterReplicaLogDirsResponse;
import org.apache.kafka.common.requests.CreateAclsRequest;
import org.apache.kafka.common.requests.CreateAclsResponse;
import org.apache.kafka.common.requests.DeleteAclsRequest;
import org.apache.kafka.common.requests.DeleteAclsResponse;
import org.apache.kafka.common.requests.DeleteRecordsRequest;
import org.apache.kafka.common.requests.DeleteRecordsResponse;
import org.apache.kafka.common.requests.DescribeConfigsRequest;
Expand Down Expand Up @@ -178,8 +178,8 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
TXN_OFFSET_COMMIT(28, "TxnOffsetCommit", false, RecordBatch.MAGIC_VALUE_V2, TxnOffsetCommitRequestData.SCHEMAS,
TxnOffsetCommitResponseData.SCHEMAS),
DESCRIBE_ACLS(29, "DescribeAcls", DescribeAclsRequestData.SCHEMAS, DescribeAclsResponseData.SCHEMAS),
CREATE_ACLS(30, "CreateAcls", CreateAclsRequest.schemaVersions(), CreateAclsResponse.schemaVersions()),
DELETE_ACLS(31, "DeleteAcls", DeleteAclsRequest.schemaVersions(), DeleteAclsResponse.schemaVersions()),
CREATE_ACLS(30, "CreateAcls", CreateAclsRequestData.SCHEMAS, CreateAclsResponseData.SCHEMAS),
DELETE_ACLS(31, "DeleteAcls", DeleteAclsRequestData.SCHEMAS, DeleteAclsResponseData.SCHEMAS),
DESCRIBE_CONFIGS(32, "DescribeConfigs", DescribeConfigsRequest.schemaVersions(),
DescribeConfigsResponse.schemaVersions()),
ALTER_CONFIGS(33, "AlterConfigs", AlterConfigsRequest.schemaVersions(),
Expand Down
27 changes: 0 additions & 27 deletions clients/src/main/java/org/apache/kafka/common/protocol/CommonFields.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
package org.apache.kafka.common.protocol;

import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.resource.PatternType;

public class CommonFields {
public static final Field.Int32 THROTTLE_TIME_MS = new Field.Int32("throttle_time_ms",
Expand All @@ -43,30 +42,4 @@ public class CommonFields {
"The transactional id or null if the producer is not transactional");
public static final Field.Int64 PRODUCER_ID = new Field.Int64("producer_id", "Current producer id in use by the transactional id.");
public static final Field.Int16 PRODUCER_EPOCH = new Field.Int16("producer_epoch", "Current epoch associated with the producer id.");

// ACL APIs
public static final Field.Int8 RESOURCE_TYPE = new Field.Int8("resource_type", "The resource type");
public static final Field.Str RESOURCE_NAME = new Field.Str("resource_name", "The resource name");
public static final Field.NullableStr RESOURCE_NAME_FILTER = new Field.NullableStr("resource_name", "The resource name filter");
public static final Field.Int8 RESOURCE_PATTERN_TYPE = new Field.Int8("resource_pattern_type", "The resource pattern type", PatternType.LITERAL.code());
public static final Field.Int8 RESOURCE_PATTERN_TYPE_FILTER = new Field.Int8("resource_pattern_type_filter", "The resource pattern type filter", PatternType.LITERAL.code());
public static final Field.Str PRINCIPAL = new Field.Str("principal", "The ACL principal");
public static final Field.NullableStr PRINCIPAL_FILTER = new Field.NullableStr("principal", "The ACL principal filter");
public static final Field.Str HOST = new Field.Str("host", "The ACL host");
public static final Field.NullableStr HOST_FILTER = new Field.NullableStr("host", "The ACL host filter");
public static final Field.Int8 OPERATION = new Field.Int8("operation", "The ACL operation");
public static final Field.Int8 PERMISSION_TYPE = new Field.Int8("permission_type", "The ACL permission type");

public static final Field.Str PRINCIPAL_TYPE = new Field.Str("principal_type", "principalType of the Kafka principal");
public static final Field.Str PRINCIPAL_NAME = new Field.Str("name", "name of the Kafka principal");

public static final Field.Int64 COMMITTED_OFFSET = new Field.Int64("offset",
"Message offset to be committed");
public static final Field.NullableStr COMMITTED_METADATA = new Field.NullableStr("metadata",
"Any associated metadata the client wants to keep.");
public static final Field.Int32 COMMITTED_LEADER_EPOCH = new Field.Int32("leader_epoch",
"The leader epoch, if provided is derived from the last consumed record. " +
"This is used by the consumer to check for log truncation and to ensure partition " +
"metadata is up to date following a group rebalance.");

}
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,9 +141,9 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct, shor
case DESCRIBE_ACLS:
return new DescribeAclsResponse(struct, version);
case CREATE_ACLS:
return new CreateAclsResponse(struct);
return new CreateAclsResponse(struct, version);
case DELETE_ACLS:
return new DeleteAclsResponse(struct);
return new DeleteAclsResponse(struct, version);
case DESCRIBE_CONFIGS:
return new DescribeConfigsResponse(struct);
case ALTER_CONFIGS:
Expand Down
Loading