@Override
    public CompletableFuture<RemotingCommand> sendMessageBack(ProxyContext ctx, ReceiptHandle handle, String messageId,
        ConsumerSendMsgBackRequestHeader requestHeader, long timeoutMillis) {
        SimpleChannel channel = channelManager.createChannel(ctx);
        ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext();
        RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader, ctx.getLanguage());
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        try {
            RemotingCommand response = brokerController.getSendMessageProcessor()
                .processRequest(channelHandlerContext, command);
            future.complete(response);
        } catch (Exception e) {
            log.error("Fail to process sendMessageBack command", e);
            future.completeExceptionally(e);
        }
        return future;
    }
@Test
    public void testSendMessageBack() throws Exception {
        RemotingCommand remotingCommand = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, "");
        Mockito.when(sendMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.argThat(argument -> {
            boolean first = argument.getCode() == RequestCode.CONSUMER_SEND_MSG_BACK;
            boolean second = argument.readCustomHeader() instanceof ConsumerSendMsgBackRequestHeader;
            return first && second;
        }))).thenReturn(remotingCommand);
        ConsumerSendMsgBackRequestHeader requestHeader = new ConsumerSendMsgBackRequestHeader();
        CompletableFuture<RemotingCommand> future = localMessageService.sendMessageBack(proxyContext, null, null, requestHeader, 1000L);
        RemotingCommand response = future.get();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }