public CompletableFuture<EndTransactionResponse> endTransaction(ProxyContext ctx, EndTransactionRequest request) {
        CompletableFuture<EndTransactionResponse> future = new CompletableFuture<>();
        try {
            validateTopic(request.getTopic());
            if (StringUtils.isBlank(request.getTransactionId())) {
                throw new GrpcProxyException(Code.INVALID_TRANSACTION_ID, "transaction id cannot be empty");
            }

            TransactionStatus transactionStatus = TransactionStatus.UNKNOWN;
            TransactionResolution transactionResolution = request.getResolution();
            switch (transactionResolution) {
                case COMMIT:
                    transactionStatus = TransactionStatus.COMMIT;
                    break;
                case ROLLBACK:
                    transactionStatus = TransactionStatus.ROLLBACK;
                    break;
                default:
                    break;
            }
            future = this.messagingProcessor.endTransaction(
                ctx,
                request.getTopic().getName(),
                request.getTransactionId(),
                request.getMessageId(),
                request.getTopic().getName(),
                transactionStatus,
                request.getSource().equals(TransactionSource.SOURCE_SERVER_CHECK))
                .thenApply(r -> EndTransactionResponse.newBuilder()
                    .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name()))
                    .build());
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }
@Test
    public void testEndTransaction() throws Throwable {
        ArgumentCaptor<TransactionStatus> transactionStatusCaptor = ArgumentCaptor.forClass(TransactionStatus.class);
        ArgumentCaptor<Boolean> fromTransactionCheckCaptor = ArgumentCaptor.forClass(Boolean.class);
        when(this.messagingProcessor.endTransaction(any(), any(), anyString(), anyString(), anyString(),
            transactionStatusCaptor.capture(),
            fromTransactionCheckCaptor.capture())).thenReturn(CompletableFuture.completedFuture(null));

        EndTransactionResponse response = this.endTransactionActivity.endTransaction(
            createContext(),
            EndTransactionRequest.newBuilder()
                .setResolution(resolution)
                .setTopic(Resource.newBuilder().setName("topic").build())
                .setMessageId(MessageClientIDSetter.createUniqID())
                .setTransactionId(MessageClientIDSetter.createUniqID())
                .setSource(source)
                .build()
        ).get();

        assertEquals(Code.OK, response.getStatus().getCode());
        assertEquals(transactionStatus, transactionStatusCaptor.getValue());
        assertEquals(fromTransactionCheck, fromTransactionCheckCaptor.getValue());
    }