@Override
    public CompletableFuture<AckResult> ackMessage(ProxyContext ctx, ReceiptHandle handle, String messageId,
        AckMessageRequestHeader requestHeader, long timeoutMillis) {
        SimpleChannel channel = channelManager.createChannel(ctx);
        ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext();
        RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader, ctx.getLanguage());
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        try {
            RemotingCommand response = brokerController.getAckMessageProcessor()
                .processRequest(channelHandlerContext, command);
            future.complete(response);
        } catch (Exception e) {
            log.error("Fail to process ackMessage command", e);
            future.completeExceptionally(e);
        }
        return future.thenApply(r -> {
            AckResult ackResult = new AckResult();
            if (ResponseCode.SUCCESS == r.getCode()) {
                ackResult.setStatus(AckStatus.OK);
            } else {
                ackResult.setStatus(AckStatus.NO_EXIST);
            }
            return ackResult;
        });
    }
@Test
    public void testAckMessage() throws Exception {
        String messageId = "messageId";
        long popTime = System.currentTimeMillis();
        long invisibleTime = 3000L;
        int reviveQueueId = 1;
        ReceiptHandle handle = ReceiptHandle.builder()
            .startOffset(0L)
            .retrieveTime(popTime)
            .invisibleTime(invisibleTime)
            .reviveQueueId(reviveQueueId)
            .topicType(ReceiptHandle.NORMAL_TOPIC)
            .brokerName(brokerName)
            .queueId(queueId)
            .offset(queueOffset)
            .build();
        RemotingCommand remotingCommand = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
        Mockito.when(ackMessageProcessorMock.processRequest(Mockito.any(SimpleChannelHandlerContext.class), Mockito.argThat(argument -> {
            boolean first = argument.getCode() == RequestCode.ACK_MESSAGE;
            boolean second = argument.readCustomHeader() instanceof AckMessageRequestHeader;
            return first && second;
        }))).thenReturn(remotingCommand);
        AckMessageRequestHeader requestHeader = new AckMessageRequestHeader();
        CompletableFuture<AckResult> future = localMessageService.ackMessage(proxyContext, handle, messageId,
            requestHeader, 1000L);
        AckResult ackResult = future.get();
        assertThat(ackResult.getStatus()).isEqualTo(AckStatus.OK);
    }