@Override
    protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) {
        checkArgument(state == State.Connected);
        final TxnID txnID = new TxnID(command.getTxnidMostBits(), command.getTxnidLeastBits());
        final TransactionCoordinatorID tcId = TransactionCoordinatorID.get(command.getTxnidMostBits());
        final long requestId = command.getRequestId();
        final List<String> partitionsList = command.getPartitionsList();
        if (log.isDebugEnabled()) {
            partitionsList.forEach(partition ->
                    log.debug("Receive add published partition to txn request {} "
                            + "from {} with txnId {}, topic: [{}]", requestId, remoteAddress, txnID, partition));
        }

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

        TransactionMetadataStoreService transactionMetadataStoreService =
                service.pulsar().getTransactionMetadataStoreService();
        verifyTxnOwnership(txnID)
                .thenCompose(isOwner -> {
                    if (!isOwner) {
                        return failedFutureTxnNotOwned(txnID);
                    }
                    return transactionMetadataStoreService
                            .addProducedPartitionToTxn(txnID, partitionsList);
                })
                .whenComplete((v, ex) -> {
                    if (ex == null) {
                        if (log.isDebugEnabled()) {
                            log.debug("Send response success for add published partition to txn request {}", requestId);
                        }
                        writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId,
                                txnID.getLeastSigBits(), txnID.getMostSigBits()));
                    } else {
                        ex = handleTxnException(ex, BaseCommand.Type.ADD_PARTITION_TO_TXN.name(), requestId);

                        writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId,
                                txnID.getLeastSigBits(),
                                txnID.getMostSigBits(),
                                BrokerServiceException.getClientErrorCode(ex),
                                ex.getMessage()));
                        transactionMetadataStoreService.handleOpFail(ex, tcId);
                    }
                });
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleAddPartitionToTxn() 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.handleAddPartitionToTxn(any());
    }