public CompletableFuture<SendMessageResponse> sendMessage(ProxyContext ctx, SendMessageRequest request) {
        CompletableFuture<SendMessageResponse> future = new CompletableFuture<>();

        try {
            if (request.getMessagesCount() <= 0) {
                throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, "no message to send");
            }

            List<apache.rocketmq.v2.Message> messageList = request.getMessagesList();
            apache.rocketmq.v2.Message message = messageList.get(0);
            Resource topic = message.getTopic();
            validateTopic(topic);

            future = this.messagingProcessor.sendMessage(
                ctx,
                new SendMessageQueueSelector(request),
                topic.getName(),
                buildSysFlag(message),
                buildMessage(ctx, request.getMessagesList(), topic)
            ).thenApply(result -> convertToSendMessageResponse(ctx, request, result));
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }
@Test
    public void testParameterValidate() {
        // too large message body
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[4 * 1024 * 1024 + 1]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.MESSAGE_BODY_TOO_LARGE, e.getCode());
                throw e;
            }
        });

        // black tag
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setTag("   ")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_TAG, e.getCode());
                throw e;
            }
        });

        // tag with '|'
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setTag("|")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_TAG, e.getCode());
                throw e;
            }
        });

        // tag with \t
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setTag("\t")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_TAG, e.getCode());
                throw e;
            }
        });

        // blank message key
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .addKeys("  ")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_KEY, e.getCode());
                throw e;
            }
        });

        // blank message with \t
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .addKeys("\t")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_KEY, e.getCode());
                throw e;
            }
        });

        // blank message group
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageGroup("  ")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_GROUP, e.getCode());
                throw e;
            }
        });

        // long message group
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageGroup(createStr(ConfigurationManager.getProxyConfig().getMaxMessageGroupSize() + 1))
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_GROUP, e.getCode());
                throw e;
            }
        });

        // message group with \t
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageGroup("\t")
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_GROUP, e.getCode());
                throw e;
            }
        });

        // too large message property
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .putUserProperties("key", createStr(16 * 1024 + 1))
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.MESSAGE_PROPERTIES_TOO_LARGE, e.getCode());
                throw e;
            }
        });

        // too large message property
        assertThrows(GrpcProxyException.class, () -> {
            Map<String, String> p = new HashMap<>();
            for (int i = 0; i <= ConfigurationManager.getProxyConfig().getUserPropertyMaxNum(); i++) {
                p.put(String.valueOf(i), String.valueOf(i));
            }
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .putAllUserProperties(p)
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.MESSAGE_PROPERTIES_TOO_LARGE, e.getCode());
                throw e;
            }
        });

        // set system properties
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .putUserProperties(MessageConst.PROPERTY_TRACE_SWITCH, "false")
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, e.getCode());
                throw e;
            }
        });

        // set the key of user property with control character
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .putUserProperties("\u0000", "hello")
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, e.getCode());
                throw e;
            }
        });

        // set the value of user property with control character
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("msgId")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .putUserProperties("p", "\u0000")
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_PROPERTY_KEY, e.getCode());
                throw e;
            }
        });

        // empty message id
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId(" ")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_MESSAGE_ID, e.getCode());
                throw e;
            }
        });

        // delay time
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("id")
                                .setDeliveryTimestamp(
                                    Timestamps.fromMillis(System.currentTimeMillis() + Duration.ofDays(1).toMillis() + Duration.ofSeconds(10).toMillis()))
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.ILLEGAL_DELIVERY_TIME, e.getCode());
                throw e;
            }
        });

        // transactionRecoverySecond
        assertThrows(GrpcProxyException.class, () -> {
            try {
                this.sendMessageActivity.sendMessage(
                    createContext(),
                    SendMessageRequest.newBuilder()
                        .addMessages(Message.newBuilder()
                            .setTopic(Resource.newBuilder()
                                .setName(TOPIC)
                                .build())
                            .setSystemProperties(SystemProperties.newBuilder()
                                .setMessageId("id")
                                .setQueueId(0)
                                .setMessageType(MessageType.NORMAL)
                                .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis()))
                                .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234"))
                                .setOrphanedTransactionRecoveryDuration(Durations.fromHours(2))
                                .setMessageType(MessageType.TRANSACTION)
                                .build())
                            .setBody(ByteString.copyFrom(new byte[3]))
                            .build())
                        .build()
                ).get();
            } catch (ExecutionException t) {
                GrpcProxyException e = (GrpcProxyException) t.getCause();
                assertEquals(Code.BAD_REQUEST, e.getCode());
                throw e;
            }
        });
    }