private CompletableFuture<Boolean> verifyTxnOwnership(TxnID txnID) {
        assert ctx.executor().inEventLoop();
        return service.pulsar().getTransactionMetadataStoreService()
                .verifyTxnOwnership(txnID, getPrincipal())
                .thenComposeAsync(isOwner -> {
                    if (isOwner) {
                        return CompletableFuture.completedFuture(true);
                    }
                    if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
                        return isSuperUser();
                    } else {
                        return CompletableFuture.completedFuture(false);
                    }
                }, ctx.executor());
    }
@Test(timeOut = 30000)
    public void sendEndTxnResponse() throws Exception {
        final TransactionMetadataStoreService txnStore = mock(TransactionMetadataStoreService.class);
        when(txnStore.getTxnMeta(any())).thenReturn(CompletableFuture.completedFuture(mock(TxnMeta.class)));
        when(txnStore.verifyTxnOwnership(any(), any())).thenReturn(CompletableFuture.completedFuture(true));
        when(txnStore.endTransaction(any(TxnID.class), anyInt(), anyBoolean()))
                .thenReturn(CompletableFuture.completedFuture(null));
        when(pulsar.getTransactionMetadataStoreService()).thenReturn(txnStore);
        svcConfig.setTransactionCoordinatorEnabled(true);
        resetChannel();
        setChannelConnected();
        ByteBuf clientCommand = Commands.serializeWithSize(Commands.newEndTxn(89L, 1L, 12L,
                TxnAction.COMMIT));
        channel.writeInbound(clientCommand);
        CommandEndTxnResponse response = (CommandEndTxnResponse) getResponse();

        assertEquals(response.getRequestId(), 89L);
        assertEquals(response.getTxnidLeastBits(), 1L);
        assertEquals(response.getTxnidMostBits(), 12L);
        assertFalse(response.hasError());
        assertFalse(response.hasMessage());

        channel.finish();
    }