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

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