default List<? extends CompletionStage<AclCreateResult>> createAcls(
            AuthorizableRequestContext requestContext,
            List<AclBinding> aclBindings) {
        List<CompletableFuture<AclCreateResult>> futures = new ArrayList<>(aclBindings.size());
        AclMutator aclMutator = aclMutatorOrException();
        aclBindings.forEach(b -> futures.add(new CompletableFuture<>()));
        ControllerRequestContext context = new ControllerRequestContext(
            requestContext, OptionalLong.empty());
        aclMutator.createAcls(context, aclBindings).whenComplete((results, throwable) -> {
            if (throwable == null && results.size() != futures.size()) {
                throwable = new UnknownServerException("Invalid size " +
                    "of result set from controller. Expected " + futures.size() +
                    "; got " + results.size());
            }
            if (throwable == null) {
                for (int i = 0; i < futures.size(); i++) {
                    futures.get(i).complete(results.get(i));
                }
            } else {
                for (CompletableFuture<AclCreateResult> future : futures) {
                    ApiException e = (throwable instanceof ApiException) ? (ApiException) throwable :
                        ApiError.fromThrowable(throwable).exception();
                    future.complete(new AclCreateResult(e));
                }
            }
        });
        return futures;
    }
Create ACLs. This function must be called on the active controller, or else
the futures will fail with NOT_CONTROLLER.

@param requestContext    The request context.
@param aclBindings       The ACL bindings to create.

@return a list of futures, one per input acl binding. Each future will be completed
once addAcl has been called on the controller, and the ACL has been persisted to
the cluster metadata log.
metadata/src/main/java/org/apache/kafka/metadata/authorizer/ClusterMetadataAuthorizer.java