root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-19411: Fix deleteAcls bug which allows more deletions than max records per user op (#19974)
CI / build (push) Waiting to run Details 

If there are more deletion filters after we initially hit the
`MAX_RECORDS_PER_USER_OP` bound, we will add an additional deletion
record ontop of that for each additional filter.

The current error message returned to the client is not useful either,
adding logic so client doesn't just get `UNKNOWN_SERVER_EXCEPTION` with
no details returned.
This commit is contained in:
Alyssa Huang 2025-06-24 11:24:50 -07:00 committed by Colin P. McCabe
parent d426d65041
commit 52a5b88512
2 changed files with 67 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java
View File

@ -176,6 +176,10 @@ public class AclControlManager {
validateFilter(filter);
AclDeleteResult result = deleteAclsForFilter(filter, records);
results.add(result);
} catch (BoundedListTooLongException e) {
// we do not return partial results here because the fact that only a portion of the deletions
// succeeded can be easily missed due to response size. instead fail the entire response
throw new InvalidRequestException(e.getMessage(), e);
} catch (Throwable e) {
results.add(new AclDeleteResult(ApiError.fromThrowable(e).exception()));
}
@ -191,13 +195,14 @@ public class AclControlManager {
StandardAcl acl = entry.getValue();
AclBinding binding = acl.toBinding();
if (filter.matches(binding)) {
deleted.add(new AclBindingDeleteResult(binding));
records.add(new ApiMessageAndVersion(
new RemoveAccessControlEntryRecord().setId(id), (short) 0));
if (records.size() > MAX_RECORDS_PER_USER_OP) {
// check size limitation first before adding additional records
if (records.size() >= MAX_RECORDS_PER_USER_OP) {
throw new BoundedListTooLongException("Cannot remove more than " +
MAX_RECORDS_PER_USER_OP + " acls in a single delete operation.");
}
deleted.add(new AclBindingDeleteResult(binding));
records.add(new ApiMessageAndVersion(
new RemoveAccessControlEntryRecord().setId(id), (short) 0));
}
}
return new AclDeleteResult(deleted);

58
metadata/src/test/java/org/apache/kafka/controller/AclControlManagerTest.java
View File

@ -48,6 +48,7 @@ import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import org.apache.kafka.server.authorizer.AuthorizationResult;
import org.apache.kafka.server.authorizer.AuthorizerServerInfo;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.server.mutable.BoundedListTooLongException;
import org.apache.kafka.timeline.SnapshotRegistry;
import org.junit.jupiter.api.Test;
@ -71,6 +72,7 @@ import static org.apache.kafka.common.acl.AclPermissionType.ALLOW;
import static org.apache.kafka.common.resource.PatternType.LITERAL;
import static org.apache.kafka.common.resource.PatternType.MATCH;
import static org.apache.kafka.common.resource.ResourceType.TOPIC;
import static org.apache.kafka.controller.QuorumController.MAX_RECORDS_PER_USER_OP;
import static org.apache.kafka.metadata.authorizer.StandardAclWithIdTest.TEST_ACLS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@ -366,4 +368,60 @@ public class AclControlManagerTest {
assertEquals(id, ((RemoveAccessControlEntryRecord) deleteAclResultsBothFilters.records().get(0).message()).id());
assertEquals(2, deleteAclResultsBothFilters.response().size());
}
@Test
public void testDeleteExceedsMaxRecords() {
AclControlManager manager = new AclControlManager.Builder().build();
MockClusterMetadataAuthorizer authorizer = new MockClusterMetadataAuthorizer();
authorizer.loadSnapshot(manager.idToAcl());
List<AclBinding> firstCreate = new ArrayList<>();
List<AclBinding> secondCreate = new ArrayList<>();
// create MAX_RECORDS_PER_USER_OP + 2 ACLs
for (int i = 0; i < MAX_RECORDS_PER_USER_OP + 2; i++) {
StandardAclWithId acl = new StandardAclWithId(Uuid.randomUuid(),
new StandardAcl(
ResourceType.TOPIC,
"mytopic_" + i,
PatternType.LITERAL,
"User:alice",
"127.0.0.1",
AclOperation.READ,
AclPermissionType.ALLOW));
// split acl creations between two create requests
if (i % 2 == 0) {
firstCreate.add(acl.toBinding());
} else {
secondCreate.add(acl.toBinding());
}
}
ControllerResult<List<AclCreateResult>> firstCreateResult = manager.createAcls(firstCreate);
assertEquals((MAX_RECORDS_PER_USER_OP / 2) + 1, firstCreateResult.response().size());
for (AclCreateResult result : firstCreateResult.response()) {
assertTrue(result.exception().isEmpty());
}
ControllerResult<List<AclCreateResult>> secondCreateResult = manager.createAcls(secondCreate);
assertEquals((MAX_RECORDS_PER_USER_OP / 2) + 1, secondCreateResult.response().size());
for (AclCreateResult result : secondCreateResult.response()) {
assertTrue(result.exception().isEmpty());
}
RecordTestUtils.replayAll(manager, firstCreateResult.records());
RecordTestUtils.replayAll(manager, secondCreateResult.records());
assertFalse(manager.idToAcl().isEmpty());
ArrayList<AclBindingFilter> filters = new ArrayList<>();
for (int i = 0; i < MAX_RECORDS_PER_USER_OP + 2; i++) {
filters.add(new AclBindingFilter(
new ResourcePatternFilter(ResourceType.TOPIC, "mytopic_" + i, PatternType.LITERAL),
AccessControlEntryFilter.ANY));
}
Exception exception = assertThrows(InvalidRequestException.class, () -> manager.deleteAcls(filters));
assertEquals(BoundedListTooLongException.class, exception.getCause().getClass());
assertEquals("Cannot remove more than " + MAX_RECORDS_PER_USER_OP + " acls in a single delete operation.", exception.getCause().getMessage());
}
}