/*
 * 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 "Rebalance.h"
#include "DefaultMQPushConsumer.h"
#include "LockBatchBody.h"
#include "Logging.h"
#include "MQClientAPIImpl.h"
#include "MQClientFactory.h"
#include "OffsetStore.h"

namespace rocketmq {
//<!************************************************************************
Rebalance::Rebalance(MQConsumer* consumer, MQClientFactory* pfactory)
    : m_pConsumer(consumer), m_pClientFactory(pfactory) {
  m_pAllocateMQStrategy = new AllocateMQAveragely();
}

Rebalance::~Rebalance() {
  {
    map<string, SubscriptionData*>::iterator it = m_subscriptionData.begin();
    for (; it != m_subscriptionData.end(); ++it)
      deleteAndZero(it->second);
    m_subscriptionData.clear();
  }
  {
    MQ2PULLREQ::iterator it = m_requestQueueTable.begin();
    for (; it != m_requestQueueTable.end(); ++it) {
      delete it->second;
      it->second = NULL;
    }
    m_requestQueueTable.clear();
  }
  m_topicSubscribeInfoTable.clear();
  m_pConsumer = NULL;
  m_pClientFactory = NULL;
  deleteAndZero(m_pAllocateMQStrategy);
}

void Rebalance::doRebalance() {
  LOG_DEBUG("start doRebalance");
  try {
    map<string, SubscriptionData*>::iterator it = m_subscriptionData.begin();
    for (; it != m_subscriptionData.end(); ++it) {
      string topic = (it->first);
      LOG_INFO("current topic is:%s", topic.c_str());
      //<!topic -> mqs
      vector<MQMessageQueue> mqAll;
      if (!getTopicSubscribeInfo(topic, mqAll)) {
        continue;
      }
      if (mqAll.empty()) {
        if (!UtilAll::startsWith_retry(topic))
          THROW_MQEXCEPTION(MQClientException, "doRebalance the topic is empty", -1);
      }

      //<!msg model;
      switch (m_pConsumer->getMessageModel()) {
        case BROADCASTING: {
          bool changed = updateRequestTableInRebalance(topic, mqAll);
          if (changed) {
            messageQueueChanged(topic, mqAll, mqAll);
          }
          break;
        }
        case CLUSTERING: {
          vector<string> cidAll;
          m_pClientFactory->findConsumerIds(topic, m_pConsumer->getGroupName(), cidAll,
                                            m_pConsumer->getSessionCredentials());

          if (cidAll.empty()) {
            /*remove the droping pullRequest changes for recovery consume fastly
                from network broken
                //drop all pullRequest
                MQ2PULLREQ::iterator it = m_requestQueueTable.begin();
                for (; it != m_requestQueueTable.end(); ++it)
                {
                    if(!(it->second->isDroped()))
                    {
                        MQMessageQueue mqtemp = it->first;
                        it->second->setDroped(true);
                        removeUnnecessaryMessageQueue(mqtemp);
                        it->second->clearAllMsgs();//add clear operation to
            avoid bad
                state when dropped pullRequest returns normal
                        LOG_INFO("find consumer failed, drop undropped mq:%s",
                mqtemp.toString().c_str());
                    }
            }*/

            THROW_MQEXCEPTION(MQClientException, "doRebalance the cidAll is empty", -1);
          }
          // log
          for (int i = 0; i < (int)cidAll.size(); ++i) {
            LOG_INFO("client id:%s of topic:%s", cidAll[i].c_str(), topic.c_str());
          }
          //<! sort;
          sort(mqAll.begin(), mqAll.end());
          sort(cidAll.begin(), cidAll.end());

          //<! allocate;
          vector<MQMessageQueue> allocateResult;
          try {
            m_pAllocateMQStrategy->allocate(m_pConsumer->getMQClientId(), mqAll, cidAll, allocateResult);
          } catch (MQException& e) {
            THROW_MQEXCEPTION(MQClientException, "allocate error", -1);
          }

          // log
          for (int i = 0; i < (int)allocateResult.size(); ++i) {
            LOG_INFO("allocate mq:%s", allocateResult[i].toString().c_str());
          }

          //<!update local;
          bool changed = updateRequestTableInRebalance(topic, allocateResult);
          if (changed) {
            messageQueueChanged(topic, mqAll, allocateResult);
            break;
          }
        }
        default:
          break;
      }
    }
  } catch (MQException& e) {
    LOG_ERROR(e.what());
  }
}

void Rebalance::persistConsumerOffset() {
  DefaultMQPushConsumer* pConsumer = static_cast<DefaultMQPushConsumer*>(m_pConsumer);
  OffsetStore* pOffsetStore = pConsumer->getOffsetStore();
  vector<MQMessageQueue> mqs;
  {
    boost::lock_guard<boost::mutex> lock(m_requestTableMutex);
    MQ2PULLREQ::iterator it = m_requestQueueTable.begin();
    for (; it != m_requestQueueTable.end(); ++it) {
      if (it->second && (!it->second->isDroped())) {
        mqs.push_back(it->first);
      }
    }
  }

  if (pConsumer->getMessageModel() == BROADCASTING) {
    pOffsetStore->persistAll(mqs);
  } else {
    vector<MQMessageQueue>::iterator it2 = mqs.begin();
    for (; it2 != mqs.end(); ++it2) {
      pOffsetStore->persist(*it2, m_pConsumer->getSessionCredentials());
    }
  }
}

void Rebalance::persistConsumerOffsetByResetOffset() {
  DefaultMQPushConsumer* pConsumer = static_cast<DefaultMQPushConsumer*>(m_pConsumer);
  OffsetStore* pOffsetStore = pConsumer->getOffsetStore();
  vector<MQMessageQueue> mqs;
  {
    boost::lock_guard<boost::mutex> lock(m_requestTableMutex);
    MQ2PULLREQ::iterator it = m_requestQueueTable.begin();
    for (; it != m_requestQueueTable.end(); ++it) {
      if (it->second) {  // even if it was dropped, also need update offset when
                         // rcv resetOffset cmd
        mqs.push_back(it->first);
      }
    }
  }
  vector<MQMessageQueue>::iterator it2 = mqs.begin();
  for (; it2 != mqs.end(); ++it2) {
    pOffsetStore->persist(*it2, m_pConsumer->getSessionCredentials());
  }
}

SubscriptionData* Rebalance::getSubscriptionData(const string& topic) {
  if (m_subscriptionData.find(topic) != m_subscriptionData.end()) {
    return m_subscriptionData[topic];
  }
  return NULL;
}

map<string, SubscriptionData*>& Rebalance::getSubscriptionInner() {
  return m_subscriptionData;
}

void Rebalance::setSubscriptionData(const string& topic, SubscriptionData* pdata) {
  if (pdata != NULL && m_subscriptionData.find(topic) == m_subscriptionData.end())
    m_subscriptionData[topic] = pdata;
}

void Rebalance::setTopicSubscribeInfo(const string& topic, vector<MQMessageQueue>& mqs) {
  if (m_subscriptionData.find(topic) != m_subscriptionData.end()) {
    {
      boost::lock_guard<boost::mutex> lock(m_topicSubscribeInfoTableMutex);
      if (m_topicSubscribeInfoTable.find(topic) != m_topicSubscribeInfoTable.end())
        m_topicSubscribeInfoTable.erase(topic);
      m_topicSubscribeInfoTable[topic] = mqs;
    }
    // log
    vector<MQMessageQueue>::iterator it = mqs.begin();
    for (; it != mqs.end(); ++it) {
      LOG_DEBUG("topic [%s] has :%s", topic.c_str(), (*it).toString().c_str());
    }
  }
}

bool Rebalance::getTopicSubscribeInfo(const string& topic, vector<MQMessageQueue>& mqs) {
  boost::lock_guard<boost::mutex> lock(m_topicSubscribeInfoTableMutex);
  if (m_topicSubscribeInfoTable.find(topic) != m_topicSubscribeInfoTable.end()) {
    mqs = m_topicSubscribeInfoTable[topic];
    return true;
  }
  return false;
}

void Rebalance::addPullRequest(MQMessageQueue mq, PullRequest* pPullRequest) {
  boost::lock_guard<boost::mutex> lock(m_requestTableMutex);
  m_requestQueueTable[mq] = pPullRequest;
}

PullRequest* Rebalance::getPullRequest(MQMessageQueue mq) {
  boost::lock_guard<boost::mutex> lock(m_requestTableMutex);
  if (m_requestQueueTable.find(mq) != m_requestQueueTable.end()) {
    return m_requestQueueTable[mq];
  }
  return NULL;
}

map<MQMessageQueue, PullRequest*> Rebalance::getPullRequestTable() {
  boost::lock_guard<boost::mutex> lock(m_requestTableMutex);
  return m_requestQueueTable;
}

void Rebalance::unlockAll(bool oneway) {
  map<string, vector<MQMessageQueue>*> brokerMqs;
  MQ2PULLREQ requestQueueTable = getPullRequestTable();
  for (MQ2PULLREQ::iterator it = requestQueueTable.begin(); it != requestQueueTable.end(); ++it) {
    if (!(it->second->isDroped())) {
      if (brokerMqs.find(it->first.getBrokerName()) == brokerMqs.end()) {
        vector<MQMessageQueue>* mqs = new vector<MQMessageQueue>;
        brokerMqs[it->first.getBrokerName()] = mqs;
      } else {
        brokerMqs[it->first.getBrokerName()]->push_back(it->first);
      }
    }
  }
  LOG_INFO("unLockAll " SIZET_FMT " broker mqs", brokerMqs.size());
  for (map<string, vector<MQMessageQueue>*>::iterator itb = brokerMqs.begin(); itb != brokerMqs.end(); ++itb) {
    unique_ptr<FindBrokerResult> pFindBrokerResult(
        m_pClientFactory->findBrokerAddressInSubscribe(itb->first, MASTER_ID, true));
    if (!pFindBrokerResult) {
      LOG_ERROR("unlockAll findBrokerAddressInSubscribe ret null for broker:%s", itb->first.data());
      continue;
    }
    unique_ptr<UnlockBatchRequestBody> unlockBatchRequest(new UnlockBatchRequestBody());
    vector<MQMessageQueue> mqs(*(itb->second));
    unlockBatchRequest->setClientId(m_pConsumer->getMQClientId());
    unlockBatchRequest->setConsumerGroup(m_pConsumer->getGroupName());
    unlockBatchRequest->setMqSet(mqs);

    try {
      m_pClientFactory->getMQClientAPIImpl()->unlockBatchMQ(pFindBrokerResult->brokerAddr, unlockBatchRequest.get(),
                                                            1000, m_pConsumer->getSessionCredentials());
      for (unsigned int i = 0; i != mqs.size(); ++i) {
        PullRequest* pullreq = getPullRequest(mqs[i]);
        if (pullreq) {
          LOG_INFO("unlockBatchMQ success of mq:%s", mqs[i].toString().c_str());
          pullreq->setLocked(true);
        } else {
          LOG_ERROR("unlockBatchMQ fails of mq:%s", mqs[i].toString().c_str());
        }
      }
    } catch (MQException& e) {
      LOG_ERROR("unlockBatchMQ fails");
    }
    deleteAndZero(itb->second);
  }
  brokerMqs.clear();
}

void Rebalance::unlock(MQMessageQueue mq) {
  unique_ptr<FindBrokerResult> pFindBrokerResult(
      m_pClientFactory->findBrokerAddressInSubscribe(mq.getBrokerName(), MASTER_ID, true));
  if (!pFindBrokerResult) {
    LOG_ERROR("unlock findBrokerAddressInSubscribe ret null for broker:%s", mq.getBrokerName().data());
    return;
  }
  unique_ptr<UnlockBatchRequestBody> unlockBatchRequest(new UnlockBatchRequestBody());
  vector<MQMessageQueue> mqs;
  mqs.push_back(mq);
  unlockBatchRequest->setClientId(m_pConsumer->getMQClientId());
  unlockBatchRequest->setConsumerGroup(m_pConsumer->getGroupName());
  unlockBatchRequest->setMqSet(mqs);

  try {
    m_pClientFactory->getMQClientAPIImpl()->unlockBatchMQ(pFindBrokerResult->brokerAddr, unlockBatchRequest.get(), 1000,
                                                          m_pConsumer->getSessionCredentials());
    for (unsigned int i = 0; i != mqs.size(); ++i) {
      PullRequest* pullreq = getPullRequest(mqs[i]);
      if (pullreq) {
        LOG_INFO("unlock success of mq:%s", mqs[i].toString().c_str());
        pullreq->setLocked(true);
      } else {
        LOG_ERROR("unlock fails of mq:%s", mqs[i].toString().c_str());
      }
    }
  } catch (MQException& e) {
    LOG_ERROR("unlock fails of mq:%s", mq.toString().c_str());
  }
}

void Rebalance::lockAll() {
  map<string, vector<MQMessageQueue>*> brokerMqs;
  MQ2PULLREQ requestQueueTable = getPullRequestTable();
  for (MQ2PULLREQ::iterator it = requestQueueTable.begin(); it != requestQueueTable.end(); ++it) {
    if (!(it->second->isDroped())) {
      string brokerKey = it->first.getBrokerName() + it->first.getTopic();
      if (brokerMqs.find(brokerKey) == brokerMqs.end()) {
        vector<MQMessageQueue>* mqs = new vector<MQMessageQueue>;
        brokerMqs[brokerKey] = mqs;
        brokerMqs[brokerKey]->push_back(it->first);
      } else {
        brokerMqs[brokerKey]->push_back(it->first);
      }
    }
  }
  LOG_INFO("LockAll " SIZET_FMT " broker mqs", brokerMqs.size());
  for (map<string, vector<MQMessageQueue>*>::iterator itb = brokerMqs.begin(); itb != brokerMqs.end(); ++itb) {
    string brokerName = (*(itb->second))[0].getBrokerName();
    unique_ptr<FindBrokerResult> pFindBrokerResult(
        m_pClientFactory->findBrokerAddressInSubscribe(brokerName, MASTER_ID, true));
    if (!pFindBrokerResult) {
      LOG_ERROR("lockAll findBrokerAddressInSubscribe ret null for broker:%s", brokerName.data());
      continue;
    }
    unique_ptr<LockBatchRequestBody> lockBatchRequest(new LockBatchRequestBody());
    lockBatchRequest->setClientId(m_pConsumer->getMQClientId());
    lockBatchRequest->setConsumerGroup(m_pConsumer->getGroupName());
    lockBatchRequest->setMqSet(*(itb->second));
    LOG_INFO("try to lock:" SIZET_FMT " mqs of broker:%s", itb->second->size(), itb->first.c_str());
    try {
      vector<MQMessageQueue> messageQueues;
      m_pClientFactory->getMQClientAPIImpl()->lockBatchMQ(pFindBrokerResult->brokerAddr, lockBatchRequest.get(),
                                                          messageQueues, 1000, m_pConsumer->getSessionCredentials());
      for (unsigned int i = 0; i != messageQueues.size(); ++i) {
        PullRequest* pullreq = getPullRequest(messageQueues[i]);
        if (pullreq) {
          LOG_INFO("lockBatchMQ success of mq:%s", messageQueues[i].toString().c_str());
          pullreq->setLocked(true);
          pullreq->setLastLockTimestamp(UtilAll::currentTimeMillis());
        } else {
          LOG_ERROR("lockBatchMQ fails of mq:%s", messageQueues[i].toString().c_str());
        }
      }
      messageQueues.clear();
    } catch (MQException& e) {
      LOG_ERROR("lockBatchMQ fails");
    }
    deleteAndZero(itb->second);
  }
  brokerMqs.clear();
}
bool Rebalance::lock(MQMessageQueue mq) {
  unique_ptr<FindBrokerResult> pFindBrokerResult(
      m_pClientFactory->findBrokerAddressInSubscribe(mq.getBrokerName(), MASTER_ID, true));
  if (!pFindBrokerResult) {
    LOG_ERROR("lock findBrokerAddressInSubscribe ret null for broker:%s", mq.getBrokerName().data());
    return false;
  }
  unique_ptr<LockBatchRequestBody> lockBatchRequest(new LockBatchRequestBody());
  lockBatchRequest->setClientId(m_pConsumer->getMQClientId());
  lockBatchRequest->setConsumerGroup(m_pConsumer->getGroupName());
  vector<MQMessageQueue> in_mqSet;
  in_mqSet.push_back(mq);
  lockBatchRequest->setMqSet(in_mqSet);
  bool lockResult = false;

  try {
    vector<MQMessageQueue> messageQueues;
    LOG_DEBUG("try to lock mq:%s", mq.toString().c_str());
    m_pClientFactory->getMQClientAPIImpl()->lockBatchMQ(pFindBrokerResult->brokerAddr, lockBatchRequest.get(),
                                                        messageQueues, 1000, m_pConsumer->getSessionCredentials());
    if (messageQueues.size() == 0) {
      LOG_ERROR("lock mq on broker:%s failed", pFindBrokerResult->brokerAddr.c_str());
      return false;
    }
    for (unsigned int i = 0; i != messageQueues.size(); ++i) {
      PullRequest* pullreq = getPullRequest(messageQueues[i]);
      if (pullreq) {
        LOG_INFO("lock success of mq:%s", messageQueues[i].toString().c_str());
        pullreq->setLocked(true);
        pullreq->setLastLockTimestamp(UtilAll::currentTimeMillis());
        lockResult = true;
      } else {
        LOG_ERROR("lock fails of mq:%s", messageQueues[i].toString().c_str());
      }
    }
    messageQueues.clear();
    return lockResult;
  } catch (MQException& e) {
    LOG_ERROR("lock fails of mq:%s", mq.toString().c_str());
    return false;
  }
}

//<!************************************************************************
RebalancePull::RebalancePull(MQConsumer* consumer, MQClientFactory* pfactory) : Rebalance(consumer, pfactory) {}

bool RebalancePull::updateRequestTableInRebalance(const string& topic, vector<MQMessageQueue>& mqsSelf) {
  return false;
}

int64 RebalancePull::computePullFromWhere(const MQMessageQueue& mq) {
  return 0;
}

void RebalancePull::messageQueueChanged(const string& topic,
                                        vector<MQMessageQueue>& mqAll,
                                        vector<MQMessageQueue>& mqDivided) {}

void RebalancePull::removeUnnecessaryMessageQueue(const MQMessageQueue& mq) {}

//<!***************************************************************************
RebalancePush::RebalancePush(MQConsumer* consumer, MQClientFactory* pfactory) : Rebalance(consumer, pfactory) {}

bool RebalancePush::updateRequestTableInRebalance(const string& topic, vector<MQMessageQueue>& mqsSelf) {
  LOG_DEBUG("updateRequestTableInRebalance Enter");
  if (mqsSelf.empty()) {
    LOG_WARN("allocated queue is empty for topic:%s", topic.c_str());
  }

  bool changed = false;

  //<!remove
  MQ2PULLREQ requestQueueTable(getPullRequestTable());
  MQ2PULLREQ::iterator it = requestQueueTable.begin();
  for (; it != requestQueueTable.end(); ++it) {
    MQMessageQueue mqtemp = it->first;
    if (mqtemp.getTopic().compare(topic) == 0) {
      if (mqsSelf.empty() || (find(mqsSelf.begin(), mqsSelf.end(), mqtemp) == mqsSelf.end())) {
        if (!(it->second->isDroped())) {
          it->second->setDroped(true);
          // delete the lastest pull request for this mq, which hasn't been response
          // m_pClientFactory->removeDropedPullRequestOpaque(it->second);
          removeUnnecessaryMessageQueue(mqtemp);
          it->second->clearAllMsgs();  // add clear operation to avoid bad state
                                       // when dropped pullRequest returns
                                       // normal
          LOG_INFO("drop mq:%s, delete opaque:%d", mqtemp.toString().c_str(), it->second->getLatestPullRequestOpaque());
        }
        changed = true;
      }
    }
  }

  //<!add
  vector<PullRequest*> pullrequestAdd;
  DefaultMQPushConsumer* pConsumer = static_cast<DefaultMQPushConsumer*>(m_pConsumer);
  vector<MQMessageQueue>::iterator it2 = mqsSelf.begin();
  for (; it2 != mqsSelf.end(); ++it2) {
    PullRequest* pPullRequest(getPullRequest(*it2));
    if (pPullRequest && pPullRequest->isDroped()) {
      LOG_DEBUG(
          "before resume the pull handle of this pullRequest, its mq is:%s, "
          "its offset is:%lld",
          (it2->toString()).c_str(), pPullRequest->getNextOffset());
      pConsumer->getOffsetStore()->removeOffset(*it2);  // remove dirty offset which maybe update to
                                                        // OffsetStore::m_offsetTable by consuming After last
                                                        // drop
      int64 nextOffset = computePullFromWhere(*it2);
      if (nextOffset >= 0) {
        /*
          Fix issue with following scenario:
          1. pullRequest was dropped
          2. the pullMsgEvent was not executed by taskQueue, so the PullMsgEvent
          was not stop
          3. pullReuest was resumed by next doRebalance, then mulitple
          pullMsgEvent were produced for pullRequest
        */
        bool bPullMsgEvent = pPullRequest->addPullMsgEvent();
        while (!bPullMsgEvent) {
          boost::this_thread::sleep_for(boost::chrono::milliseconds(50));
          LOG_INFO("pullRequest with mq :%s has unfinished pullMsgEvent", (it2->toString()).c_str());
          bPullMsgEvent = pPullRequest->addPullMsgEvent();
        }
        pPullRequest->setDroped(false);
        pPullRequest->clearAllMsgs();  // avoid consume accumulation and consume
                                       // dumplication issues
        pPullRequest->setNextOffset(nextOffset);
        pPullRequest->updateQueueMaxOffset(nextOffset);
        LOG_INFO(
            "after resume the pull handle of this pullRequest, its mq is:%s, "
            "its offset is:%lld",
            (it2->toString()).c_str(), pPullRequest->getNextOffset());
        changed = true;
        pConsumer->producePullMsgTask(pPullRequest);
      } else {
        LOG_ERROR("get fatel error QueryOffset of mq:%s, do not reconsume this queue", (it2->toString()).c_str());
      }
    }

    if (!pPullRequest) {
      LOG_INFO("updateRequestTableInRebalance Doesn't find old mq");
      PullRequest* pullRequest = new PullRequest(m_pConsumer->getGroupName());
      pullRequest->m_messageQueue = *it2;

      int64 nextOffset = computePullFromWhere(*it2);
      if (nextOffset >= 0) {
        pullRequest->setNextOffset(nextOffset);
        pullRequest->clearAllMsgs();  // avoid consume accumulation and consume
                                      // dumplication issues
        changed = true;
        //<! mq-> pq;
        addPullRequest(*it2, pullRequest);
        pullrequestAdd.push_back(pullRequest);
        LOG_INFO("add mq:%s, request initiall offset:%lld", (*it2).toString().c_str(), nextOffset);
      }
    }
  }

  vector<PullRequest*>::iterator it3 = pullrequestAdd.begin();
  for (; it3 != pullrequestAdd.end(); ++it3) {
    LOG_DEBUG("start pull request");
    pConsumer->producePullMsgTask(*it3);
  }

  LOG_DEBUG("updateRequestTableInRebalance exit");
  return changed;
}

int64 RebalancePush::computePullFromWhere(const MQMessageQueue& mq) {
  int64 result = -1;
  DefaultMQPushConsumer* pConsumer = static_cast<DefaultMQPushConsumer*>(m_pConsumer);
  ConsumeFromWhere consumeFromWhere = pConsumer->getConsumeFromWhere();
  OffsetStore* pOffsetStore = pConsumer->getOffsetStore();
  switch (consumeFromWhere) {
    case CONSUME_FROM_LAST_OFFSET: {
      int64 lastOffset = pOffsetStore->readOffset(mq, READ_FROM_STORE, m_pConsumer->getSessionCredentials());
      if (lastOffset >= 0) {
        LOG_INFO("CONSUME_FROM_LAST_OFFSET, lastOffset of mq:%s is:%lld", mq.toString().c_str(), lastOffset);
        result = lastOffset;
      } else if (-1 == lastOffset) {
        LOG_WARN("CONSUME_FROM_LAST_OFFSET, lastOffset of mq:%s is -1", mq.toString().c_str());
        if (UtilAll::startsWith_retry(mq.getTopic())) {
          LOG_INFO("CONSUME_FROM_LAST_OFFSET, lastOffset of mq:%s is 0", mq.toString().c_str());
          result = 0;
        } else {
          try {
            result = pConsumer->maxOffset(mq);
            LOG_INFO("CONSUME_FROM_LAST_OFFSET, maxOffset of mq:%s is:%lld", mq.toString().c_str(), result);
          } catch (MQException& e) {
            LOG_ERROR("CONSUME_FROM_LAST_OFFSET error, lastOffset  of mq:%s is -1", mq.toString().c_str());
            result = -1;
          }
        }
      } else {
        LOG_ERROR("CONSUME_FROM_LAST_OFFSET error, lastOffset  of mq:%s is -1", mq.toString().c_str());
        result = -1;
      }
      break;
    }
    case CONSUME_FROM_FIRST_OFFSET: {
      int64 lastOffset = pOffsetStore->readOffset(mq, READ_FROM_STORE, m_pConsumer->getSessionCredentials());
      if (lastOffset >= 0) {
        LOG_INFO("CONSUME_FROM_FIRST_OFFSET, lastOffset of mq:%s is:%lld", mq.toString().c_str(), lastOffset);
        result = lastOffset;
      } else if (-1 == lastOffset) {
        LOG_INFO("CONSUME_FROM_FIRST_OFFSET, lastOffset of mq:%s, return 0", mq.toString().c_str());
        result = 0;
      } else {
        LOG_ERROR("CONSUME_FROM_FIRST_OFFSET, lastOffset of mq:%s, return -1", mq.toString().c_str());
        result = -1;
      }
      break;
    }
    case CONSUME_FROM_TIMESTAMP: {
      int64 lastOffset = pOffsetStore->readOffset(mq, READ_FROM_STORE, m_pConsumer->getSessionCredentials());
      if (lastOffset >= 0) {
        LOG_INFO("CONSUME_FROM_TIMESTAMP, lastOffset of mq:%s is:%lld", mq.toString().c_str(), lastOffset);
        result = lastOffset;
      } else if (-1 == lastOffset) {
        if (UtilAll::startsWith_retry(mq.getTopic())) {
          try {
            result = pConsumer->maxOffset(mq);
            LOG_INFO("CONSUME_FROM_TIMESTAMP, maxOffset  of mq:%s is:%lld", mq.toString().c_str(), result);
          } catch (MQException& e) {
            LOG_ERROR("CONSUME_FROM_TIMESTAMP error, lastOffset  of mq:%s is -1", mq.toString().c_str());
            result = -1;
          }
        } else {
          try {
          } catch (MQException& e) {
            LOG_ERROR("CONSUME_FROM_TIMESTAMP error, lastOffset  of mq:%s, return 0", mq.toString().c_str());
            result = -1;
          }
        }
      } else {
        LOG_ERROR("CONSUME_FROM_TIMESTAMP error, lastOffset  of mq:%s, return -1", mq.toString().c_str());
        result = -1;
      }
      break;
    }
    default:
      break;
  }
  return result;
}

void RebalancePush::messageQueueChanged(const string& topic,
                                        vector<MQMessageQueue>& mqAll,
                                        vector<MQMessageQueue>& mqDivided) {}

void RebalancePush::removeUnnecessaryMessageQueue(const MQMessageQueue& mq) {
  DefaultMQPushConsumer* pConsumer = static_cast<DefaultMQPushConsumer*>(m_pConsumer);
  OffsetStore* pOffsetStore = pConsumer->getOffsetStore();

  pOffsetStore->persist(mq, m_pConsumer->getSessionCredentials());
  pOffsetStore->removeOffset(mq);
  if (pConsumer->getMessageListenerType() == messageListenerOrderly) {
    unlock(mq);
  }
}

//<!************************************************************************
}  //<!end namespace;