public CompletableFuture<Void> endTransaction(ProxyContext ctx, String topic, String transactionId, String messageId, String producerGroup,
        TransactionStatus transactionStatus, boolean fromTransactionCheck, long timeoutMillis) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        try {
            EndTransactionRequestData headerData = serviceManager.getTransactionService().genEndTransactionRequestHeader(
                ctx,
                topic,
                producerGroup,
                buildCommitOrRollback(transactionStatus),
                fromTransactionCheck,
                messageId,
                transactionId
            );
            if (headerData == null) {
                future.completeExceptionally(new ProxyException(ProxyExceptionCode.TRANSACTION_DATA_NOT_FOUND, "cannot found transaction data"));
                return future;
            }
            return this.serviceManager.getMessageService().endTransactionOneway(
                ctx,
                headerData.getBrokerName(),
                headerData.getRequestHeader(),
                timeoutMillis
            );
        } catch (Throwable t) {
            future.completeExceptionally(t);
        }
        return future;
    }
@Test
    public void testEndTransaction() throws Throwable {
        testEndTransaction(MessageSysFlag.TRANSACTION_COMMIT_TYPE, TransactionStatus.COMMIT);
        testEndTransaction(MessageSysFlag.TRANSACTION_NOT_TYPE, TransactionStatus.UNKNOWN);
        testEndTransaction(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE, TransactionStatus.ROLLBACK);
    }