ControllerResult<List<AclCreateResult>> createAcls(List<AclBinding> acls) {
        List<AclCreateResult> results = new ArrayList<>(acls.size());
        List<ApiMessageAndVersion> records =
                BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
        for (AclBinding acl : acls) {
            try {
                validateNewAcl(acl);
            } catch (Throwable t) {
                ApiException e = (t instanceof ApiException) ? (ApiException) t :
                    new UnknownServerException("Unknown error while trying to create ACL", t);
                results.add(new AclCreateResult(e));
                continue;
            }
            StandardAcl standardAcl = StandardAcl.fromAclBinding(acl);
            if (existingAcls.add(standardAcl)) {
                StandardAclWithId standardAclWithId = new StandardAclWithId(newAclId(), standardAcl);
                idToAcl.put(standardAclWithId.id(), standardAcl);
                records.add(new ApiMessageAndVersion(standardAclWithId.toRecord(), (short) 0));
            }
            results.add(AclCreateResult.SUCCESS);
        }
        return new ControllerResult<>(records, results, true);
    }
The AclControlManager manages any ACLs that are stored in the __cluster_metadata topic.
If the ACLs are stored externally (such as in ZooKeeper) then there will be nothing for
this manager to do.
metadata/src/main/java/org/apache/kafka/controller/AclControlManager.java