@Override
    protected void handleEndTxn(CommandEndTxn command) {
        checkArgument(state == State.Connected);
        final long requestId = command.getRequestId();
        final int txnAction = command.getTxnAction().getValue();
        TxnID txnID = new TxnID(command.getTxnidMostBits(), command.getTxnidLeastBits());
        final TransactionCoordinatorID tcId = TransactionCoordinatorID.get(command.getTxnidMostBits());

        if (!checkTransactionEnableAndSendError(requestId)) {
            return;
        }

        TransactionMetadataStoreService transactionMetadataStoreService =
                service.pulsar().getTransactionMetadataStoreService();

        verifyTxnOwnership(txnID)
                .thenCompose(isOwner -> {
                    if (!isOwner) {
                        return failedFutureTxnNotOwned(txnID);
                    }
                    return transactionMetadataStoreService.endTransaction(txnID, txnAction, false);
                })
                .whenComplete((v, ex) -> {
                    if (ex == null) {
                        commandSender.sendEndTxnResponse(requestId, txnID, txnAction);
                    } else {
                        ex = handleTxnException(ex, BaseCommand.Type.END_TXN.name(), requestId);
                        commandSender.sendEndTxnErrorResponse(requestId, txnID,
                                BrokerServiceException.getClientErrorCode(ex), ex.getMessage());

                        transactionMetadataStoreService.handleOpFail(ex, tcId);
                    }
                });
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleEndTxn() throws Exception {
        ServerCnx serverCnx = mock(ServerCnx.class, CALLS_REAL_METHODS);
        Field stateUpdater = ServerCnx.class.getDeclaredField("state");
        stateUpdater.setAccessible(true);
        stateUpdater.set(serverCnx, ServerCnx.State.Failed);
        serverCnx.handleEndTxn(any());
    }