private RemotingCommand lockBatchMQ(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        LockBatchRequestBody requestBody = LockBatchRequestBody.decode(request.getBody(), LockBatchRequestBody.class);

        Set<MessageQueue> lockOKMQSet = new HashSet<>();
        Set<MessageQueue> selfLockOKMQSet = this.brokerController.getRebalanceLockManager().tryLockBatch(
            requestBody.getConsumerGroup(),
            requestBody.getMqSet(),
            requestBody.getClientId());
        if (requestBody.isOnlyThisBroker() || !brokerController.getBrokerConfig().isLockInStrictMode()) {
            lockOKMQSet = selfLockOKMQSet;
        } else {
            requestBody.setOnlyThisBroker(true);
            int replicaSize = this.brokerController.getMessageStoreConfig().getTotalReplicas();

            int quorum = replicaSize / 2 + 1;

            if (quorum <= 1) {
                lockOKMQSet = selfLockOKMQSet;
            } else {
                final ConcurrentMap<MessageQueue, Integer> mqLockMap = new ConcurrentHashMap<>();
                for (MessageQueue mq : selfLockOKMQSet) {
                    if (!mqLockMap.containsKey(mq)) {
                        mqLockMap.put(mq, 0);
                    }
                    mqLockMap.put(mq, mqLockMap.get(mq) + 1);
                }

                BrokerMemberGroup memberGroup = this.brokerController.getBrokerMemberGroup();

                if (memberGroup != null) {
                    Map<Long, String> addrMap = new HashMap<>(memberGroup.getBrokerAddrs());
                    addrMap.remove(this.brokerController.getBrokerConfig().getBrokerId());
                    final CountDownLatch countDownLatch = new CountDownLatch(addrMap.size());
                    requestBody.setMqSet(selfLockOKMQSet);
                    requestBody.setOnlyThisBroker(true);
                    for (Long brokerId : addrMap.keySet()) {
                        try {
                            this.brokerController.getBrokerOuterAPI().lockBatchMQAsync(addrMap.get(brokerId),
                                requestBody, 1000, new LockCallback() {
                                    @Override
                                    public void onSuccess(Set<MessageQueue> lockOKMQSet) {
                                        for (MessageQueue mq : lockOKMQSet) {
                                            if (!mqLockMap.containsKey(mq)) {
                                                mqLockMap.put(mq, 0);
                                            }
                                            mqLockMap.put(mq, mqLockMap.get(mq) + 1);
                                        }
                                        countDownLatch.countDown();
                                    }

                                    @Override
                                    public void onException(Throwable e) {
                                        LOGGER.warn("lockBatchMQAsync on {} failed, {}", addrMap.get(brokerId), e);
                                        countDownLatch.countDown();
                                    }
                                });
                        } catch (Exception e) {
                            LOGGER.warn("lockBatchMQAsync on {} failed, {}", addrMap.get(brokerId), e);
                            countDownLatch.countDown();
                        }
                    }
                    try {
                        countDownLatch.await(2000, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        LOGGER.warn("lockBatchMQ exception on {}, {}", this.brokerController.getBrokerConfig().getBrokerName(), e);
                    }
                }

                for (MessageQueue mq : mqLockMap.keySet()) {
                    if (mqLockMap.get(mq) >= quorum) {
                        lockOKMQSet.add(mq);
                    }
                }
            }
        }

        LockBatchResponseBody responseBody = new LockBatchResponseBody();
        responseBody.setLockOKMQSet(lockOKMQSet);

        response.setBody(responseBody.encode());
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }
@Test
    public void testLockBatchMQ() throws Exception {
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.LOCK_BATCH_MQ, null);
        LockBatchRequestBody lockBatchRequestBody = new LockBatchRequestBody();
        lockBatchRequestBody.setClientId("1111");
        lockBatchRequestBody.setConsumerGroup("group");
        request.setBody(JSON.toJSON(lockBatchRequestBody).toString().getBytes());
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }