public CompletableFuture<SendResult> sendMessageAsync(
        String brokerAddr,
        String brokerName,
        Message msg,
        SendMessageRequestHeader requestHeader,
        long timeoutMillis
    ) {
        SendMessageRequestHeaderV2 requestHeaderV2 = SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, requestHeaderV2);
        request.setBody(msg.getBody());

        return this.getRemotingClient().invoke(brokerAddr, request, timeoutMillis).thenCompose(response -> {
            CompletableFuture<SendResult> future0 = new CompletableFuture<>();
            try {
                future0.complete(this.processSendResponse(brokerName, msg, response, brokerAddr));
            } catch (Exception e) {
                future0.completeExceptionally(e);
            }
            return future0;
        });
    }
@Test
    public void sendMessageAsync() {
        String topic = "test";
        Message msg = new Message(topic, "test".getBytes());
        SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
        requestHeader.setTopic(topic);
        requestHeader.setProducerGroup("test");
        requestHeader.setDefaultTopic("test");
        requestHeader.setDefaultTopicQueueNums(1);
        requestHeader.setQueueId(0);
        requestHeader.setSysFlag(0);
        requestHeader.setBornTimestamp(0L);
        requestHeader.setFlag(0);
        requestHeader.setProperties("test");
        requestHeader.setReconsumeTimes(0);
        requestHeader.setUnitMode(false);
        requestHeader.setBatch(false);
        CompletableFuture<SendResult> future = mqClientAPIExt.sendMessageAsync("127.0.0.1:10911", "test", msg, requestHeader, 10);
        assertThatThrownBy(future::get).getCause().isInstanceOf(RemotingTimeoutException.class);
    }