text: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  #include "MQClientFactory.h"
  #include "ConsumerRunningInfo.h"
  #include "Logging.h"
  #include "MQClientManager.h"
  #include "MQVersion.h"
  #include "PullRequest.h"
  #include "Rebalance.h"
  #include "TopicPublishInfo.h"
  #include "TransactionMQProducerImpl.h"
  
  #define MAX_BUFF_SIZE 8192
  #define SAFE_BUFF_SIZE 7936  // 8192 - 256 = 7936
  #define PROCESS_NAME_BUF_SIZE 256
  
  namespace rocketmq {
  //<!***************************************************************************
  MQClientFactory::MQClientFactory(const string& clientID) : m_bFetchNSService(true) {
    m_clientId = clientID;
  }
  MQClientFactory::MQClientFactory(const string& clientID,
                                   int pullThreadNum,
                                   uint64_t tcpConnectTimeout,
                                   uint64_t tcpTransportTryLockTimeout,
                                   string unitName)
      : m_bFetchNSService(true) {
    m_clientId = clientID;
    // default Topic register;
    boost::shared_ptr<TopicPublishInfo> pDefaultTopicInfo(new TopicPublishInfo());
    m_topicPublishInfoTable[DEFAULT_TOPIC] = pDefaultTopicInfo;
    m_pClientRemotingProcessor.reset(new ClientRemotingProcessor(this));
    m_pClientAPIImpl.reset(new MQClientAPIImpl(m_clientId, m_pClientRemotingProcessor.get(), pullThreadNum,
                                               tcpConnectTimeout, tcpTransportTryLockTimeout, unitName));
    m_serviceState = CREATE_JUST;
    LOG_DEBUG("MQClientFactory construct");
  }
  
  MQClientFactory::~MQClientFactory() {
    LOG_INFO("MQClientFactory:%s destruct", m_clientId.c_str());
  
    for (TRDMAP::iterator itp = m_topicRouteTable.begin(); itp != m_topicRouteTable.end(); ++itp) {
      delete itp->second;
    }
  
    m_producerTable.clear();
    m_consumerTable.clear();
    m_topicRouteTable.clear();
    m_brokerAddrTable.clear();
    m_topicPublishInfoTable.clear();
  
    m_pClientAPIImpl = NULL;
  }
  
  void MQClientFactory::start() {
    switch (m_serviceState) {
      case CREATE_JUST:
        LOG_INFO("MQClientFactory:%s start", m_clientId.c_str());
        m_serviceState = START_FAILED;
        //<!start time task;
        m_async_service_thread.reset(
            new boost::thread(boost::bind(&MQClientFactory::startScheduledTask, this, m_bFetchNSService)));
        m_serviceState = RUNNING;
        break;
      case RUNNING:
        LOG_INFO("The Factory object:%s start before with now state:%d", m_clientId.c_str(), m_serviceState);
        break;
      case SHUTDOWN_ALREADY:
      case START_FAILED:
        LOG_INFO("The Factory object:%s start failed with fault state:%d", m_clientId.c_str(), m_serviceState);
        break;
      default:
        break;
    }
  }
  
  void MQClientFactory::updateTopicRouteInfo(boost::system::error_code& ec,
                                             boost::shared_ptr<boost::asio::deadline_timer> t) {
    if ((getConsumerTableSize() == 0) && (getProducerTableSize() == 0)) {
      return;
    }
  
    set<string> topicList;
    //<!Consumer;
    getTopicListFromConsumerSubscription(topicList);
  
    //<!Producer;
    getTopicListFromTopicPublishInfo(topicList);
  
    //<! update;
    {
      SessionCredentials session_credentials;
      getSessionCredentialsFromOneOfProducerOrConsumer(session_credentials);
      set<string>::iterator it = topicList.begin();
      for (; it != topicList.end(); ++it) {
        updateTopicRouteInfoFromNameServer(*it, session_credentials);
      }
    }
  
    boost::system::error_code e;
    t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(30), e);
    t->async_wait(boost::bind(&MQClientFactory::updateTopicRouteInfo, this, ec, t));
  }
  
  TopicRouteData* MQClientFactory::getTopicRouteData(const string& topic) {
    boost::lock_guard<boost::mutex> lock(m_topicRouteTableMutex);
    if (m_topicRouteTable.find(topic) != m_topicRouteTable.end()) {
      return m_topicRouteTable[topic];
    }
    return NULL;
  }
  
  void MQClientFactory::addTopicRouteData(const string& topic, TopicRouteData* pTopicRouteData) {
    boost::lock_guard<boost::mutex> lock(m_topicRouteTableMutex);
    if (m_topicRouteTable.find(topic) != m_topicRouteTable.end()) {
      delete m_topicRouteTable[topic];
      m_topicRouteTable.erase(topic);
    }
    m_topicRouteTable[topic] = pTopicRouteData;
  }
  
  boost::shared_ptr<TopicPublishInfo> MQClientFactory::tryToFindTopicPublishInfo(
      const string& topic,
      const SessionCredentials& session_credentials) {
    boost::lock_guard<boost::mutex> lock(m_topicPublishInfoLock);  // add topicPublishInfoLock to avoid con-current
                                                                   // excuting updateTopicRouteInfoFromNameServer
                                                                   // when producer send msg  before topicRouteInfo
                                                                   // was got;
    if (!isTopicInfoValidInTable(topic)) {
      updateTopicRouteInfoFromNameServer(topic, session_credentials);
    }
    //<!if not exist ,update default topic;
    if (!isTopicInfoValidInTable(topic)) {
      LOG_INFO("updateTopicRouteInfoFromNameServer with default");
      updateTopicRouteInfoFromNameServer(topic, session_credentials, true);
    }
  
    if (!isTopicInfoValidInTable(topic)) {
      LOG_WARN("tryToFindTopicPublishInfo null:%s", topic.c_str());
      boost::shared_ptr<TopicPublishInfo> pTopicPublishInfo;
      return pTopicPublishInfo;
    }
  
    return getTopicPublishInfoFromTable(topic);
  }
  
  bool MQClientFactory::updateTopicRouteInfoFromNameServer(const string& topic,
                                                           const SessionCredentials& session_credentials,
                                                           bool isDefault /* = false */) {
    boost::lock_guard<boost::mutex> lock(m_factoryLock);
    unique_ptr<TopicRouteData> pTopicRouteData;
    LOG_DEBUG("updateTopicRouteInfoFromNameServer start. Topic:%s", topic.c_str());
  
    if (isDefault) {
      pTopicRouteData.reset(
          m_pClientAPIImpl->getTopicRouteInfoFromNameServer(DEFAULT_TOPIC, 1000 * 5, session_credentials));
      if (pTopicRouteData != NULL) {
        vector<QueueData>& queueDatas = pTopicRouteData->getQueueDatas();
        vector<QueueData>::iterator it = queueDatas.begin();
        for (; it != queueDatas.end(); ++it) {
          int queueNums = std::min(4, it->readQueueNums);
          it->readQueueNums = queueNums;
          it->writeQueueNums = queueNums;
        }
      }
      LOG_DEBUG("getTopicRouteInfoFromNameServer is null for topic :%s", topic.c_str());
    } else {
      pTopicRouteData.reset(m_pClientAPIImpl->getTopicRouteInfoFromNameServer(topic, 1000 * 5, session_credentials));
    }
  
    if (pTopicRouteData != NULL) {
      LOG_DEBUG("updateTopicRouteInfoFromNameServer has data");
      TopicRouteData* pTemp = getTopicRouteData(topic);
      bool changed = true;
      if (pTemp != NULL) {
        changed = !(*pTemp == *pTopicRouteData);
      }
  
      if (getConsumerTableSize() > 0) {
        vector<MQMessageQueue> mqs;
        topicRouteData2TopicSubscribeInfo(topic, pTopicRouteData.get(), mqs);
        updateConsumerSubscribeTopicInfo(topic, mqs);
      }
  
      if (changed) {
        //<!update Broker addr
        LOG_INFO("updateTopicRouteInfoFromNameServer changed:%s", topic.c_str());
        vector<BrokerData> brokerList = pTopicRouteData->getBrokerDatas();
        vector<BrokerData>::iterator it = brokerList.begin();
        for (; it != brokerList.end(); ++it) {
          LOG_INFO("updateTopicRouteInfoFromNameServer changed with broker name:%s", (*it).brokerName.c_str());
          addBrokerToAddrMap((*it).brokerName, (*it).brokerAddrs);
        }
  
        //<! update publish info;
        {
          boost::shared_ptr<TopicPublishInfo> publishInfo(topicRouteData2TopicPublishInfo(topic, pTopicRouteData.get()));
          addTopicInfoToTable(topic, publishInfo);  // erase first, then add
        }
  
        //<! update subscribe info
        addTopicRouteData(topic, pTopicRouteData.release());
      }
      LOG_DEBUG("updateTopicRouteInfoFromNameServer end:%s", topic.c_str());
      return true;
    }
    LOG_DEBUG("updateTopicRouteInfoFromNameServer end:%s", topic.c_str());
    return false;
  }
  
  boost::shared_ptr<TopicPublishInfo> MQClientFactory::topicRouteData2TopicPublishInfo(const string& topic,
                                                                                       TopicRouteData* pRoute) {
    boost::shared_ptr<TopicPublishInfo> info(new TopicPublishInfo());
    string OrderTopicConf = pRoute->getOrderTopicConf();
    //<! order msg
    if (!OrderTopicConf.empty()) {
      // "broker-a:8";"broker-b:8"
      vector<string> brokers;
      UtilAll::Split(brokers, OrderTopicConf, ';');
      for (size_t i = 0; i < brokers.size(); i++) {
        vector<string> item;
        UtilAll::Split(item, brokers[i], ':');
        size_t nums = (item.size() > 1) ? atoi(item[1].c_str()) : 0;
        for (size_t i = 0; i < nums; i++) {
          MQMessageQueue mq(topic, item[0], i);
          info->updateMessageQueueList(mq);
        }
      }
    }
    //<!no order msg
    else {
      vector<QueueData>& queueDatas = pRoute->getQueueDatas();
      vector<QueueData>::iterator it = queueDatas.begin();
      for (; it != queueDatas.end(); ++it) {
        QueueData& qd = (*it);
        if (PermName::isWriteable(qd.perm)) {
          string addr = findBrokerAddressInPublish(qd.brokerName);
          if (addr.empty()) {
            continue;
          }
          for (int i = 0; i < qd.writeQueueNums; i++) {
            MQMessageQueue mq(topic, qd.brokerName, i);
            info->updateMessageQueueList(mq);
          }
        }
      }
    }
    return info;
  }
  
  void MQClientFactory::topicRouteData2TopicSubscribeInfo(const string& topic,
                                                          TopicRouteData* pRoute,
                                                          vector<MQMessageQueue>& mqs) {
    mqs.clear();
    vector<QueueData>& queueDatas = pRoute->getQueueDatas();
    vector<QueueData>::iterator it = queueDatas.begin();
    for (; it != queueDatas.end(); ++it) {
      QueueData& qd = (*it);
      if (PermName::isReadable(qd.perm)) {
        for (int i = 0; i < qd.readQueueNums; i++) {
          MQMessageQueue mq(topic, qd.brokerName, i);
          mqs.push_back(mq);
        }
      }
    }
  }
  
  void MQClientFactory::shutdown() {
    if (getConsumerTableSize() != 0)
      return;
  
    if (getProducerTableSize() != 0)
      return;
  
    switch (m_serviceState) {
      case RUNNING: {
        if (m_consumer_async_service_thread) {
          m_consumer_async_ioService.stop();
          m_consumer_async_service_thread->interrupt();
          m_consumer_async_service_thread->join();
        }
        m_async_ioService.stop();
        m_async_service_thread->interrupt();
        m_async_service_thread->join();
        m_pClientAPIImpl->stopAllTcpTransportThread();  // Note: stop all
                                                        // TcpTransport Threads
                                                        // and release all
                                                        // responseFuture
                                                        // conditions
        m_serviceState = SHUTDOWN_ALREADY;
        LOG_INFO("MQClientFactory:%s shutdown", m_clientId.c_str());
        break;
      }
      case SHUTDOWN_ALREADY:
      case CREATE_JUST:
        break;
      default:
        break;
    }
  
    MQClientManager::getInstance()->removeClientFactory(m_clientId);
  }
  
  bool MQClientFactory::registerProducer(MQProducer* pProducer) {
    string groupName = pProducer->getGroupName();
    string namesrvaddr = pProducer->getNamesrvAddr();
    if (groupName.empty()) {
      return false;
    }
  
    if (!addProducerToTable(groupName, pProducer)) {
      return false;
    }
  
    LOG_DEBUG("registerProducer success:%s", groupName.c_str());
    //<!set nameserver;
    if (namesrvaddr.empty()) {
      string nameSrvDomain(pProducer->getNamesrvDomain());
      if (!nameSrvDomain.empty())
        m_nameSrvDomain = nameSrvDomain;
      pProducer->setNamesrvAddr(m_pClientAPIImpl->fetchNameServerAddr(m_nameSrvDomain));
    } else {
      m_bFetchNSService = false;
      m_pClientAPIImpl->updateNameServerAddr(namesrvaddr);
      LOG_INFO("user specfied name server address: %s", namesrvaddr.c_str());
    }
    return true;
  }
  
  void MQClientFactory::unregisterProducer(MQProducer* pProducer) {
    string groupName = pProducer->getGroupName();
    unregisterClient(groupName, "", pProducer->getSessionCredentials());
  
    eraseProducerFromTable(groupName);
  }
  
  bool MQClientFactory::registerConsumer(MQConsumer* pConsumer) {
    string groupName = pConsumer->getGroupName();
    string namesrvaddr = pConsumer->getNamesrvAddr();
    if (groupName.empty()) {
      return false;
    }
  
    if (!addConsumerToTable(groupName, pConsumer)) {
      return false;
    }
    LOG_DEBUG("registerConsumer success:%s", groupName.c_str());
    //<!set nameserver;
    if (namesrvaddr.empty()) {
      string nameSrvDomain(pConsumer->getNamesrvDomain());
      if (!nameSrvDomain.empty())
        m_nameSrvDomain = nameSrvDomain;
      pConsumer->setNamesrvAddr(m_pClientAPIImpl->fetchNameServerAddr(m_nameSrvDomain));
    } else {
      m_bFetchNSService = false;
      m_pClientAPIImpl->updateNameServerAddr(namesrvaddr);
      LOG_INFO("user specfied name server address: %s", namesrvaddr.c_str());
    }
  
    return true;
  }
  
  void MQClientFactory::unregisterConsumer(MQConsumer* pConsumer) {
    string groupName = pConsumer->getGroupName();
    unregisterClient("", groupName, pConsumer->getSessionCredentials());
  
    eraseConsumerFromTable(groupName);
  }
  
  MQProducer* MQClientFactory::selectProducer(const string& producerName) {
    boost::lock_guard<boost::mutex> lock(m_producerTableMutex);
    if (m_producerTable.find(producerName) != m_producerTable.end()) {
      return m_producerTable[producerName];
    }
    return NULL;
  }
  
  bool MQClientFactory::getSessionCredentialFromProducerTable(SessionCredentials& sessionCredentials) {
    boost::lock_guard<boost::mutex> lock(m_producerTableMutex);
    for (MQPMAP::iterator it = m_producerTable.begin(); it != m_producerTable.end(); ++it) {
      if (it->second)
        sessionCredentials = it->second->getSessionCredentials();
    }
  
    if (sessionCredentials.isValid())
      return true;
  
    return false;
  }
  
  bool MQClientFactory::addProducerToTable(const string& producerName, MQProducer* pMQProducer) {
    boost::lock_guard<boost::mutex> lock(m_producerTableMutex);
    if (m_producerTable.find(producerName) != m_producerTable.end())
      return false;
    m_producerTable[producerName] = pMQProducer;
    return true;
  }
  
  void MQClientFactory::eraseProducerFromTable(const string& producerName) {
    boost::lock_guard<boost::mutex> lock(m_producerTableMutex);
    if (m_producerTable.find(producerName) != m_producerTable.end())
      m_producerTable.erase(producerName);
  }
  
  int MQClientFactory::getProducerTableSize() {
    boost::lock_guard<boost::mutex> lock(m_producerTableMutex);
    return m_producerTable.size();
  }
  
  void MQClientFactory::insertProducerInfoToHeartBeatData(HeartbeatData* pHeartbeatData) {
    boost::lock_guard<boost::mutex> lock(m_producerTableMutex);
    for (MQPMAP::iterator it = m_producerTable.begin(); it != m_producerTable.end(); ++it) {
      ProducerData producerData;
      producerData.groupName = it->first;
      pHeartbeatData->insertDataToProducerDataSet(producerData);
    }
  }
  
  MQConsumer* MQClientFactory::selectConsumer(const string& group) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    if (m_consumerTable.find(group) != m_consumerTable.end()) {
      return m_consumerTable[group];
    }
    return NULL;
  }
  
  bool MQClientFactory::getSessionCredentialFromConsumerTable(SessionCredentials& sessionCredentials) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {
      if (it->second)
        sessionCredentials = it->second->getSessionCredentials();
    }
  
    if (sessionCredentials.isValid())
      return true;
  
    return false;
  }
  
  bool MQClientFactory::getSessionCredentialFromConsumer(const string& consumerGroup,
                                                         SessionCredentials& sessionCredentials) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    if (m_consumerTable.find(consumerGroup) != m_consumerTable.end()) {
      sessionCredentials = m_consumerTable[consumerGroup]->getSessionCredentials();
    }
  
    if (sessionCredentials.isValid())
      return true;
  
    return false;
  }
  
  bool MQClientFactory::addConsumerToTable(const string& consumerName, MQConsumer* pMQConsumer) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    if (m_consumerTable.find(consumerName) != m_consumerTable.end())
      return false;
    m_consumerTable[consumerName] = pMQConsumer;
    return true;
  }
  
  void MQClientFactory::eraseConsumerFromTable(const string& consumerName) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    if (m_consumerTable.find(consumerName) != m_consumerTable.end())
      m_consumerTable.erase(consumerName);  // do not need freee pConsumer, as it
                                            // was allocated by user
    else
      LOG_WARN("could not find consumer:%s from table", consumerName.c_str());
  }
  
  int MQClientFactory::getConsumerTableSize() {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    return m_consumerTable.size();
  }
  
  void MQClientFactory::getTopicListFromConsumerSubscription(set<string>& topicList) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {
      vector<SubscriptionData> result;
      it->second->getSubscriptions(result);
  
      vector<SubscriptionData>::iterator iter = result.begin();
      for (; iter != result.end(); ++iter) {
        topicList.insert((*iter).getTopic());
      }
    }
  }
  
  void MQClientFactory::updateConsumerSubscribeTopicInfo(const string& topic, vector<MQMessageQueue> mqs) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {
      it->second->updateTopicSubscribeInfo(topic, mqs);
    }
  }
  
  void MQClientFactory::insertConsumerInfoToHeartBeatData(HeartbeatData* pHeartbeatData) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {
      MQConsumer* pConsumer = it->second;
      ConsumerData consumerData;
      consumerData.groupName = pConsumer->getGroupName();
      consumerData.consumeType = pConsumer->getConsumeType();
      consumerData.messageModel = pConsumer->getMessageModel();
      consumerData.consumeFromWhere = pConsumer->getConsumeFromWhere();
  
      //<!fill data;
      vector<SubscriptionData> result;
      pConsumer->getSubscriptions(result);
      consumerData.subscriptionDataSet.swap(result);
  
      pHeartbeatData->insertDataToConsumerDataSet(consumerData);
    }
  }
  
  void MQClientFactory::addTopicInfoToTable(const string& topic, boost::shared_ptr<TopicPublishInfo> pTopicPublishInfo) {
    boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);
    if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {
      m_topicPublishInfoTable.erase(topic);
    }
    m_topicPublishInfoTable[topic] = pTopicPublishInfo;
  }
  
  void MQClientFactory::eraseTopicInfoFromTable(const string& topic) {
    boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);
    if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {
      m_topicPublishInfoTable.erase(topic);
    }
  }
  
  bool MQClientFactory::isTopicInfoValidInTable(const string& topic) {
    boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);
    if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {
      if (m_topicPublishInfoTable[topic]->ok())
        return true;
    }
    return false;
  }
  
  boost::shared_ptr<TopicPublishInfo> MQClientFactory::getTopicPublishInfoFromTable(const string& topic) {
    boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);
    if (m_topicPublishInfoTable.find(topic) != m_topicPublishInfoTable.end()) {
      return m_topicPublishInfoTable[topic];
    }
    boost::shared_ptr<TopicPublishInfo> pTopicPublishInfo;
    return pTopicPublishInfo;
  }
  
  void MQClientFactory::getTopicListFromTopicPublishInfo(set<string>& topicList) {
    boost::lock_guard<boost::mutex> lock(m_topicPublishInfoTableMutex);
    for (TPMap::iterator itp = m_topicPublishInfoTable.begin(); itp != m_topicPublishInfoTable.end(); ++itp) {
      topicList.insert(itp->first);
    }
  }
  
  void MQClientFactory::clearBrokerAddrMap() {
    boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);
    m_brokerAddrTable.clear();
  }
  
  bool MQClientFactory::isBrokerAddressInUse(const std::string& address) {
    if (m_topicRouteTableMutex.try_lock()) {
      boost::lock_guard<boost::mutex> lk(m_topicRouteTableMutex, boost::adopt_lock_t());
      for (TRDMAP::iterator it = m_topicRouteTable.begin(); it != m_topicRouteTable.end(); it++) {
        TopicRouteData* topicRouteData = it->second;
        vector<BrokerData>& brokerData = topicRouteData->getBrokerDatas();
        for (vector<BrokerData>::iterator next = brokerData.begin(); next != brokerData.end(); next++) {
          map<int, string>& brokerAddresses = next->brokerAddrs;
          for (map<int, string>::iterator entry = brokerAddresses.begin(); entry != brokerAddresses.end(); entry++) {
            if (address == entry->second) {
              return true;
            }
          }
        }
      }
      return false;
    } else {
      LOG_WARN("Cannot lock m_topicRouteTableMutex. Assume %s is still in use", address.c_str());
      return true;
    }
  }
  void MQClientFactory::addBrokerToAddrMap(const string& brokerName, map<int, string>& brokerAddrs) {
    boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);
    if (m_brokerAddrTable.find(brokerName) != m_brokerAddrTable.end()) {
      m_brokerAddrTable.erase(brokerName);
    }
    m_brokerAddrTable[brokerName] = brokerAddrs;
  }
  
  MQClientFactory::BrokerAddrMAP MQClientFactory::getBrokerAddrMap() {
    boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);
    return m_brokerAddrTable;
  }
  
  string MQClientFactory::findBrokerAddressInPublish(const string& brokerName) {
    /*reslove the concurrent access m_brokerAddrTable by
    findBrokerAddressInPublish(called by sendKernlImpl) And
    sendHeartbeatToAllBroker, which leads hign RT of sendMsg
    1. change m_brokerAddrTable from hashMap to map;
    2. do not add m_factoryLock here, but copy m_brokerAddrTable,
        this is used to avoid con-current access m_factoryLock by
    findBrokerAddressInPublish(called by sendKernlImpl) And
    updateTopicRouteInfoFromNameServer
  
     Note: after copying m_brokerAddrTable, updateTopicRouteInfoFromNameServer
    modify m_brokerAddrTable imediatly,
             after 1st send fail, producer will get topicPushlibshInfo again
    before next try, so 2nd try will get correct broker to send ms;
     */
    BrokerAddrMAP brokerTable(getBrokerAddrMap());
    string brokerAddr;
    bool found = false;
  
    if (brokerTable.find(brokerName) != brokerTable.end()) {
      map<int, string> brokerMap(brokerTable[brokerName]);
      map<int, string>::iterator it1 = brokerMap.find(MASTER_ID);
      if (it1 != brokerMap.end()) {
        brokerAddr = it1->second;
        found = true;
      }
    }
  
    brokerTable.clear();
    if (found)
      return brokerAddr;
  
    return "";
  }
  
  FindBrokerResult* MQClientFactory::findBrokerAddressInSubscribe(const string& brokerName,
                                                                  int brokerId,
                                                                  bool onlyThisBroker) {
    string brokerAddr;
    bool slave = false;
    bool found = false;
    BrokerAddrMAP brokerTable(getBrokerAddrMap());
  
    if (brokerTable.find(brokerName) != brokerTable.end()) {
      map<int, string> brokerMap(brokerTable[brokerName]);
      if (!brokerMap.empty()) {
        auto iter = brokerMap.find(brokerId);
        if (iter != brokerMap.end()) {
          brokerAddr = iter->second;
          slave = (brokerId != MASTER_ID);
          found = true;
        } else if (!onlyThisBroker) {  // not only from master
          iter = brokerMap.begin();
          brokerAddr = iter->second;
          slave = iter->first != MASTER_ID;
          found = true;
        }
      }
    }
  
    brokerTable.clear();
  
    if (found) {
      return new FindBrokerResult(brokerAddr, slave);
    }
  
    return nullptr;
  }
  
  FindBrokerResult* MQClientFactory::findBrokerAddressInAdmin(const string& brokerName) {
    BrokerAddrMAP brokerTable(getBrokerAddrMap());
    bool found = false;
    bool slave = false;
    string brokerAddr;
  
    if (brokerTable.find(brokerName) != brokerTable.end()) {
      map<int, string> brokerMap(brokerTable[brokerName]);
      map<int, string>::iterator it1 = brokerMap.begin();
      if (it1 != brokerMap.end()) {
        slave = (it1->first != MASTER_ID);
        found = true;
        brokerAddr = it1->second;
      }
    }
  
    brokerTable.clear();
    if (found)
      return new FindBrokerResult(brokerAddr, slave);
  
    return NULL;
  }
  
  void MQClientFactory::checkTransactionState(const std::string& addr,
                                              const MQMessageExt& messageExt,
                                              const CheckTransactionStateRequestHeader& checkRequestHeader) {
    string group = messageExt.getProperty(MQMessage::PROPERTY_PRODUCER_GROUP);
    if (!group.empty()) {
      MQProducer* producer = selectProducer(group);
      if (producer != nullptr) {
        TransactionMQProducerImpl* transProducer = dynamic_cast<TransactionMQProducerImpl*>(producer);
        if (transProducer != nullptr) {
          transProducer->checkTransactionState(addr, messageExt, checkRequestHeader.m_tranStateTableOffset,
                                               checkRequestHeader.m_commitLogOffset, checkRequestHeader.m_msgId,
                                               checkRequestHeader.m_transactionId, checkRequestHeader.m_offsetMsgId);
        } else {
          LOG_ERROR("checkTransactionState, producer not TransactionMQProducer failed, msg:%s",
                    messageExt.toString().data());
        }
      } else {
        LOG_ERROR("checkTransactionState, pick producer by group[%s] failed, msg:%s", group.data(),
                  messageExt.toString().data());
      }
    } else {
      LOG_ERROR("checkTransactionState, pick producer group failed, msg:%s", messageExt.toString().data());
    }
  }
  
  MQClientAPIImpl* MQClientFactory::getMQClientAPIImpl() {
    return m_pClientAPIImpl.get();
  }
  
  void MQClientFactory::cleanOfflineBrokers() {
    LOG_DEBUG("Begin to clean offline brokers");
    boost::lock_guard<boost::mutex> lock(m_brokerAddrlock);
  
    for (BrokerAddrMAP::iterator it = m_brokerAddrTable.begin(); it != m_brokerAddrTable.end();) {
      std::string brokerName = it->first;
      map<int, std::string> brokerIdAddressMap = it->second;
  
      for (map<int, std::string>::iterator next = brokerIdAddressMap.begin(); next != brokerIdAddressMap.end();) {
        if (!isBrokerAddressInUse(next->second)) {
          LOG_INFO("Remove broker address: %s", (next->second).c_str());
          brokerIdAddressMap.erase(next++);
        } else {
          next++;
        }
      }
  
      if (brokerIdAddressMap.empty()) {
        m_brokerAddrTable.erase(it++);
        LOG_INFO("Broker name: %s is purged from client", brokerName.c_str());
      } else {
        LOG_DEBUG("Broker: %s is alive", brokerName.c_str());
        it++;
      }
    }
  
    LOG_DEBUG("Exit of cleaning offline brokers");
  }
  
  void MQClientFactory::sendHeartbeatToAllBroker() {
    BrokerAddrMAP brokerTable(getBrokerAddrMap());
    if (brokerTable.size() == 0) {
      LOG_WARN("sendheartbeat brokeradd is empty");
      return;
    }
  
    unique_ptr<HeartbeatData> heartbeatData(prepareHeartbeatData());
    bool producerEmpty = heartbeatData->isProducerDataSetEmpty();
    bool consumerEmpty = heartbeatData->isConsumerDataSetEmpty();
    if (producerEmpty && consumerEmpty) {
      LOG_WARN("sendheartbeat heartbeatData empty");
      brokerTable.clear();
      return;
    }
  
    SessionCredentials session_credentials;
    getSessionCredentialsFromOneOfProducerOrConsumer(session_credentials);
    for (BrokerAddrMAP::iterator it = brokerTable.begin(); it != brokerTable.end(); ++it) {
      map<int, string> brokerMap(it->second);
      map<int, string>::iterator it1 = brokerMap.begin();
      for (; it1 != brokerMap.end(); ++it1) {
        string& addr = it1->second;
        if (consumerEmpty && it1->first != MASTER_ID)
          continue;
  
        try {
          m_pClientAPIImpl->sendHeartbeat(addr, heartbeatData.get(), session_credentials);
        } catch (MQException& e) {
          LOG_ERROR(e.what());
        }
      }
    }
    brokerTable.clear();
  }
  
  void MQClientFactory::persistAllConsumerOffset(boost::system::error_code& ec,
                                                 boost::shared_ptr<boost::asio::deadline_timer> t) {
    {
      boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
      if (m_consumerTable.size() > 0) {
        for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {
          LOG_DEBUG("Client factory start persistAllConsumerOffset");
          it->second->persistConsumerOffset();
        }
      }
    }
  
    boost::system::error_code e;
    t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(5), e);
    t->async_wait(boost::bind(&MQClientFactory::persistAllConsumerOffset, this, ec, t));
  }
  
  HeartbeatData* MQClientFactory::prepareHeartbeatData() {
    HeartbeatData* pHeartbeatData = new HeartbeatData();
    // clientID
    pHeartbeatData->setClientID(m_clientId);
  
    // Consumer
    insertConsumerInfoToHeartBeatData(pHeartbeatData);
  
    // Producer
    insertProducerInfoToHeartBeatData(pHeartbeatData);
  
    return pHeartbeatData;
  }
  
  void MQClientFactory::timerCB_sendHeartbeatToAllBroker(boost::system::error_code& ec,
                                                         boost::shared_ptr<boost::asio::deadline_timer> t) {
    sendHeartbeatToAllBroker();
  
    boost::system::error_code e;
    t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(30), e);
    t->async_wait(boost::bind(&MQClientFactory::timerCB_sendHeartbeatToAllBroker, this, ec, t));
  }
  
  void MQClientFactory::timerCB_cleanOfflineBrokers(boost::system::error_code& ec,
                                                    boost::shared_ptr<boost::asio::deadline_timer> t) {
    cleanOfflineBrokers();
  
    boost::system::error_code e;
    t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(30), e);
    t->async_wait(boost::bind(&MQClientFactory::timerCB_cleanOfflineBrokers, this, ec, t));
  }
  
  void MQClientFactory::fetchNameServerAddr(boost::system::error_code& ec,
                                            boost::shared_ptr<boost::asio::deadline_timer> t) {
    m_pClientAPIImpl->fetchNameServerAddr(m_nameSrvDomain);
  
    boost::system::error_code e;
    t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(60 * 2), e);
    t->async_wait(boost::bind(&MQClientFactory::fetchNameServerAddr, this, ec, t));
  }
  
  void MQClientFactory::startScheduledTask(bool startFetchNSService) {
    boost::asio::io_service::work work(m_async_ioService);  // avoid async io
                                                            // service stops after
                                                            // first timer timeout
                                                            // callback
  
    boost::system::error_code ec1;
    boost::shared_ptr<boost::asio::deadline_timer> t1 =
        boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::seconds(3));
    t1->async_wait(boost::bind(&MQClientFactory::updateTopicRouteInfo, this, ec1, t1));
  
    boost::system::error_code ec2;
    boost::shared_ptr<boost::asio::deadline_timer> t2 =
        boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::milliseconds(10));
    t2->async_wait(boost::bind(&MQClientFactory::timerCB_sendHeartbeatToAllBroker, this, ec2, t2));
  
    boost::system::error_code ec3;
    boost::shared_ptr<boost::asio::deadline_timer> t3 =
        boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::seconds(3));
    t3->async_wait(boost::bind(&MQClientFactory::timerCB_cleanOfflineBrokers, this, ec3, t3));
  
    if (startFetchNSService) {
      boost::system::error_code ec5;
      boost::shared_ptr<boost::asio::deadline_timer> t5 =
          boost::make_shared<boost::asio::deadline_timer>(m_async_ioService, boost::posix_time::seconds(60 * 2));
      t5->async_wait(boost::bind(&MQClientFactory::fetchNameServerAddr, this, ec5, t5));
    }
  
    LOG_INFO("start scheduled task:%s", m_clientId.c_str());
    boost::system::error_code ec;
    m_async_ioService.run(ec);
  }
  
  void MQClientFactory::rebalanceImmediately() {
    // m_consumer_async_service_thread will be only started once for all consumer
    if (m_consumer_async_service_thread == NULL) {
      doRebalance();
      m_consumer_async_service_thread.reset(
          new boost::thread(boost::bind(&MQClientFactory::consumer_timerOperation, this)));
    }
  }
  
  void MQClientFactory::consumer_timerOperation() {
    LOG_INFO("clientFactory:%s start consumer_timerOperation", m_clientId.c_str());
    boost::asio::io_service::work work(m_consumer_async_ioService);  // avoid async io
                                                                     // service stops after
                                                                     // first timer timeout
                                                                     // callback
  
    boost::system::error_code ec1;
    boost::shared_ptr<boost::asio::deadline_timer> t1 =
        boost::make_shared<boost::asio::deadline_timer>(m_consumer_async_ioService, boost::posix_time::seconds(10));
    t1->async_wait(boost::bind(&MQClientFactory::timerCB_doRebalance, this, ec1, t1));
  
    boost::system::error_code ec2;
    boost::shared_ptr<boost::asio::deadline_timer> t2 =
        boost::make_shared<boost::asio::deadline_timer>(m_consumer_async_ioService, boost::posix_time::seconds(5));
    t2->async_wait(boost::bind(&MQClientFactory::persistAllConsumerOffset, this, ec2, t2));
  
    boost::system::error_code ec;
    m_consumer_async_ioService.run(ec);
    LOG_INFO("clientFactory:%s stop consumer_timerOperation", m_clientId.c_str());
  }
  
  void MQClientFactory::timerCB_doRebalance(boost::system::error_code& ec,
                                            boost::shared_ptr<boost::asio::deadline_timer> t) {
    doRebalance();
  
    boost::system::error_code e;
    t->expires_from_now(t->expires_from_now() + boost::posix_time::seconds(10), e);
    t->async_wait(boost::bind(&MQClientFactory::timerCB_doRebalance, this, ec, t));
  }
  
  void MQClientFactory::doRebalance() {
    LOG_DEBUG("Client factory:%s start doRebalance", m_clientId.c_str());
    if (getConsumerTableSize() > 0) {
      boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
      for (MQCMAP::iterator it = m_consumerTable.begin(); it != m_consumerTable.end(); ++it) {
        it->second->doRebalance();
      }
    }
    LOG_DEBUG("Client factory:%s finish doRebalance", m_clientId.c_str());
  }
  
  void MQClientFactory::doRebalanceByConsumerGroup(const string& consumerGroup) {
    boost::lock_guard<boost::recursive_mutex> lock(m_consumerTableMutex);
    if (m_consumerTable.find(consumerGroup) != m_consumerTable.end()) {
      LOG_INFO("Client factory:%s start dorebalance for consumer:%s", m_clientId.c_str(), consumerGroup.c_str());
      MQConsumer* pMQConsumer = m_consumerTable[consumerGroup];
      pMQConsumer->doRebalance();
    }
  }
  
  void MQClientFactory::endTransactionOneway(const MQMessageQueue& mq,
                                             EndTransactionRequestHeader* requestHeader,
                                             const SessionCredentials& sessionCredentials) {
    string brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    string remark = "";
    if (!brokerAddr.empty()) {
      try {
        getMQClientAPIImpl()->endTransactionOneway(brokerAddr, requestHeader, remark, sessionCredentials);
      } catch (MQException& e) {
        LOG_ERROR("endTransactionOneway exception:%s", e.what());
        throw e;
      }
    } else {
      THROW_MQEXCEPTION(MQClientException, "The broker[" + mq.getBrokerName() + "] not exist", -1);
    }
  }
  
  void MQClientFactory::unregisterClient(const string& producerGroup,
                                         const string& consumerGroup,
                                         const SessionCredentials& sessionCredentials) {
    BrokerAddrMAP brokerTable(getBrokerAddrMap());
    for (BrokerAddrMAP::iterator it = brokerTable.begin(); it != brokerTable.end(); ++it) {
      map<int, string> brokerMap(it->second);
      map<int, string>::iterator it1 = brokerMap.begin();
      for (; it1 != brokerMap.end(); ++it1) {
        string& addr = it1->second;
        m_pClientAPIImpl->unregisterClient(addr, m_clientId, producerGroup, consumerGroup, sessionCredentials);
      }
    }
  }
  
  //<!************************************************************************
  void MQClientFactory::fetchSubscribeMessageQueues(const string& topic,
                                                    vector<MQMessageQueue>& mqs,
                                                    const SessionCredentials& sessionCredentials) {
    TopicRouteData* pTopicRouteData = getTopicRouteData(topic);
    if (pTopicRouteData == NULL) {
      updateTopicRouteInfoFromNameServer(topic, sessionCredentials);
      pTopicRouteData = getTopicRouteData(topic);
    }
    if (pTopicRouteData != NULL) {
      topicRouteData2TopicSubscribeInfo(topic, pTopicRouteData, mqs);
      if (mqs.empty()) {
        THROW_MQEXCEPTION(MQClientException, "Can not find Message Queue", -1);
      }
      return;
    }
    THROW_MQEXCEPTION(MQClientException, "Can not find Message Queue", -1);
  }
  
  //<!***************************************************************************
  void MQClientFactory::createTopic(const string& key,
                                    const string& newTopic,
                                    int queueNum,
                                    const SessionCredentials& sessionCredentials) {}
  
  int64 MQClientFactory::minOffset(const MQMessageQueue& mq, const SessionCredentials& sessionCredentials) {
    string brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    if (brokerAddr.empty()) {
      updateTopicRouteInfoFromNameServer(mq.getTopic(), sessionCredentials);
      brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    }
  
    if (!brokerAddr.empty()) {
      try {
        return m_pClientAPIImpl->getMinOffset(brokerAddr, mq.getTopic(), mq.getQueueId(), 1000 * 3, sessionCredentials);
      } catch (MQException& e) {
        LOG_ERROR(e.what());
      }
    }
    THROW_MQEXCEPTION(MQClientException, "The broker is not exist", -1);
  }
  
  int64 MQClientFactory::maxOffset(const MQMessageQueue& mq, const SessionCredentials& sessionCredentials) {
    string brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    if (brokerAddr.empty()) {
      updateTopicRouteInfoFromNameServer(mq.getTopic(), sessionCredentials);
      brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    }
  
    if (!brokerAddr.empty()) {
      try {
        return m_pClientAPIImpl->getMaxOffset(brokerAddr, mq.getTopic(), mq.getQueueId(), 1000 * 3, sessionCredentials);
      } catch (MQException& e) {
        THROW_MQEXCEPTION(MQClientException, "Invoke Broker exception", -1);
      }
    }
    THROW_MQEXCEPTION(MQClientException, "The broker is not exist", -1);
  }
  
  int64 MQClientFactory::searchOffset(const MQMessageQueue& mq,
                                      int64 timestamp,
                                      const SessionCredentials& sessionCredentials) {
    string brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    if (brokerAddr.empty()) {
      updateTopicRouteInfoFromNameServer(mq.getTopic(), sessionCredentials);
      brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    }
  
    if (!brokerAddr.empty()) {
      try {
        return m_pClientAPIImpl->searchOffset(brokerAddr, mq.getTopic(), mq.getQueueId(), timestamp, 1000 * 3,
                                              sessionCredentials);
      } catch (MQException& e) {
        THROW_MQEXCEPTION(MQClientException, "Invoke Broker exception", -1);
      }
    }
    THROW_MQEXCEPTION(MQClientException, "The broker is not exist", -1);
  }
  
  MQMessageExt* MQClientFactory::viewMessage(const string& msgId, const SessionCredentials& sessionCredentials) {
    try {
      return NULL;
    } catch (MQException& e) {
      THROW_MQEXCEPTION(MQClientException, "message id illegal", -1);
    }
  }
  
  int64 MQClientFactory::earliestMsgStoreTime(const MQMessageQueue& mq, const SessionCredentials& sessionCredentials) {
    string brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    if (brokerAddr.empty()) {
      updateTopicRouteInfoFromNameServer(mq.getTopic(), sessionCredentials);
      brokerAddr = findBrokerAddressInPublish(mq.getBrokerName());
    }
  
    if (!brokerAddr.empty()) {
      try {
        return m_pClientAPIImpl->getEarliestMsgStoretime(brokerAddr, mq.getTopic(), mq.getQueueId(), 1000 * 3,
                                                         sessionCredentials);
      } catch (MQException& e) {
        THROW_MQEXCEPTION(MQClientException, "Invoke Broker exception", -1);
      }
    }
    THROW_MQEXCEPTION(MQClientException, "The broker is not exist", -1);
  }
  
  QueryResult MQClientFactory::queryMessage(const string& topic,
                                            const string& key,
                                            int maxNum,
                                            int64 begin,
                                            int64 end,
                                            const SessionCredentials& sessionCredentials) {
    THROW_MQEXCEPTION(MQClientException, "queryMessage", -1);
  }
  
  void MQClientFactory::findConsumerIds(const string& topic,
                                        const string& group,
                                        vector<string>& cids,
                                        const SessionCredentials& sessionCredentials) {
    string brokerAddr;
    TopicRouteData* pTopicRouteData = getTopicRouteData(topic);
    if (pTopicRouteData == NULL) {
      updateTopicRouteInfoFromNameServer(topic, sessionCredentials);
      pTopicRouteData = getTopicRouteData(topic);
    }
    if (pTopicRouteData != NULL) {
      brokerAddr = pTopicRouteData->selectBrokerAddr();
    }
  
    if (!brokerAddr.empty()) {
      try {
        LOG_INFO("getConsumerIdList from broker:%s", brokerAddr.c_str());
        return m_pClientAPIImpl->getConsumerIdListByGroup(brokerAddr, group, cids, 5000, sessionCredentials);
      } catch (MQException& e) {
        LOG_ERROR(e.what());
      }
    }
  }
  
  void MQClientFactory::resetOffset(const string& group,
                                    const string& topic,
                                    const map<MQMessageQueue, int64>& offsetTable) {
    MQConsumer* pConsumer = selectConsumer(group);
    if (pConsumer) {
      map<MQMessageQueue, int64>::const_iterator it = offsetTable.begin();
  
      for (; it != offsetTable.end(); ++it) {
        MQMessageQueue mq = it->first;
        boost::weak_ptr<PullRequest> pullRequest = pConsumer->getRebalance()->getPullRequest(mq);
        boost::shared_ptr<PullRequest> pullreq = pullRequest.lock();
        // PullRequest* pullreq = pConsumer->getRebalance()->getPullRequest(mq);
        if (pullreq) {
          pullreq->setDropped(true);
          LOG_INFO("resetOffset setDropped for mq:%s", mq.toString().data());
          pullreq->clearAllMsgs();
          pullreq->updateQueueMaxOffset(it->second);
        } else {
          LOG_ERROR("no corresponding pullRequest found for topic:%s", topic.c_str());
        }
      }
  
      for (it = offsetTable.begin(); it != offsetTable.end(); ++it) {
        MQMessageQueue mq = it->first;
        if (topic == mq.getTopic()) {
          LOG_INFO("offset sets to:%lld", it->second);
          pConsumer->updateConsumeOffset(mq, it->second);
        }
      }
      pConsumer->persistConsumerOffsetByResetOffset();
  
      boost::this_thread::sleep_for(boost::chrono::milliseconds(10));
  
      for (it = offsetTable.begin(); it != offsetTable.end(); ++it) {
        MQMessageQueue mq = it->first;
        if (topic == mq.getTopic()) {
          LOG_DEBUG("resetOffset sets to:%lld for mq:%s", it->second, mq.toString().c_str());
          pConsumer->updateConsumeOffset(mq, it->second);
        }
      }
      pConsumer->persistConsumerOffsetByResetOffset();
  
      for (it = offsetTable.begin(); it != offsetTable.end(); ++it) {
        MQMessageQueue mq = it->first;
        if (topic == mq.getTopic()) {
          pConsumer->removeConsumeOffset(mq);
        }
      }
  
      // do call pConsumer->doRebalance directly here, as it is conflict with
      // timerCB_doRebalance;
      doRebalanceByConsumerGroup(pConsumer->getGroupName());
    } else {
      LOG_ERROR("no corresponding consumer found for group:%s", group.c_str());
    }
  }
  
  ConsumerRunningInfo* MQClientFactory::consumerRunningInfo(const string& consumerGroup) {
    MQConsumer* pConsumer = selectConsumer(consumerGroup);
    if (pConsumer) {
      ConsumerRunningInfo* runningInfo = pConsumer->getConsumerRunningInfo();
      if (runningInfo) {
        runningInfo->setProperty(ConsumerRunningInfo::PROP_NAMESERVER_ADDR, pConsumer->getNamesrvAddr());
        if (pConsumer->getConsumeType() == CONSUME_PASSIVELY) {
          runningInfo->setProperty(ConsumerRunningInfo::PROP_CONSUME_TYPE, "CONSUME_PASSIVELY");
        } else {
          runningInfo->setProperty(ConsumerRunningInfo::PROP_CONSUME_TYPE, "CONSUME_ACTIVELY");
        }
        runningInfo->setProperty(
            ConsumerRunningInfo::PROP_CLIENT_VERSION,
            MQVersion::GetVersionDesc(MQVersion::s_CurrentVersion));  // MQVersion::s_CurrentVersion ));
        runningInfo->setProperty(ConsumerRunningInfo::PROP_CLIENT_SDK_VERSION,
                                 pConsumer->getClientVersionString());  // in DefaultMQClient.cpp;
  
        return runningInfo;
      }
    }
  
    LOG_ERROR("no corresponding consumer found for group:%s", consumerGroup.c_str());
    return NULL;
  }
  
  void MQClientFactory::getSessionCredentialsFromOneOfProducerOrConsumer(SessionCredentials& session_credentials) {
    // Note: on the same MQClientFactory, all producers and consumers used the
    // same
    // sessionCredentials,
    // So only need get sessionCredentials from the first one producer or consumer
    // now.
    // this function was only used by updateTopicRouteInfo() and
    // sendHeartbeatToAllBrokers() now.
    // if this strategy was changed in future, need get sessionCredentials for
    // each
    // producer and consumer.
    getSessionCredentialFromProducerTable(session_credentials);
    if (!session_credentials.isValid())
      getSessionCredentialFromConsumerTable(session_credentials);
  
    if (!session_credentials.isValid()) {
      LOG_INFO(
          "updateTopicRouteInfo: didn't get the session_credentials from any "
          "producers and consumers, please re-intialize it if application needs authentication");
    }
  }
  
  //<!************************************************************************
  }  // namespace rocketmq
  