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

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

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

        transactionMetadataStoreService.handleTcClientConnect(tcId).thenAccept(connection -> {
            if (log.isDebugEnabled()) {
                log.debug("Handle tc client connect request {} to transaction meta store {} from {} success.",
                        requestId, tcId, remoteAddress);
            }
            commandSender.sendTcClientConnectResponse(requestId);
        }).exceptionally(e -> {
            log.error("Handle tc client connect request {} to transaction meta store {} from {} fail.",
                    requestId, tcId, remoteAddress, e.getCause());
            commandSender.sendTcClientConnectResponse(requestId,
                    BrokerServiceException.getClientErrorCode(e), e.getMessage());
            return null;
        });
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleTcClientConnectRequest() 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.handleTcClientConnectRequest(any());
    }