protected CompletableFuture<AckMessageResponse> ackMessageInBatch(ProxyContext ctx, String group, String topic, AckMessageRequest request) {
        List<ReceiptHandleMessage> handleMessageList = new ArrayList<>(request.getEntriesCount());

        for (AckMessageEntry ackMessageEntry : request.getEntriesList()) {
            String handleString = getHandleString(ctx, group, request, ackMessageEntry);
            handleMessageList.add(new ReceiptHandleMessage(ReceiptHandle.decode(handleString), ackMessageEntry.getMessageId()));
        }
        return this.messagingProcessor.batchAckMessage(ctx, handleMessageList, group, topic)
            .thenApply(batchAckResultList -> {
                AckMessageResponse.Builder responseBuilder = AckMessageResponse.newBuilder();
                Set<Code> responseCodes = new HashSet<>();
                for (BatchAckResult batchAckResult : batchAckResultList) {
                    AckMessageResultEntry entry = convertToAckMessageResultEntry(batchAckResult);
                    responseBuilder.addEntries(entry);
                    responseCodes.add(entry.getStatus().getCode());
                }
                setAckResponseStatus(responseBuilder, responseCodes);
                return responseBuilder.build();
            });
    }
@Test
    public void testAckMessageInBatch() throws Throwable {
        ConfigurationManager.getProxyConfig().setEnableBatchAck(true);

        String successMessageId = "msg1";
        String notOkMessageId = "msg2";
        String exceptionMessageId = "msg3";

        doAnswer((Answer<CompletableFuture<List<BatchAckResult>>>) invocation -> {
            List<ReceiptHandleMessage> receiptHandleMessageList = invocation.getArgument(1, List.class);
            List<BatchAckResult> batchAckResultList = new ArrayList<>();
            for (ReceiptHandleMessage receiptHandleMessage : receiptHandleMessageList) {
                BatchAckResult batchAckResult;
                if (receiptHandleMessage.getMessageId().equals(successMessageId)) {
                    AckResult ackResult = new AckResult();
                    ackResult.setStatus(AckStatus.OK);
                    batchAckResult = new BatchAckResult(receiptHandleMessage, ackResult);
                } else if (receiptHandleMessage.getMessageId().equals(notOkMessageId)) {
                    AckResult ackResult = new AckResult();
                    ackResult.setStatus(AckStatus.NO_EXIST);
                    batchAckResult = new BatchAckResult(receiptHandleMessage, ackResult);
                } else {
                    batchAckResult = new BatchAckResult(receiptHandleMessage, new ProxyException(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, ""));
                }
                batchAckResultList.add(batchAckResult);
            }
            return CompletableFuture.completedFuture(batchAckResultList);
        }).when(this.messagingProcessor).batchAckMessage(any(), anyList(), anyString(), anyString());

        {
            AckMessageResponse response = this.ackMessageActivity.ackMessage(
                createContext(),
                AckMessageRequest.newBuilder()
                    .setTopic(Resource.newBuilder().setName(TOPIC).build())
                    .setGroup(Resource.newBuilder().setName(GROUP).build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(successMessageId)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .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(notOkMessageId)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .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(exceptionMessageId)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .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(successMessageId)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(notOkMessageId)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .build())
                    .addEntries(AckMessageEntry.newBuilder()
                        .setMessageId(exceptionMessageId)
                        .setReceiptHandle(buildReceiptHandle(TOPIC, System.currentTimeMillis(), 3000))
                        .build())
                    .build()
            ).get();

            assertEquals(Code.MULTIPLE_RESULTS, response.getStatus().getCode());
            assertEquals(3, response.getEntriesCount());
            Map<String, Code> msgCode = new HashMap<>();
            for (AckMessageResultEntry entry : response.getEntriesList()) {
                msgCode.put(entry.getMessageId(), entry.getStatus().getCode());
            }
            assertEquals(Code.OK, msgCode.get(successMessageId));
            assertEquals(Code.INTERNAL_SERVER_ERROR, msgCode.get(notOkMessageId));
            assertEquals(Code.INVALID_RECEIPT_HANDLE, msgCode.get(exceptionMessageId));
        }
    }