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

        CreateUserRequestHeader requestHeader = request.decodeCommandCustomHeader(CreateUserRequestHeader.class);
        if (StringUtils.isEmpty(requestHeader.getUsername())) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("The username is blank");
            return response;
        }

        UserInfo userInfo = RemotingSerializable.decode(request.getBody(), UserInfo.class);
        userInfo.setUsername(requestHeader.getUsername());
        User user = UserConverter.convertUser(userInfo);

        if (user.getUserType() == UserType.SUPER && isNotSuperUserLogin(request)) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("The super user can only be create by super user");
            return response;
        }

        this.brokerController.getAuthenticationMetadataManager().createUser(user)
            .thenAccept(nil -> response.setCode(ResponseCode.SUCCESS))
            .exceptionally(ex -> {
                LOGGER.error("create user {} error", user.getUsername(), ex);
                return handleAuthException(response, ex);
            })
            .join();

        return response;
    }
@Test
    public void testCreateUser() throws RemotingCommandException {
        when(authenticationMetadataManager.createUser(any(User.class)))
            .thenReturn(CompletableFuture.completedFuture(null));

        CreateUserRequestHeader createUserRequestHeader = new CreateUserRequestHeader();
        createUserRequestHeader.setUsername("abc");
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.AUTH_CREATE_USER, createUserRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        UserInfo userInfo = UserInfo.of("abc", "123", UserType.NORMAL.getName());
        request.setBody(JSON.toJSONBytes(userInfo));
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        when(authenticationMetadataManager.isSuperUser(eq("rocketmq"))).thenReturn(CompletableFuture.completedFuture(true));
        createUserRequestHeader = new CreateUserRequestHeader();
        createUserRequestHeader.setUsername("super");
        request = RemotingCommand.createRequestCommand(RequestCode.AUTH_CREATE_USER, createUserRequestHeader);
        request.setVersion(441);
        request.addExtField("AccessKey", "rocketmq");
        request.makeCustomHeaderToNet();
        userInfo = UserInfo.of("super", "123", UserType.SUPER.getName());
        request.setBody(JSON.toJSONBytes(userInfo));
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        when(authenticationMetadataManager.isSuperUser(eq("rocketmq"))).thenReturn(CompletableFuture.completedFuture(false));
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
    }