@Override
    protected void handleNewTxn(CommandNewTxn command) {
        checkArgument(state == State.Connected);
        final long requestId = command.getRequestId();
        final TransactionCoordinatorID tcId = TransactionCoordinatorID.get(command.getTcId());
        if (log.isDebugEnabled()) {
            log.debug("Receive new txn request {} to transaction meta store {} from {}.",
                    requestId, tcId, remoteAddress);
        }

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

        TransactionMetadataStoreService transactionMetadataStoreService =
                service.pulsar().getTransactionMetadataStoreService();
        final String owner = getPrincipal();
        transactionMetadataStoreService.newTransaction(tcId, command.getTxnTtlSeconds(), owner)
            .whenComplete(((txnID, ex) -> {
                if (ex == null) {
                    if (log.isDebugEnabled()) {
                        log.debug("Send response {} for new txn request {}", tcId.getId(), requestId);
                    }
                    commandSender.sendNewTxnResponse(requestId, txnID, tcId.getId());
                } else {
                    if (ex instanceof CoordinatorException.ReachMaxActiveTxnException) {
                        // if new txn throw ReachMaxActiveTxnException, don't return any response to client,
                        // otherwise client will retry, it will wast o lot of resources
                        // link https://github.com/apache/pulsar/issues/15133
                        log.warn("New txn op reach max active transactions! tcId : {}, requestId : {}",
                                tcId.getId(), requestId, ex);
                        // do-nothing
                    } else {
                        ex = handleTxnException(ex, BaseCommand.Type.NEW_TXN.name(), requestId);

                        commandSender.sendNewTxnErrorResponse(requestId, tcId.getId(),
                                BrokerServiceException.getClientErrorCode(ex), ex.getMessage());
                        transactionMetadataStoreService.handleOpFail(ex, tcId);
                    }
                }
            }));
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleNewTxn() 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.handleNewTxn(any());
    }