cmccabe commented on code in PR #12636:
URL: https://github.com/apache/kafka/pull/12636#discussion_r971280346
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizerData.java:
##########
@@ -182,59 +174,58 @@ StandardAuthorizerData copyWithNewConfig(int nodeId,
loadingComplete,
newSuperUsers,
newDefaultResult,
- aclsByResource,
- aclsById);
+ acls);
}
- StandardAuthorizerData copyWithNewAcls(Collection<Entry<Uuid,
StandardAcl>> aclEntries) {
- StandardAuthorizerData newData = new StandardAuthorizerData(
- log,
- aclMutator,
- loadingComplete,
- superUsers,
- defaultRule.result,
- new ConcurrentSkipListSet<>(),
- new ConcurrentHashMap<>());
- for (Entry<Uuid, StandardAcl> entry : aclEntries) {
- newData.addAcl(entry.getKey(), entry.getValue());
- }
- log.info("Applied {} acl(s) from image.", aclEntries.size());
- return newData;
+ StandardAuthorizerData copyWithAllNewAcls(
+ Collection<Entry<Uuid, StandardAcl>> newAclEntries
+ ) {
+ return copyWithNewAcls(EMPTY_ACLS, newAclEntries,
Collections.emptySet());
}
- void addAcl(Uuid id, StandardAcl acl) {
- try {
- StandardAcl prevAcl = aclsById.putIfAbsent(id, acl);
- if (prevAcl != null) {
- throw new RuntimeException("An ACL with ID " + id + " already
exists.");
- }
- if (!aclsByResource.add(acl)) {
- aclsById.remove(id);
- throw new RuntimeException("Unable to add the ACL with ID " +
id +
- " to aclsByResource");
- }
- log.trace("Added ACL {}: {}", id, acl);
- } catch (Throwable e) {
- log.error("addAcl error", e);
- throw e;
- }
+ StandardAuthorizerData copyWithAclChanges(
+ Collection<Entry<Uuid, StandardAcl>> newAclEntries,
+ Set<Uuid> removedAclIds
+ ) {
+ return copyWithNewAcls(acls, newAclEntries, removedAclIds);
}
- void removeAcl(Uuid id) {
- try {
- StandardAcl acl = aclsById.remove(id);
- if (acl == null) {
- throw new RuntimeException("ID " + id + " not found in
aclsById.");
+ StandardAuthorizerData copyWithNewAcls(
+ StandardAclWithId[] existingAcls,
+ Collection<Entry<Uuid, StandardAcl>> newAclEntries,
+ Set<Uuid> removedAclIds
+ ) {
+ StandardAclWithId[] newAcls = new StandardAclWithId[
+ existingAcls.length + newAclEntries.size() -
removedAclIds.size()];
+ int numRemoved = 0, j = 0;
+ for (int i = 0; i < existingAcls.length; i++) {
+ StandardAclWithId aclWithId = existingAcls[i];
+ if (removedAclIds.contains(aclWithId.id())) {
+ numRemoved++;
+ } else {
+ newAcls[j++] = aclWithId;
}
- if (!aclsByResource.remove(acl)) {
- throw new RuntimeException("Unable to remove the ACL with ID "
+ id +
- " from aclsByResource");
+ }
+ if (numRemoved < removedAclIds.size()) {
+ throw new RuntimeException("Only located " + numRemoved + " out of
" +
+ removedAclIds.size() + " removed ACL ID(s). removedAclIds = " +
+ removedAclIds.stream().map(a ->
a.toString()).collect(Collectors.joining(", ")));
+ }
+ if (!newAclEntries.isEmpty()) {
+ int i = 0;
+ for (Entry<Uuid, StandardAcl> entry : newAclEntries) {
+ newAcls[existingAcls.length + i] = new
StandardAclWithId(entry.getKey(), entry.getValue());
+ i++;
}
Review Comment:
Duplicate IDs should not happen unless there is a bug. I do wish we could
check for it here, but it would be very inefficient to do so, since we'd have
to scan the whole array.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
