public CompletableFuture<AckMessageResponse> ackMessage(ProxyContext ctx, AckMessageRequest request) {
        CompletableFuture<AckMessageResponse> future = new CompletableFuture<>();

        try {
            validateTopicAndConsumerGroup(request.getTopic(), request.getGroup());
            String group = request.getGroup().getName();
            String topic = request.getTopic().getName();
            if (ConfigurationManager.getProxyConfig().isEnableBatchAck()) {
                future = ackMessageInBatch(ctx, group, topic, request);
            } else {
                future = ackMessageOneByOne(ctx, group, topic, request);
            }
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }
@Test
    public void testAckMessage() throws Throwable {
        ConfigurationManager.getProxyConfig().setEnableBatchAck(false);

        String msg1 = "msg1";
        String msg2 = "msg2";
        String msg3 = "msg3";

        when(this.messagingProcessor.ackMessage(any(), any(), eq(msg1), anyString(), anyString()))
            .thenThrow(new ProxyException(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, "receipt handle is expired"));

        AckResult msg2AckResult = new AckResult();
        msg2AckResult.setStatus(AckStatus.OK);
        when(this.messagingProcessor.ackMessage(any(), any(), eq(msg2), anyString(), anyString()))
            .thenReturn(CompletableFuture.completedFuture(msg2AckResult));

        AckResult msg3AckResult = new AckResult();
        msg3AckResult.setStatus(AckStatus.NO_EXIST);
        when(this.messagingProcessor.ackMessage(any(), any(), eq(msg3), anyString(), anyString()))
            .thenReturn(CompletableFuture.completedFuture(msg3AckResult));

        {
            AckMessageResponse response = this.ackMessageActivity.ackMessage(
                createContext(),
                AckMessageRequest.newBuilder()
                    .setTopic(Resource.newBuilder().setName(TOPIC).build())
                    .setGroup(Resource.newBuilder().setName(GROUP).build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(msg1)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis() - 10000, 1000))
                        .build())
                    .build()
            ).get();
            assertEquals(Code.INVALID_RECEIPT_HANDLE, response.getStatus().getCode());
        }
        {
            AckMessageResponse response = this.ackMessageActivity.ackMessage(
                createContext(),
                AckMessageRequest.newBuilder()
                    .setTopic(Resource.newBuilder().setName(TOPIC).build())
                    .setGroup(Resource.newBuilder().setName(GROUP).build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(msg2)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis() - 10000, 1000))
                        .build())
                    .build()
            ).get();
            assertEquals(Code.OK, response.getStatus().getCode());
        }
        {
            AckMessageResponse response = this.ackMessageActivity.ackMessage(
                createContext(),
                AckMessageRequest.newBuilder()
                    .setTopic(Resource.newBuilder().setName(TOPIC).build())
                    .setGroup(Resource.newBuilder().setName(GROUP).build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(msg3)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis() - 10000, 1000))
                        .build())
                    .build()
            ).get();
            assertEquals(Code.INTERNAL_SERVER_ERROR, response.getStatus().getCode());
        }
        {
            AckMessageResponse response = this.ackMessageActivity.ackMessage(
                createContext(),
                AckMessageRequest.newBuilder()
                    .setTopic(Resource.newBuilder().setName(TOPIC).build())
                    .setGroup(Resource.newBuilder().setName(GROUP).build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(msg1)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis() - 10000, 1000))
                        .build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(msg2)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(msg3)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .build())
                    .build()
            ).get();

            assertEquals(Code.MULTIPLE_RESULTS, response.getStatus().getCode());
            assertEquals(3, response.getEntriesCount());
            assertEquals(Code.INVALID_RECEIPT_HANDLE, response.getEntries(0).getStatus().getCode());
            assertEquals(Code.OK, response.getEntries(1).getStatus().getCode());
            assertEquals(Code.INTERNAL_SERVER_ERROR, response.getEntries(2).getStatus().getCode());
        }
    }