@Override
    protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) {
        checkArgument(state == State.Connected);
        final long requestId = command.getRequestId();
        final long txnidMostBits = command.getTxnidMostBits();
        final long txnidLeastBits = command.getTxnidLeastBits();
        final String topic = command.getSubscription().getTopic();
        final String subName = command.getSubscription().getSubscription();
        final int txnAction = command.getTxnAction().getValue();
        final TxnID txnID = new TxnID(txnidMostBits, txnidLeastBits);
        final long lowWaterMark = command.getTxnidLeastBitsOfLowWatermark();

        if (log.isDebugEnabled()) {
            log.debug("[{}] [{}] handleEndTxnOnSubscription txnId: [{}], txnAction: [{}]", topic, subName,
                    new TxnID(txnidMostBits, txnidLeastBits), txnAction);
        }

        CompletableFuture<Optional<Topic>> topicFuture = service.getTopicIfExists(TopicName.get(topic).toString());
        topicFuture.thenAcceptAsync(optionalTopic -> {
            if (optionalTopic.isPresent()) {
                Subscription subscription = optionalTopic.get().getSubscription(subName);
                if (subscription == null) {
                    log.warn("handleEndTxnOnSubscription fail! "
                                    + "topic {} subscription {} does not exist. txnId: [{}], txnAction: [{}]",
                            optionalTopic.get().getName(), subName, txnID, TxnAction.valueOf(txnAction));
                    writeAndFlush(
                            Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits));
                    return;
                }
                // we only accept super user because this endpoint is reserved for tc to broker communication
                isSuperUser()
                        .thenCompose(isOwner -> {
                            if (!isOwner) {
                                return failedFutureTxnTcNotAllowed(txnID);
                            }
                            return subscription.endTxn(txnidMostBits, txnidLeastBits, txnAction, lowWaterMark);
                        }).whenComplete((ignored, e) -> {
                            if (e != null) {
                                e = FutureUtil.unwrapCompletionException(e);
                                log.error("handleEndTxnOnSubscription fail ! topic: {}, subscription: {}"
                                                + "txnId: [{}], txnAction: [{}]", topic, subName,
                                        txnID, TxnAction.valueOf(txnAction), e.getCause());
                                writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(
                                        requestId, txnidLeastBits, txnidMostBits,
                                        BrokerServiceException.getClientErrorCode(e),
                                        "Handle end txn on subscription failed: " + e.getMessage()));
                                return;
                            }
                            writeAndFlush(
                                    Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits));
                        });
            } else {
                getBrokerService().getManagedLedgerFactory()
                        .asyncExists(TopicName.get(topic).getPersistenceNamingEncoding())
                        .thenAccept((b) -> {
                            if (b) {
                                log.error("handleEndTxnOnSubscription fail! The topic {} does not exist in broker, "
                                                + "subscription: {}, txnId: [{}], txnAction: [{}]", topic, subName,
                                        txnID, TxnAction.valueOf(txnAction));
                                writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(
                                        requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(),
                                        ServerError.ServiceNotReady,
                                        "The topic " + topic + " does not exist in broker."));
                            } else {
                                log.warn("handleEndTxnOnSubscription fail ! The topic {} has not been created, "
                                                + "subscription: {} txnId: [{}], txnAction: [{}]",
                                        topic, subName, txnID, TxnAction.valueOf(txnAction));
                                writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(requestId,
                                        txnID.getLeastSigBits(), txnID.getMostSigBits()));
                            }
                        }).exceptionally(e -> {
                            log.error("handleEndTxnOnSubscription fail ! topic {}, subscription: {}"
                                            + "txnId: [{}], txnAction: [{}]", topic, subName,
                                    txnID, TxnAction.valueOf(txnAction), e.getCause());
                            writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(
                                    requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(),
                                    ServerError.ServiceNotReady, e.getMessage()));
                            return null;
                });
            }
        }, ctx.executor()).exceptionally(e -> {
            log.error("handleEndTxnOnSubscription fail ! topic: {}, subscription: {}"
                            + "txnId: [{}], txnAction: [{}]", topic, subName,
                    txnID, TxnAction.valueOf(txnAction), e.getCause());
            writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(
                    requestId, txnidLeastBits, txnidMostBits,
                    ServerError.ServiceNotReady,
                    "Handle end txn on subscription failed: " + e.getMessage()));
            return null;
        });
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleEndTxnOnSubscription() 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.handleEndTxnOnSubscription(any());
    }