@Override
    public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, AddressableMessageQueue messageQueue,
        List<Message> msgList, SendMessageRequestHeader requestHeader, long timeoutMillis) {
        byte[] body;
        String messageId;
        if (msgList.size() > 1) {
            requestHeader.setBatch(true);
            MessageBatch msgBatch = MessageBatch.generateFromList(msgList);
            MessageClientIDSetter.setUniqID(msgBatch);
            body = msgBatch.encode();
            msgBatch.setBody(body);
            messageId = MessageClientIDSetter.getUniqID(msgBatch);
        } else {
            Message message = msgList.get(0);
            body = message.getBody();
            messageId = MessageClientIDSetter.getUniqID(message);
        }
        RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader, ctx.getLanguage());
        request.setBody(body);
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        SimpleChannel channel = channelManager.createInvocationChannel(ctx);
        InvocationContext invocationContext = new InvocationContext(future);
        channel.registerInvocationContext(request.getOpaque(), invocationContext);
        ChannelHandlerContext simpleChannelHandlerContext = channel.getChannelHandlerContext();
        try {
            RemotingCommand response = brokerController.getSendMessageProcessor().processRequest(simpleChannelHandlerContext, request);
            if (response != null) {
                invocationContext.handle(response);
                channel.eraseInvocationContext(request.getOpaque());
            }
        } catch (Exception e) {
            future.completeExceptionally(e);
            channel.eraseInvocationContext(request.getOpaque());
            log.error("Failed to process sendMessage command", e);
        }
        return future.thenApply(r -> {
            SendResult sendResult = new SendResult();
            SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) r.readCustomHeader();
            SendStatus sendStatus;
            switch (r.getCode()) {
                case ResponseCode.FLUSH_DISK_TIMEOUT: {
                    sendStatus = SendStatus.FLUSH_DISK_TIMEOUT;
                    break;
                }
                case ResponseCode.FLUSH_SLAVE_TIMEOUT: {
                    sendStatus = SendStatus.FLUSH_SLAVE_TIMEOUT;
                    break;
                }
                case ResponseCode.SLAVE_NOT_AVAILABLE: {
                    sendStatus = SendStatus.SLAVE_NOT_AVAILABLE;
                    break;
                }
                case ResponseCode.SUCCESS: {
                    sendStatus = SendStatus.SEND_OK;
                    break;
                }
                default: {
                    throw new ProxyException(ProxyExceptionCode.INTERNAL_SERVER_ERROR, r.getRemark());
                }
            }
            sendResult.setSendStatus(sendStatus);
            sendResult.setMsgId(messageId);
            sendResult.setMessageQueue(new MessageQueue(requestHeader.getTopic(), brokerController.getBrokerConfig().getBrokerName(), requestHeader.getQueueId()));
            sendResult.setQueueOffset(responseHeader.getQueueOffset());
            sendResult.setTransactionId(responseHeader.getTransactionId());
            sendResult.setOffsetMsgId(responseHeader.getMsgId());
            return Collections.singletonList(sendResult);
        });
    }
@Test
    public void testSendMessageWithException() throws Exception {
        Mockito.when(sendMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.any(RemotingCommand.class)))
            .thenThrow(new RemotingCommandException("test"));
        Message message = new Message("topic", "body".getBytes(StandardCharsets.UTF_8));
        MessageClientIDSetter.setUniqID(message);
        List<Message> messagesList = Collections.singletonList(message);
        SendMessageRequestHeader sendMessageRequestHeader = new SendMessageRequestHeader();
        CompletableFuture<List<SendResult>> future = localMessageService.sendMessage(proxyContext, null, messagesList, sendMessageRequestHeader, 1000L);
        ExecutionException exception = catchThrowableOfType(future::get, ExecutionException.class);
        assertThat(exception.getCause()).isInstanceOf(RemotingCommandException.class);
    }