protected RemotingCommand sendMessage(ChannelHandlerContext ctx, RemotingCommand request,
        ProxyContext context) throws Exception {
        SendMessageRequestHeader requestHeader = SendMessageRequestHeader.parseRequestHeader(request);
        String topic = requestHeader.getTopic();
        Map<String, String> property = MessageDecoder.string2messageProperties(requestHeader.getProperties());
        TopicMessageType messageType = TopicMessageType.parseFromMessageProperty(property);
        if (ConfigurationManager.getProxyConfig().isEnableTopicMessageTypeCheck()) {
            if (topicMessageTypeValidator != null) {
                // Do not check retry or dlq topic
                if (!NamespaceUtil.isRetryTopic(topic) && !NamespaceUtil.isDLQTopic(topic)) {
                    TopicMessageType topicMessageType = messagingProcessor.getMetadataService().getTopicMessageType(context, topic);
                    topicMessageTypeValidator.validate(topicMessageType, messageType);
                }
            }
        }
        if (!NamespaceUtil.isRetryTopic(topic) && !NamespaceUtil.isDLQTopic(topic)) {
            if (TopicMessageType.TRANSACTION.equals(messageType)) {
                messagingProcessor.addTransactionSubscription(context, requestHeader.getProducerGroup(), requestHeader.getTopic());
            }
        }
        return request(ctx, request, context, Duration.ofSeconds(3).toMillis());
    }
@Test
    public void testSendMessage() throws Exception {
        when(metadataServiceMock.getTopicMessageType(any(), eq(topic))).thenReturn(TopicMessageType.NORMAL);
        Message message = new Message(topic, "123".getBytes());
        message.putUserProperty("a", "b");
        SendMessageRequestHeader sendMessageRequestHeader = new SendMessageRequestHeader();
        sendMessageRequestHeader.setTopic(topic);
        sendMessageRequestHeader.setProducerGroup(producerGroup);
        sendMessageRequestHeader.setDefaultTopic("");
        sendMessageRequestHeader.setDefaultTopicQueueNums(0);
        sendMessageRequestHeader.setQueueId(0);
        sendMessageRequestHeader.setSysFlag(0);
        sendMessageRequestHeader.setBrokerName(brokerName);
        sendMessageRequestHeader.setProperties(MessageDecoder.messageProperties2String(message.getProperties()));
        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, sendMessageRequestHeader);
        remotingCommand.setBody(message.getBody());
        remotingCommand.makeCustomHeaderToNet();

        RemotingCommand expectResponse = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, "success");
        when(messagingProcessorMock.request(any(), eq(brokerName), eq(remotingCommand), anyLong()))
            .thenReturn(CompletableFuture.completedFuture(expectResponse));
        RemotingCommand response = sendMessageActivity.processRequest0(ctx, remotingCommand, null);
        assertThat(response).isNull();
        verify(ctx, times(1)).writeAndFlush(eq(expectResponse));
    }