private RemotingCommand createAcl(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        RemotingCommand response = RemotingCommand.createResponseCommand(null);

        CreateAclRequestHeader requestHeader = request.decodeCommandCustomHeader(CreateAclRequestHeader.class);
        Subject subject = Subject.of(requestHeader.getSubject());

        AclInfo aclInfo = RemotingSerializable.decode(request.getBody(), AclInfo.class);
        if (aclInfo == null || CollectionUtils.isEmpty(aclInfo.getPolicies())) {
            throw new AuthorizationException("The body of acl is null");
        }

        Acl acl = AclConverter.convertAcl(aclInfo);
        if (acl != null && acl.getSubject() == null) {
            acl.setSubject(subject);
        }

        this.brokerController.getAuthorizationMetadataManager().createAcl(acl)
            .thenAccept(nil -> response.setCode(ResponseCode.SUCCESS))
            .exceptionally(ex -> {
                LOGGER.error("create acl for {} error", requestHeader.getSubject(), ex);
                return handleAuthException(response, ex);
            })
            .join();
        return response;
    }
@Test
    public void testCreateAcl() throws RemotingCommandException {
        when(authorizationMetadataManager.createAcl(any(Acl.class)))
            .thenReturn(CompletableFuture.completedFuture(null));

        CreateAclRequestHeader createAclRequestHeader = new CreateAclRequestHeader();
        createAclRequestHeader.setSubject("User:abc");
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.AUTH_CREATE_ACL, createAclRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        AclInfo aclInfo = AclInfo.of("User:abc", Arrays.asList("Topic:*"), Arrays.asList("PUB"), Arrays.asList("192.168.0.1"), "Grant");
        request.setBody(JSON.toJSONBytes(aclInfo));
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }