taosdata / TDengine / #5050

coverage: 73.398% (+0.09%) from 73.313%
#5050

push

travis-ci

web-flow 
merge: from main to 3.0 branch #35319

90 of 101 new or added lines in 2 files covered. (89.11%)

489 existing lines in 125 files now uncovered.

281602 of 383662 relevant lines covered (73.4%)

138099127.08 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.17
/source/dnode/mnode/impl/src/mndSubscribe.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#define _DEFAULT_SOURCE










17



#include "mndSubscribe.h"










18



#include "mndConsumer.h"










19



#include "mndPrivilege.h"










20



#include "mndShow.h"










21



#include "mndTopic.h"










22



#include "mndTrans.h"










23



#include "mndUser.h"










24



#include "mndVgroup.h"










25



#include "tcompare.h"










26



#include "tname.h"










27












28



#define MND_SUBSCRIBE_VER_NUMBER   4










29



#define MND_SUBSCRIBE_RESERVE_SIZE 64










30












31



//#define MND_CONSUMER_LOST_HB_CNT          6










32












33



static int32_t mqRebInExecCnt = 0;










34












35



static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *);










36



static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw);










37



static int32_t  mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *);










38



static int32_t  mndSubActionDelete(SSdb *pSdb, SMqSubscribeObj *);










39



static int32_t  mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub);










40



static int32_t  mndProcessRebalanceReq(SRpcMsg *pMsg);










41



static int32_t  mndProcessDropCgroupReq(SRpcMsg *pMsg);










42



static int32_t  mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);










43



static void     mndCancelGetNextSubscribe(SMnode *pMnode, void *pIter);










44



static int32_t  mndCheckConsumer(SRpcMsg *pMsg, SHashObj *hash);










45












46



static int32_t mndSetSubCommitLogs(STrans *pTrans, SMqSubscribeObj *pSub) {




333,422✔





47



  if (pTrans == NULL || pSub == NULL) {




333,422✔





48



    return TSDB_CODE_INVALID_PARA;




×







49



  }










50



  int32_t code = 0;




333,422✔





51



  int32_t lino = 0;




333,422✔





52



  PRINT_LOG_START




333,422✔





53



  SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);




333,422✔





54



  MND_TMQ_NULL_CHECK(pCommitRaw);




333,422✔





55



  code = mndTransAppendCommitlog(pTrans, pCommitRaw);




333,422✔





56



  if (code != 0) {




333,422✔





57



    sdbFreeRaw(pCommitRaw);




×







58



    goto END;




×







59



  }










60



  code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);




333,422✔





61












62



END:




333,422✔





63



  PRINT_LOG_END




333,422✔





64



  return code;




333,422✔





65



}










66












67



int32_t mndInitSubscribe(SMnode *pMnode) {




526,944✔





68



  SSdbTable table = {




526,944✔





69



      .sdbType = SDB_SUBSCRIBE,










70



      .keyType = SDB_KEY_BINARY,










71



      .encodeFp = (SdbEncodeFp)mndSubActionEncode,










72



      .decodeFp = (SdbDecodeFp)mndSubActionDecode,










73



      .insertFp = (SdbInsertFp)mndSubActionInsert,










74



      .updateFp = (SdbUpdateFp)mndSubActionUpdate,










75



      .deleteFp = (SdbDeleteFp)mndSubActionDelete,










76



  };










77












78



  if (pMnode == NULL) {




526,944✔





79



    return TSDB_CODE_INVALID_PARA;




×







80



  }










81



  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_SUBSCRIBE_RSP, mndTransProcessRsp);




526,944✔





82



  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DELETE_SUB_RSP, mndTransProcessRsp);




526,944✔





83



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessRebalanceReq);




526,944✔





84



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP, mndProcessDropCgroupReq);




526,944✔





85



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP_RSP, mndTransProcessRsp);




526,944✔





86












87



  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_SUBSCRIPTIONS, mndRetrieveSubscribe);




526,944✔





88



  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextSubscribe);




526,944✔





89












90



  return sdbSetTable(pMnode->pSdb, table);




526,944✔





91



}










92












93



int32_t mndSchedInitSubEp(SMnode *pMnode, const SMqTopicObj *pTopic, SMqSubscribeObj *pSub) {




153,951✔





94



  int32_t code = 0;




153,951✔





95



  SSdb *  pSdb = pMnode->pSdb;




153,951✔





96



  SVgObj *pVgroup = NULL;




153,951✔





97












98



  void *pIter = NULL;




153,951✔





99



  while (1) {




866,031✔





100



    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);




1,019,982✔





101



    if (pIter == NULL) {




1,019,982✔





102



      break;




153,951✔





103



    }










104












105



    if (!mndVgroupInDb(pVgroup, pTopic->dbUid)) {




866,031✔





106



      sdbRelease(pSdb, pVgroup);




484,808✔





107



      continue;




484,808✔





108



    }










109












110



    pSub->vgNum++;




381,223✔





111












112



    int32_t vgId = pVgroup->vgId;




381,223✔





113



    if (taosArrayPush(pSub->unassignedVgs, &vgId) == NULL) {




762,446✔





114



      code = terrno;




×







115



      sdbRelease(pSdb, pVgroup);




×







116



      sdbCancelFetch(pSdb, pIter);




×







117



      goto END;




×







118



    }










119



    mInfo("init subscription %s for topic:%s assign vgId:%d", pSub->key, pTopic->name, vgId);




381,223✔





120



    sdbRelease(pSdb, pVgroup);




381,223✔





121



  }










122












123



END:




153,951✔





124



  return code;




153,951✔





125



}










126












127



static int32_t mndCreateSubscription(SMnode *pMnode, const SMqTopicObj *pTopic, const char *subKey,




153,951✔





128



                                     SMqSubscribeObj **pSub) {










129



  if (pMnode == NULL || pTopic == NULL || subKey == NULL || pSub == NULL) {




153,951✔





130



    return TSDB_CODE_INVALID_PARA;




×







131



  }










132



  int32_t lino = 0;




153,951✔





133



  int32_t code = 0;




153,951✔





134



  PRINT_LOG_START




153,951✔





135



  MND_TMQ_RETURN_CHECK(tNewSubscribeObj(subKey, pSub));




153,951✔





136



  (*pSub)->dbUid = pTopic->dbUid;




153,951✔





137



  (*pSub)->stbUid = pTopic->stbUid;




153,951✔





138



  (*pSub)->subType = pTopic->subType;




153,951✔





139



  (*pSub)->withMeta = pTopic->withMeta;




153,951✔





140












141



  MND_TMQ_RETURN_CHECK(mndSchedInitSubEp(pMnode, pTopic, *pSub));




153,951✔





142












143



END:




153,951✔





144



  PRINT_LOG_END




153,951✔





145



  return code;




153,951✔





146



}










147












148



static void mndSplitSubscribeKey(const char *key, char *topic, char *cgroup, bool fullName) {




2,035,843✔





149



  if (key == NULL || topic == NULL || cgroup == NULL) {




2,035,843✔





150



    return;




×







151



  }










152



  int32_t i = 0;




2,035,843✔





153



  while (key[i] != TMQ_SEPARATOR_CHAR) {




13,661,069✔





154



    i++;




11,625,226✔





155



  }










156



  (void)memcpy(cgroup, key, i);




2,035,843✔





157



  cgroup[i] = 0;




2,035,843✔





158



  if (fullName) {




2,035,843✔





159



    tstrncpy(topic, &key[i + 1], TSDB_TOPIC_FNAME_LEN);




1,971,192✔





160



  } else {










161



    while (key[i] != '.') {




193,953✔





162



      i++;




129,302✔





163



    }










164



    tstrncpy(topic, &key[i + 1], TSDB_CGROUP_LEN);




64,651✔





165



  }










166



}










167












168



static int32_t mndBuildSubChangeReq(SMnode *pMnode, void **pBuf, int32_t *pLen, SMqSubscribeObj *pSub,




949,444✔





169



                                    const SMqRebOutputVg *pRebVg) {










170



  if (pSub == NULL || pRebVg == NULL || pBuf == NULL || pLen == NULL) {




949,444✔





171



    return TSDB_CODE_INVALID_PARA;




×







172



  }










173



  SMqRebVgReq  req = {0};




949,444✔





174



  int32_t      code = 0;




949,444✔





175



  int32_t      lino = 0;




949,444✔





176



  SEncoder     encoder = {0};




949,444✔





177



  SMqTopicObj *pTopic = NULL;




949,444✔





178



  void *       buf = NULL;




949,444✔





179












180



  PRINT_LOG_START




949,444✔





181



  char topic[TSDB_TOPIC_FNAME_LEN] = {0};




949,444✔





182



  char cgroup[TSDB_CGROUP_LEN] = {0};




949,444✔





183



  mndSplitSubscribeKey(pSub->key, topic, cgroup, true);




949,444✔





184



  MND_TMQ_RETURN_CHECK(mndAcquireTopic(pMnode, topic, &pTopic));




949,444✔





185



  taosRLockLatch(&pTopic->lock);




949,444✔





186



  req.oldConsumerId = pRebVg->oldConsumerId;




949,444✔





187



  req.newConsumerId = pRebVg->newConsumerId;




949,444✔





188



  req.vgId = pRebVg->vgId;




949,444✔





189



  req.qmsg = pTopic->physicalPlan;




949,444✔





190



  req.schema = pTopic->schema;




949,444✔





191



  req.subType = pSub->subType;




949,444✔





192



  req.withMeta = pSub->withMeta;




949,444✔





193



  req.suid = pSub->stbUid;




949,444✔





194



  tstrncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);




949,444✔





195












196



  int32_t tlen = 0;




949,444✔





197



  tEncodeSize(tEncodeSMqRebVgReq, &req, tlen, code);




949,444✔





198



  if (code < 0) {




949,444✔





199



    goto END;




×







200



  }










201












202



  tlen += sizeof(SMsgHead);




949,444✔





203



  buf = taosMemoryMalloc(tlen);




949,444✔





204



  MND_TMQ_NULL_CHECK(buf);




949,444✔





205



  SMsgHead *pMsgHead = (SMsgHead *)buf;




949,444✔





206



  pMsgHead->contLen = htonl(tlen);




949,444✔





207



  pMsgHead->vgId = htonl(pRebVg->vgId);




949,444✔





208












209



  tEncoderInit(&encoder, POINTER_SHIFT(buf, sizeof(SMsgHead)), tlen);




949,444✔





210



  MND_TMQ_RETURN_CHECK(tEncodeSMqRebVgReq(&encoder, &req));




949,444✔





211



  *pBuf = buf;




949,444✔





212



  buf = NULL;




949,444✔





213



  *pLen = tlen;




949,444✔





214












215



END:




949,444✔





216



  PRINT_LOG_END




949,444✔





217



  taosMemoryFree(buf);




949,444✔





218



  if (pTopic != NULL) {




949,444✔





219



    taosRUnLockLatch(&pTopic->lock);




949,444✔





220



  }










221



  mndReleaseTopic(pMnode, pTopic);




949,444✔





222



  tEncoderClear(&encoder);




949,444✔





223



  return code;




949,444✔





224



}










225












226



static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub,




949,444✔





227



                                        const SMqRebOutputVg *pRebVg) {










228



  if (pMnode == NULL || pTrans == NULL || pSub == NULL || pRebVg == NULL) {




949,444✔





229



    return TSDB_CODE_INVALID_PARA;




×







230



  }










231



  int32_t code = 0;




949,444✔





232



  int32_t lino = 0;




949,444✔





233



  void *  buf = NULL;




949,444✔





234



  PRINT_LOG_START




949,444✔





235



  if (pRebVg->oldConsumerId == pRebVg->newConsumerId) {




949,444✔






UNCOV

236



    if (pRebVg->oldConsumerId == -1) return 0;  // drop stream, no consumer, while split vnode,all consumerId is -1




×







237



    code = TSDB_CODE_MND_INVALID_SUB_OPTION;




×







238



    goto END;




×







239



  }










240












241



  int32_t tlen = 0;




949,444✔





242



  MND_TMQ_RETURN_CHECK(mndBuildSubChangeReq(pMnode, &buf, &tlen, pSub, pRebVg));




949,444✔





243



  int32_t vgId = pRebVg->vgId;




949,444✔





244



  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);




949,444✔





245



  if (pVgObj == NULL) {




949,444✔





246



    code = TSDB_CODE_MND_VGROUP_NOT_EXIST;




×







247



    goto END;




×







248



  }










249












250



  STransAction action = {0};




949,444✔





251



  action.epSet = mndGetVgroupEpset(pMnode, pVgObj);




949,444✔





252



  action.pCont = buf;




949,444✔





253



  buf = NULL;




949,444✔





254



  action.contLen = tlen;




949,444✔





255



  action.msgType = TDMT_VND_TMQ_SUBSCRIBE;




949,444✔





256












257



  mndReleaseVgroup(pMnode, pVgObj);




949,444✔





258



  MND_TMQ_RETURN_CHECK(mndTransAppendRedoAction(pTrans, &action));




949,444✔





259












260



END:




949,444✔





261



  PRINT_LOG_END




949,444✔





262



  taosMemoryFree(buf);




949,444✔





263



  return code;




949,444✔





264



}










265












266



static void freeRebalanceItem(void *param) {




419,521✔





267



  if (param == NULL) return;




419,521✔





268



  SMqRebInfo *pInfo = param;




419,521✔





269



  taosArrayDestroy(pInfo->newConsumers);




419,521✔





270



  taosArrayDestroy(pInfo->removedConsumers);




419,521✔





271



}










272












273



static int32_t mndGetOrCreateRebSub(SHashObj *pHash, const char *key, SMqRebInfo **pReb) {




426,098✔





274



  if (pHash == NULL || key == NULL) {




426,098✔





275



    return TSDB_CODE_INVALID_PARA;




×







276



  }










277



  int32_t code = 0;




426,098✔





278



  int32_t lino = 0;




426,098✔





279



  PRINT_LOG_START




426,098✔





280



  SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);




426,098✔





281



  if (pRebInfo == NULL) {




426,098✔





282



    pRebInfo = tNewSMqRebSubscribe(key);




419,521✔





283



    if (pRebInfo == NULL) {




419,521✔





284



      code = terrno;




×







285



      goto END;




×







286



    }










287



    code = taosHashPut(pHash, key, strlen(key) + 1, pRebInfo, sizeof(SMqRebInfo));




419,521✔





288



    if (code != 0) {




419,521✔





289



      freeRebalanceItem(pRebInfo);




×







290



      taosMemoryFreeClear(pRebInfo);




×







291



      goto END;




×







292



    }










293



    taosMemoryFreeClear(pRebInfo);




419,521✔





294












295



    pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);




419,521✔





296



    MND_TMQ_NULL_CHECK(pRebInfo);




419,521✔





297



  }










298



  if (pReb) {




426,098✔





299



    *pReb = pRebInfo;




337,202✔





300



  }










301












302



END:




88,896✔





303



  PRINT_LOG_END




426,098✔





304



  return code;




426,098✔





305



}










306












307



static int32_t pushVgDataToHash(SArray *vgs, SHashObj *pHash, int64_t consumerId, char *key) {




1,121,002✔





308



  if (vgs == NULL || pHash == NULL || key == NULL) {




1,121,002✔





309



    return TSDB_CODE_INVALID_PARA;




×







310



  }










311



  int32_t code = 0;




1,121,002✔





312



  int32_t lino = 0;




1,121,002✔





313



  PRINT_LOG_START




1,121,002✔





314



  int32_t *pVgId = (int32_t *)taosArrayPop(vgs);




1,121,002✔





315



  MND_TMQ_NULL_CHECK(pVgId);




1,121,002✔





316



  SMqRebOutputVg outputVg = {consumerId, -1, *pVgId};




1,121,002✔





317



  MND_TMQ_RETURN_CHECK(taosHashPut(pHash, pVgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg)));




1,121,002✔





318



  mInfo("tmq rebalance sub:%s mq rebalance remove vgId:%d from consumer:0x%" PRIx64, key, *pVgId, consumerId);




1,121,002✔





319












320



END:




1,120,666✔





321



  PRINT_LOG_END




1,121,002✔





322



  return code;




1,121,002✔





323



}










324












325



static int32_t processRemovedConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, const SMqRebInputObj *pInput) {




419,521✔





326



  if (pHash == NULL || pOutput == NULL || pInput == NULL) {




419,521✔





327



    return TSDB_CODE_INVALID_PARA;




×







328



  }










329



  int32_t code = 0;




419,521✔





330



  int32_t lino = 0;




419,521✔





331



  PRINT_LOG_START




419,521✔





332



  int32_t numOfRemoved = taosArrayGetSize(pInput->pRebInfo->removedConsumers);




419,521✔





333



  int32_t actualRemoved = 0;




419,521✔





334



  for (int32_t i = 0; i < numOfRemoved; i++) {




568,959✔





335



    int64_t *consumerId = (int64_t *)taosArrayGet(pInput->pRebInfo->removedConsumers, i);




149,438✔





336



    MND_TMQ_NULL_CHECK(consumerId);




149,438✔





337



    SMqConsumerEp *pConsumerEp = taosHashGet(pOutput->pSub->consumerHash, consumerId, sizeof(int64_t));




149,438✔





338



    if (pConsumerEp == NULL) {




149,438✔





339



      continue;




×







340



    }










341












342



    int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);




149,438✔





343



    for (int32_t j = 0; j < consumerVgNum; j++) {




601,971✔





344



      MND_TMQ_RETURN_CHECK(pushVgDataToHash(pConsumerEp->vgs, pHash, *consumerId, pOutput->pSub->key));




452,533✔





345



    }










346












347



    MND_TMQ_RETURN_CHECK(taosHashRemove(pOutput->pSub->consumerHash, consumerId, sizeof(int64_t)));




149,438✔





348



    MND_TMQ_NULL_CHECK(taosArrayPush(pOutput->removedConsumers, consumerId));




298,876✔





349



    actualRemoved++;




149,438✔





350



  }










351












352



  if (numOfRemoved != actualRemoved) {




419,521✔





353



    mError("tmq rebalance sub:%s mq rebalance removedNum:%d not matched with actual:%d", pOutput->pSub->key,




×







354



           numOfRemoved, actualRemoved);










355



  } else {










356



    mInfo("tmq rebalance sub:%s removed %d consumers", pOutput->pSub->key, numOfRemoved);




419,521✔





357



  }










358



END:




×







359



  PRINT_LOG_END




419,521✔





360



  return code;




419,521✔





361



}










362












363



static int32_t processNewConsumers(SMqRebOutputObj *pOutput, const SMqRebInputObj *pInput) {




419,521✔





364



  if (pOutput == NULL || pInput == NULL) {




419,521✔





365



    return TSDB_CODE_INVALID_PARA;




×







366



  }










367



  int32_t code = 0;




419,521✔





368



  int32_t lino = 0;




419,521✔





369



  PRINT_LOG_START




419,521✔





370



  int32_t numOfNewConsumers = taosArrayGetSize(pInput->pRebInfo->newConsumers);




419,521✔





371












372



  for (int32_t i = 0; i < numOfNewConsumers; i++) {




607,285✔





373



    int64_t *consumerId = (int64_t *)taosArrayGet(pInput->pRebInfo->newConsumers, i);




187,764✔





374



    MND_TMQ_NULL_CHECK(consumerId);




187,764✔





375



    SMqConsumerEp newConsumerEp = {0};




187,764✔





376



    newConsumerEp.consumerId = *consumerId;




187,764✔





377



    newConsumerEp.vgs = taosArrayInit(0, sizeof(int32_t));




187,764✔





378



    MND_TMQ_NULL_CHECK(newConsumerEp.vgs);




187,764✔





379



    if (taosHashPut(pOutput->pSub->consumerHash, consumerId, sizeof(int64_t), &newConsumerEp, sizeof(SMqConsumerEp)) !=




187,764✔





380



        0) {










381



      freeSMqConsumerEp(&newConsumerEp);




×







382



      code = terrno;




×







383



      goto END;




×







384



    }










385



    MND_TMQ_NULL_CHECK(taosArrayPush(pOutput->newConsumers, consumerId));




375,528✔





386



    mInfo("tmq rebalance sub:%s mq rebalance add new consumer:0x%" PRIx64, pOutput->pSub->key, *consumerId);




187,764✔





387



  }










388



END:




419,521✔





389



  PRINT_LOG_END




419,521✔





390



  return code;




419,521✔





391



}










392












393



static int32_t processUnassignedVgroups(SMqRebOutputObj *pOutput, SHashObj *pHash) {




419,521✔





394



  if (pOutput == NULL || pHash == NULL) {




419,521✔





395



    return TSDB_CODE_INVALID_PARA;




×







396



  }










397



  int32_t code = 0;




419,521✔





398



  int32_t lino = 0;




419,521✔





399



  PRINT_LOG_START




419,521✔





400



  int32_t numOfVgroups = taosArrayGetSize(pOutput->pSub->unassignedVgs);




419,521✔





401



  for (int32_t i = 0; i < numOfVgroups; i++) {




1,084,110✔





402



    MND_TMQ_RETURN_CHECK(pushVgDataToHash(pOutput->pSub->unassignedVgs, pHash, -1, pOutput->pSub->key));




664,589✔





403



  }










404



END:




419,521✔





405



  PRINT_LOG_END




419,521✔





406



  return code;




419,521✔





407



}










408












409



static int32_t processModifiedConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, int32_t minVgCnt,




419,521✔





410



                                        int32_t remainderVgCnt) {










411



  if (pOutput == NULL || pHash == NULL) {




419,521✔





412



    return TSDB_CODE_INVALID_PARA;




×







413



  }










414



  int32_t code = 0;




419,521✔





415



  int32_t lino = 0;




419,521✔





416



  int32_t cnt = 0;




419,521✔





417



  void *  pIter = NULL;




419,521✔





418



  PRINT_LOG_START




419,521✔





419












420



  while (1) {




96,747✔





421



    pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);




516,268✔





422



    if (pIter == NULL) {




516,268✔





423



      break;




419,521✔





424



    }










425












426



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;




96,747✔





427



    int32_t        consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);




96,747✔





428












429



    MND_TMQ_NULL_CHECK(taosArrayPush(pOutput->modifyConsumers, &pConsumerEp->consumerId));




193,494✔





430



    if (consumerVgNum > minVgCnt) {




96,747✔





431



      if (cnt < remainderVgCnt) {




3,001✔





432



        while (taosArrayGetSize(pConsumerEp->vgs) > minVgCnt + 1) {  // pop until equal minVg + 1




1,061✔





433



          MND_TMQ_RETURN_CHECK(pushVgDataToHash(pConsumerEp->vgs, pHash, pConsumerEp->consumerId, pOutput->pSub->key));




×







434



        }










435



        cnt++;




1,061✔





436



      } else {










437



        while (taosArrayGetSize(pConsumerEp->vgs) > minVgCnt) {




5,820✔





438



          MND_TMQ_RETURN_CHECK(pushVgDataToHash(pConsumerEp->vgs, pHash, pConsumerEp->consumerId, pOutput->pSub->key));




3,880✔





439



        }










440



      }










441



    }










442



  }










443



END:




419,521✔





444



  PRINT_LOG_END




419,521✔





445



  taosHashCancelIterate(pOutput->pSub->consumerHash, pIter);




419,521✔





446



  return code;




419,521✔





447



}










448












449



static int32_t processRemoveAddVgs(SMnode *pMnode, SMqRebOutputObj *pOutput) {




419,521✔





450



  if (pMnode == NULL || pOutput == NULL) {




419,521✔





451



    return TSDB_CODE_INVALID_PARA;




×







452



  }










453



  int32_t code = 0;




419,521✔





454



  int32_t lino = 0;




419,521✔





455



  int32_t totalVgNum = 0;




419,521✔





456



  SVgObj *pVgroup = NULL;




419,521✔





457



  void *  pIter = NULL;




419,521✔





458



  void *  pIterHash = NULL;




419,521✔





459



  SArray *newVgs = taosArrayInit(0, sizeof(int32_t));




419,521✔





460



  MND_TMQ_NULL_CHECK(newVgs);




419,521✔





461



  while (1) {




2,315,436✔





462



    pIter = sdbFetch(pMnode->pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);




2,734,957✔





463



    if (pIter == NULL) {




2,734,957✔





464



      break;




419,521✔





465



    }










466



    if (pVgroup->mountVgId) {




2,315,436✔





467



      sdbRelease(pMnode->pSdb, pVgroup);




×







468



      continue;




×







469



    }










470












471



    if (!mndVgroupInDb(pVgroup, pOutput->pSub->dbUid)) {




2,315,436✔





472



      sdbRelease(pMnode->pSdb, pVgroup);




1,184,109✔





473



      continue;




1,184,109✔





474



    }










475












476



    totalVgNum++;




1,131,327✔





477



    int32_t vgId = pVgroup->vgId;




1,131,327✔





478



    MND_TMQ_NULL_CHECK(taosArrayPush(newVgs, &vgId));




1,131,327✔





479



    sdbRelease(pMnode->pSdb, pVgroup);




1,131,327✔





480



  }










481












482



  while (1) {




246,185✔





483



    pIterHash = taosHashIterate(pOutput->pSub->consumerHash, pIterHash);




665,706✔





484



    if (pIterHash == NULL) break;




665,706✔





485



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIterHash;




246,185✔





486



    int32_t        j = 0;




246,185✔





487



    while (j < taosArrayGetSize(pConsumerEp->vgs)) {




801,819✔





488



      int32_t *pVgIdTmp = taosArrayGet(pConsumerEp->vgs, j);




555,634✔





489



      MND_TMQ_NULL_CHECK(pVgIdTmp);




555,634✔





490



      bool find = false;




555,634✔





491



      for (int32_t k = 0; k < taosArrayGetSize(newVgs); k++) {




745,922✔





492



        int32_t *pnewVgId = taosArrayGet(newVgs, k);




657,026✔





493



        MND_TMQ_NULL_CHECK(pnewVgId);




657,026✔





494



        if (*pVgIdTmp == *pnewVgId) {




657,026✔





495



          taosArrayRemove(newVgs, k);




466,738✔





496



          find = true;




466,738✔





497



          break;




466,738✔





498



        }










499



      }










500



      if (!find) {




555,634✔





501



        mInfo("tmq rebalance processRemoveAddVgs old vgId:%d", *pVgIdTmp);




88,896✔





502



        taosArrayRemove(pConsumerEp->vgs, j);




88,896✔





503



        continue;




88,896✔





504



      }










505



      j++;




466,738✔





506



    }










507



  }










508












509



  if (taosArrayGetSize(pOutput->pSub->unassignedVgs) == 0 && taosArrayGetSize(newVgs) != 0) {




419,521✔





510



    MND_TMQ_NULL_CHECK(taosArrayAddAll(pOutput->pSub->unassignedVgs, newVgs));




88,896✔





511



    mInfo("tmq rebalance processRemoveAddVgs add new vg num:%d", (int)taosArrayGetSize(newVgs));




88,896✔





512



  }










513












514



END:




419,521✔





515



  sdbRelease(pMnode->pSdb, pVgroup);




419,521✔





516



  sdbCancelFetch(pMnode->pSdb, pIter);




419,521✔





517



  taosHashCancelIterate(pOutput->pSub->consumerHash, pIterHash);




419,521✔





518



  taosArrayDestroy(newVgs);




419,521✔





519



  if (code != 0) {




419,521✔





520



    mError("tmq rebalance processRemoveAddVgs failed, code:%d", code);




×







521



    return code;




×







522



  } else {










523



    mInfo("tmq rebalance processRemoveAddVgs completed, total vg num:%d", totalVgNum);




419,521✔





524



    return totalVgNum;




419,521✔





525



  }










526



}










527












528



static int32_t processSubOffsetRows(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) {




419,521✔





529



  if (pMnode == NULL || pInput == NULL || pOutput == NULL) {




419,521✔





530



    return TSDB_CODE_INVALID_PARA;




×







531



  }










532



  void *           pIter = NULL;




419,521✔





533



  SMqSubscribeObj *pSub = NULL;




419,521✔





534



  int32_t          lino = 0;




419,521✔





535



  int32_t          code = mndAcquireSubscribeByKey(pMnode, pInput->pRebInfo->key, &pSub);  // put all offset rows




419,521✔





536



  if (code != 0) {




419,521✔





537



    return 0;




153,951✔





538



  }










539



  taosRLockLatch(&pSub->lock);




265,570✔





540



  PRINT_LOG_START




265,570✔





541



  if (pOutput->pSub->offsetRows == NULL) {




265,570✔





542



    pOutput->pSub->offsetRows = taosArrayInit(4, sizeof(OffsetRows));




207,499✔





543



    MND_TMQ_NULL_CHECK(pOutput->pSub->offsetRows);




207,499✔





544



  }










545



  while (1) {




246,185✔





546



    pIter = taosHashIterate(pSub->consumerHash, pIter);




511,755✔





547



    if (pIter == NULL) break;




511,755✔





548



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;




246,185✔





549



    SMqConsumerEp *pConsumerEpNew = taosHashGet(pOutput->pSub->consumerHash, &pConsumerEp->consumerId, sizeof(int64_t));




246,185✔





550












551



    for (int j = 0; j < taosArrayGetSize(pConsumerEp->offsetRows); j++) {




787,564✔





552



      OffsetRows *d1 = taosArrayGet(pConsumerEp->offsetRows, j);




541,379✔





553



      MND_TMQ_NULL_CHECK(d1);




541,379✔





554



      bool jump = false;




541,379✔





555



      for (int i = 0; pConsumerEpNew && i < taosArrayGetSize(pConsumerEpNew->vgs); i++) {




721,863✔





556



        int32_t *pVgId = taosArrayGet(pConsumerEpNew->vgs, i);




186,581✔





557



        MND_TMQ_NULL_CHECK(pVgId);




186,581✔





558



        if (*pVgId == d1->vgId) {




186,581✔





559



          jump = true;




6,097✔





560



          mInfo("pSub->offsetRows jump, because consumer id:0x%" PRIx64 " and vgId:%d not change",




6,097✔





561



                pConsumerEp->consumerId, *pVgId);










562



          break;




6,097✔





563



        }










564



      }










565



      if (jump) continue;




541,379✔





566



      bool find = false;




535,282✔





567



      for (int i = 0; i < taosArrayGetSize(pOutput->pSub->offsetRows); i++) {




1,334,814✔





568



        OffsetRows *d2 = taosArrayGet(pOutput->pSub->offsetRows, i);




906,907✔





569



        MND_TMQ_NULL_CHECK(d2);




906,907✔





570



        if (d1->vgId == d2->vgId) {




906,907✔





571



          d2->rows += d1->rows;




107,375✔





572



          d2->offset = d1->offset;




107,375✔





573



          d2->ever = d1->ever;




107,375✔





574



          find = true;




107,375✔





575



          mInfo("pSub->offsetRows add vgId:%d, after:%" PRId64 ", before:%" PRId64, d2->vgId, d2->rows, d1->rows);




107,375✔





576



          break;




107,375✔





577



        }










578



      }










579



      if (!find) {




535,282✔





580



        MND_TMQ_NULL_CHECK(taosArrayPush(pOutput->pSub->offsetRows, d1));




855,814✔





581



      }










582



    }










583



  }










584












585



END:




265,570✔





586



  taosRUnLockLatch(&pSub->lock);




265,570✔





587



  taosHashCancelIterate(pSub->consumerHash, pIter);




265,570✔





588



  mndReleaseSubscribe(pMnode, pSub);




265,570✔





589



  PRINT_LOG_END




265,570✔





590



  return code;




265,570✔





591



}










592












593



static void printRebalanceLog(SMqRebOutputObj *pOutput) {




419,521✔





594



  if (pOutput == NULL) return;




419,521✔





595



  mInfo("sub:%s mq rebalance calculation completed, re-balanced vg", pOutput->pSub->key);




419,521✔





596



  for (int32_t i = 0; i < taosArrayGetSize(pOutput->rebVgs); i++) {




1,540,523✔





597



    SMqRebOutputVg *pOutputRebVg = taosArrayGet(pOutput->rebVgs, i);




1,121,002✔





598



    if (pOutputRebVg == NULL) continue;




1,121,002✔





599



    mInfo("sub:%s mq rebalance vgId:%d, moved from consumer:0x%" PRIx64 ", to consumer:0x%" PRIx64, pOutput->pSub->key,




1,121,002✔





600



          pOutputRebVg->vgId, pOutputRebVg->oldConsumerId, pOutputRebVg->newConsumerId);










601



  }










602












603



  void *pIter = NULL;




419,521✔





604



  while (1) {




284,511✔





605



    pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);




704,032✔





606



    if (pIter == NULL) break;




704,032✔





607



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;




284,511✔





608



    int32_t        sz = taosArrayGetSize(pConsumerEp->vgs);




284,511✔





609



    mInfo("sub:%s mq rebalance final cfg: consumer:0x%" PRIx64 " has %d vg", pOutput->pSub->key,




284,511✔





610



          pConsumerEp->consumerId, sz);










611



    for (int32_t i = 0; i < sz; i++) {




970,847✔





612



      int32_t *pVgId = taosArrayGet(pConsumerEp->vgs, i);




686,336✔





613



      if (pVgId == NULL) continue;




686,336✔





614



      mInfo("sub:%s mq rebalance final cfg: vg %d to consumer:0x%" PRIx64, pOutput->pSub->key, *pVgId,




686,336✔





615



            pConsumerEp->consumerId);










616



    }










617



  }










618



}










619












620



static void calcVgroupsCnt(const SMqRebInputObj *pInput, int32_t totalVgNum, const char *pSubKey, int32_t *minVgCnt,




419,521✔





621



                           int32_t *remainderVgCnt) {










622



  if (pInput == NULL || pSubKey == NULL || minVgCnt == NULL || remainderVgCnt == NULL) {




419,521✔





623



    return;




×







624



  }










625



  int32_t numOfRemoved = taosArrayGetSize(pInput->pRebInfo->removedConsumers);




419,521✔





626



  int32_t numOfAdded = taosArrayGetSize(pInput->pRebInfo->newConsumers);




419,521✔





627



  int32_t numOfFinal = pInput->oldConsumerNum + numOfAdded - numOfRemoved;




419,521✔





628












629



  // calc num










630



  if (numOfFinal != 0) {




419,521✔





631



    *minVgCnt = totalVgNum / numOfFinal;




277,297✔





632



    *remainderVgCnt = totalVgNum % numOfFinal;




277,297✔





633



  } else {










634



    mInfo("tmq rebalance sub:%s no consumer subscribe this topic", pSubKey);




142,224✔





635



  }










636



  mInfo(




419,521✔





637



      "tmq rebalance sub:%s mq rebalance %d vgroups, existed consumers:%d, added:%d, removed:%d, minVg:%d "










638



      "remainderVg:%d",










639



      pSubKey, totalVgNum, pInput->oldConsumerNum, numOfAdded, numOfRemoved, *minVgCnt, *remainderVgCnt);










640



}










641












642



static int32_t assignVgroups(SMqRebOutputObj *pOutput, SHashObj *pHash, int32_t minVgCnt) {




419,521✔





643



  if (pOutput == NULL || pHash == NULL) {




419,521✔





644



    return TSDB_CODE_INVALID_PARA;




×







645



  }










646



  SMqRebOutputVg *pRebVg = NULL;




419,521✔





647



  void *          pAssignIter = NULL;




419,521✔





648



  void *          pIter = NULL;




419,521✔





649



  int32_t         code = 0;




419,521✔





650



  int32_t         lino = 0;




419,521✔





651



  PRINT_LOG_START




419,521✔





652












653



  while (1) {




284,511✔





654



    pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);




704,032✔





655



    if (pIter == NULL) {




704,032✔





656



      break;




419,521✔





657



    }










658



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;




284,511✔





659



    while (taosArrayGetSize(pConsumerEp->vgs) < minVgCnt) {




959,425✔





660



      pAssignIter = taosHashIterate(pHash, pAssignIter);




674,914✔





661



      if (pAssignIter == NULL) {




674,914✔





662



        mError("tmq rebalance sub:%s assign iter is NULL, never should reach here", pOutput->pSub->key);




×







663



        break;




×







664



      }










665












666



      pRebVg = (SMqRebOutputVg *)pAssignIter;




674,914✔





667



      pRebVg->newConsumerId = pConsumerEp->consumerId;




674,914✔





668



      MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerEp->vgs, &pRebVg->vgId));




1,349,828✔





669



      mInfo("tmq rebalance mq rebalance: add vgId:%d to consumer:0x%" PRIx64 " for average", pRebVg->vgId,




674,914✔





670



            pConsumerEp->consumerId);










671



    }










672



  }










673












674



  while (1) {




1,438✔





675



    pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);




420,959✔





676



    if (pIter == NULL) {




420,959✔





677



      break;




142,224✔





678



    }










679



    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;




278,735✔





680



    if (taosArrayGetSize(pConsumerEp->vgs) == minVgCnt) {




278,735✔





681



      pAssignIter = taosHashIterate(pHash, pAssignIter);




278,394✔





682



      if (pAssignIter == NULL) {




278,394✔





683



        mInfo("tmq rebalance sub:%s assign iter is used up", pOutput->pSub->key);




277,297✔





684



        break;




277,297✔





685



      }










686












687



      pRebVg = (SMqRebOutputVg *)pAssignIter;




1,097✔





688



      pRebVg->newConsumerId = pConsumerEp->consumerId;




1,097✔





689



      MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerEp->vgs, &pRebVg->vgId));




2,194✔





690



      mInfo("tmq rebalance mq rebalance: add vgId:%d to consumer:0x%" PRIx64 " for average + 1", pRebVg->vgId,




1,097✔





691



            pConsumerEp->consumerId);










692



    }










693



  }










694












695



  if (pAssignIter != NULL) {




419,521✔





696



    mError("tmq rebalancesub:%s assign iter is not NULL, never should reach here", pOutput->pSub->key);




×







697



    code = TSDB_CODE_MND_INTERNAL_ERROR;




×







698



    goto END;




×







699



  }










700



  while (1) {




1,121,002✔





701



    pAssignIter = taosHashIterate(pHash, pAssignIter);




1,540,523✔





702



    if (pAssignIter == NULL) {




1,540,523✔





703



      break;




419,521✔





704



    }










705












706



    SMqRebOutputVg *pRebOutput = (SMqRebOutputVg *)pAssignIter;




1,121,002✔





707



    MND_TMQ_NULL_CHECK(taosArrayPush(pOutput->rebVgs, pRebOutput));




2,242,004✔





708



    if (taosHashGetSize(pOutput->pSub->consumerHash) == 0) {  // if all consumer is removed




1,121,002✔





709



      MND_TMQ_NULL_CHECK(




889,982✔





710



          taosArrayPush(pOutput->pSub->unassignedVgs, &pRebOutput->vgId));  // put all vg into unassigned










711



    }










712



  }










713












714



END:




419,521✔





715



  taosHashCancelIterate(pOutput->pSub->consumerHash, pIter);




419,521✔





716



  taosHashCancelIterate(pHash, pAssignIter);




419,521✔





717



  PRINT_LOG_END




419,521✔





718



  return code;




419,521✔





719



}










720












721



static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) {




419,521✔





722



  if (pMnode == NULL || pInput == NULL || pOutput == NULL) {




419,521✔





723



    return TSDB_CODE_INVALID_PARA;




×







724



  }










725



  int32_t totalVgNum = processRemoveAddVgs(pMnode, pOutput);




419,521✔





726



  if (totalVgNum < 0) {




419,521✔





727



    return totalVgNum;




×







728



  }










729



  const char *pSubKey = pOutput->pSub->key;




419,521✔





730



  int32_t     minVgCnt = 0;




419,521✔





731



  int32_t     remainderVgCnt = 0;




419,521✔





732



  int32_t     code = 0;




419,521✔





733



  int32_t     lino = 0;




419,521✔





734



  PRINT_LOG_START




419,521✔





735



  SHashObj *pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);




419,521✔





736



  MND_TMQ_NULL_CHECK(pHash);




419,521✔





737



  MND_TMQ_RETURN_CHECK(processRemovedConsumers(pOutput, pHash, pInput));




419,521✔





738



  MND_TMQ_RETURN_CHECK(processUnassignedVgroups(pOutput, pHash));




419,521✔





739



  calcVgroupsCnt(pInput, totalVgNum, pSubKey, &minVgCnt, &remainderVgCnt);




419,521✔





740



  MND_TMQ_RETURN_CHECK(processModifiedConsumers(pOutput, pHash, minVgCnt, remainderVgCnt));




419,521✔





741



  MND_TMQ_RETURN_CHECK(processNewConsumers(pOutput, pInput));




419,521✔





742



  MND_TMQ_RETURN_CHECK(assignVgroups(pOutput, pHash, minVgCnt));




419,521✔





743



  MND_TMQ_RETURN_CHECK(processSubOffsetRows(pMnode, pInput, pOutput));




419,521✔





744



  printRebalanceLog(pOutput);




419,521✔





745












746



END:




419,521✔





747



  taosHashCleanup(pHash);




419,521✔





748



  PRINT_LOG_END




419,521✔





749



  return code;




419,521✔





750



}










751












752



static int32_t presistConsumerByType(STrans *pTrans, SArray *consumers, int8_t type, char *cgroup, char *topic) {




992,979✔





753



  if (pTrans == NULL || consumers == NULL || cgroup == NULL) {




992,979✔





754



    return TSDB_CODE_INVALID_PARA;




×







755



  }










756



  int32_t code = 0;




992,979✔





757



  int32_t lino = 0;




992,979✔





758



  PRINT_LOG_START




992,979✔





759



  SMqConsumerObj *pConsumerNew = NULL;




992,979✔





760



  int32_t         consumerNum = taosArrayGetSize(consumers);




992,979✔





761



  for (int32_t i = 0; i < consumerNum; i++) {




1,338,400✔





762



    int64_t *consumerId = (int64_t *)taosArrayGet(consumers, i);




345,421✔





763



    MND_TMQ_NULL_CHECK(consumerId);




345,421✔





764



    MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(*consumerId, cgroup, type, topic, NULL, &pConsumerNew));




345,421✔





765



    MND_TMQ_RETURN_CHECK(mndSetConsumerCommitLogs(pTrans, pConsumerNew));




345,421✔





766



    tDeleteSMqConsumerObj(pConsumerNew);




345,421✔





767



  }










768



  pConsumerNew = NULL;




992,979✔





769












770



END:




992,979✔





771



  PRINT_LOG_END




992,979✔





772



  tDeleteSMqConsumerObj(pConsumerNew);




992,979✔





773



  return code;




992,979✔





774



}










775












776



static int32_t mndPresistConsumer(STrans *pTrans, const SMqRebOutputObj *pOutput, char *cgroup, char *topic) {




330,993✔





777



  if (pTrans == NULL || pOutput == NULL || cgroup == NULL || topic == NULL) {




330,993✔





778



    return TSDB_CODE_INVALID_PARA;




×







779



  }










780



  int32_t code = 0;




330,993✔





781



  int32_t lino = 0;




330,993✔





782



  PRINT_LOG_START




330,993✔





783



  MND_TMQ_RETURN_CHECK(presistConsumerByType(pTrans, pOutput->modifyConsumers, CONSUMER_UPDATE_REB, cgroup, NULL));




330,993✔





784



  MND_TMQ_RETURN_CHECK(presistConsumerByType(pTrans, pOutput->newConsumers, CONSUMER_ADD_REB, cgroup, topic));




330,993✔





785



  MND_TMQ_RETURN_CHECK(presistConsumerByType(pTrans, pOutput->removedConsumers, CONSUMER_REMOVE_REB, cgroup, topic));




330,993✔





786



END:




330,993✔





787



  PRINT_LOG_END




330,993✔





788



  return code;




330,993✔





789



}










790












791



static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOutputObj *pOutput) {




421,950✔





792



  if (pMnode == NULL || pMsg == NULL || pOutput == NULL) {




421,950✔





793



    return TSDB_CODE_INVALID_PARA;




×







794



  }










795



  int32_t code = 0;




421,950✔





796



  int32_t lino = 0;




421,950✔





797



  STrans *pTrans = NULL;




421,950✔





798



  PRINT_LOG_START




421,950✔





799












800



  char topic[TSDB_TOPIC_FNAME_LEN] = {0};




421,950✔





801



  char cgroup[TSDB_CGROUP_LEN] = {0};




421,950✔





802



  mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true);




421,950✔





803












804



  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pMsg, "tmq-reb");




421,950✔





805



  if (pTrans == NULL) {




421,950✔





806



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







807



    if (terrno != 0) code = terrno;




×







808



    goto END;




×







809



  }










810












811



  mndTransSetDbName(pTrans, pOutput->pSub->dbName, pOutput->pSub->key);




421,950✔





812



  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));




421,950✔





813












814



  // 1. redo action: action to all vg










815



  const SArray *rebVgs = pOutput->rebVgs;




333,422✔





816



  int32_t       vgNum = taosArrayGetSize(rebVgs);




333,422✔





817



  for (int32_t i = 0; i < vgNum; i++) {




1,282,866✔





818



    SMqRebOutputVg *pRebVg = taosArrayGet(rebVgs, i);




949,444✔





819



    MND_TMQ_NULL_CHECK(pRebVg);




949,444✔





820



    MND_TMQ_RETURN_CHECK(mndPersistSubChangeVgReq(pMnode, pTrans, pOutput->pSub, pRebVg));




949,444✔





821



  }










822












823



  // 2. commit log: subscribe and vg assignment










824



  MND_TMQ_RETURN_CHECK(mndSetSubCommitLogs(pTrans, pOutput->pSub));




333,422✔





825












826



  // 3. commit log: consumer to update status and epoch










827



  if (!pOutput->isReload) {




333,422✔





828



    MND_TMQ_RETURN_CHECK(mndPresistConsumer(pTrans, pOutput, cgroup, topic));




330,993✔





829



  }










830












831



  // 4. set cb










832



  mndTransSetCb(pTrans, TRANS_START_FUNC_MQ_REB, TRANS_STOP_FUNC_MQ_REB, NULL, 0);




333,422✔





833












834



  // 5. execution










835



  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));




333,422✔





836












837



END:




421,950✔





838



  mndTransDrop(pTrans);




421,950✔





839



  PRINT_LOG_END




421,950✔





840



  TAOS_RETURN(code);




421,950✔





841



}










842












843



// type = 0 remove  type = 1 add










844



static int32_t buildRebInfo(SHashObj *rebSubHash, SArray *topicList, int8_t type, SMqConsumerObj *pConsumer) {




656,740✔





845



  if (rebSubHash == NULL || topicList == NULL) {




656,740✔





846



    return TSDB_CODE_INVALID_PARA;




×







847



  }










848



  int32_t code = 0;




656,740✔





849



  int32_t lino = 0;




656,740✔





850



  PRINT_LOG_START




656,740✔





851



  int32_t topicNum = taosArrayGetSize(topicList);




656,740✔





852



  for (int32_t i = 0; i < topicNum; i++) {




993,942✔





853



    char *removedTopic = taosArrayGetP(topicList, i);




337,202✔





854



    MND_TMQ_NULL_CHECK(removedTopic);




337,202✔





855



    char key[TSDB_SUBSCRIBE_KEY_LEN] = {0};




337,202✔





856



    (void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", pConsumer->cgroup, TMQ_SEPARATOR, removedTopic);




337,202✔





857



    SMqRebInfo *pRebSub = NULL;




337,202✔





858



    MND_TMQ_RETURN_CHECK(mndGetOrCreateRebSub(rebSubHash, key, &pRebSub));




337,202✔





859



    if (type == 0)




337,202✔





860



      MND_TMQ_NULL_CHECK(taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId));




298,876✔





861



    else if (type == 1)




187,764✔





862



      MND_TMQ_NULL_CHECK(taosArrayPush(pRebSub->newConsumers, &pConsumer->consumerId));




375,528✔





863



  }










864












865



END:




656,740✔





866



  PRINT_LOG_END




656,740✔





867



  return code;




656,740✔





868



}










869












870



static void checkOneTopic(SMnode *pMnode, SMqConsumerObj *pConsumer, SHashObj *rebSubHash, const char *topic) {




1,165,596✔





871



  SMqSubscribeObj *pSub = NULL;




1,165,596✔





872



  char             key[TSDB_SUBSCRIBE_KEY_LEN] = {0};




1,165,596✔





873



  (void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", pConsumer->cgroup, TMQ_SEPARATOR, topic);




1,165,596✔





874



  int32_t code = 0;




1,165,596✔





875



  int32_t lino = 0;




1,165,596✔





876












877



  MND_TMQ_RETURN_CHECK(mndAcquireSubscribeByKey(pMnode, key, &pSub));




1,165,596✔





878



  taosRLockLatch(&pSub->lock);




1,165,596✔





879



  // iterate all vg assigned to the consumer of that topic










880



  SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));




1,165,596✔





881



  MND_TMQ_NULL_CHECK(pConsumerEp);




1,165,596✔





882



  int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs);




1,165,596✔





883



  for (int32_t j = 0; j < vgNum; j++) {




3,659,776✔





884



    int32_t *pVgId = taosArrayGet(pConsumerEp->vgs, j);




2,494,180✔





885



    if (pVgId == NULL) {




2,494,180✔





886



      continue;




×







887



    }










888



    SVgObj *pVgroup = mndAcquireVgroup(pMnode, *pVgId);




2,494,180✔





889



    if (!pVgroup) {




2,494,180✔





890



      code = mndGetOrCreateRebSub(rebSubHash, key, NULL);




88,896✔





891



      if (code != 0) {




88,896✔





892



        mError("failed to mndGetOrCreateRebSub vgroup:%d, error:%s", *pVgId, tstrerror(code))




×







893



      } else {










894



        mInfo("vnode splitted, vgId:%d rebalance will be triggered", *pVgId);




88,896✔





895



      }










896



    }










897



    mndReleaseVgroup(pMnode, pVgroup);




2,494,180✔





898



  }










899












900



END:




1,165,596✔





901



  if (pSub != NULL) {




1,165,596✔





902



    taosRUnLockLatch(&pSub->lock);




1,165,596✔





903



  }










904



  mndReleaseSubscribe(pMnode, pSub);




1,165,596✔





905



}




1,165,596✔





906












907



static void checkForVgroupSplit(SMnode *pMnode, SMqConsumerObj *pConsumer, SHashObj *rebSubHash) {




1,122,033✔





908



  if (pMnode == NULL || pConsumer == NULL || rebSubHash == NULL) {




1,122,033✔





909



    return;




×







910



  }










911



  int32_t newTopicNum = taosArrayGetSize(pConsumer->currentTopics);




1,122,033✔





912



  for (int32_t i = 0; i < newTopicNum; i++) {




2,287,629✔





913



    char *topic = taosArrayGetP(pConsumer->currentTopics, i);




1,165,596✔





914



    if (topic == NULL) {




1,165,596✔





915



      continue;




×







916



    }










917



    checkOneTopic(pMnode, pConsumer, rebSubHash, topic);




1,165,596✔





918



  }










919



}










920












921



static bool isOffLine(int32_t hbStatus, int32_t pollStatus, SMqConsumerObj *pConsumer) {




1,453,479✔





922



  return hbStatus * tsMqRebalanceInterval * 1000 >= pConsumer->sessionTimeoutMs ||




2,902,758✔





923



               pollStatus * tsMqRebalanceInterval * 1000 >= pConsumer->maxPollIntervalMs;




1,449,279✔





924



}










925












926



static int32_t checkOneConsumer(SMqConsumerObj *pConsumer, SMnode *pMnode, SRpcMsg *pMsg, SHashObj *rebSubHash) {




1,586,603✔





927



  int32_t code = 0;




1,586,603✔





928



  int32_t lino = 0;




1,586,603✔





929



  PRINT_LOG_START




1,586,603✔





930



  taosRLockLatch(&pConsumer->lock);




1,586,603✔





931












932



  int32_t hbStatus = atomic_add_fetch_32(&pConsumer->hbStatus, 1);




1,586,603✔





933



  int32_t pollStatus = atomic_add_fetch_32(&pConsumer->pollStatus, 1);




1,586,603✔





934



  int32_t status = atomic_load_32(&pConsumer->status);




1,586,603✔





935












936



  mDebug("tmq rebalance check for consumer:0x%" PRIx64 " status:%d(%s), sub-time:%" PRId64 ", createTime:%" PRId64




1,586,603✔





937



         ", hbstatus:%d, pollStatus:%d",










938



         pConsumer->consumerId, status, mndConsumerStatusName(status), pConsumer->subscribeTime, pConsumer->createTime,










939



         hbStatus, pollStatus);










940












941



  if (status == MQ_CONSUMER_STATUS_READY) {




1,586,603✔





942



    if (taosArrayGetSize(pConsumer->currentTopics) == 0) {  // unsubscribe or close




1,260,545✔





943



      MND_TMQ_RETURN_CHECK(




133,124✔





944



          mndSendConsumerMsg(pMnode, pConsumer->consumerId, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, &pMsg->info));










945



    } else if (isOffLine(hbStatus, pollStatus, pConsumer)) {




1,127,421✔





946



      mInfo("tmq rebalance for consumer:0x%" PRIx64 " status:%d(%s), sub-time:%" PRId64 ", createTime:%" PRId64




5,388✔





947



            ", hb lost cnt:%d, or long time no poll cnt:%d",










948



            pConsumer->consumerId, status, mndConsumerStatusName(status), pConsumer->subscribeTime,










949



            pConsumer->createTime, hbStatus, pollStatus);










950



      MND_TMQ_RETURN_CHECK(buildRebInfo(rebSubHash, pConsumer->currentTopics, 0, pConsumer));




5,388✔





951



    } else {










952



      checkForVgroupSplit(pMnode, pConsumer, rebSubHash);




1,122,033✔





953



    }










954



  } else if (status == MQ_CONSUMER_STATUS_REBALANCE && !isOffLine(hbStatus, pollStatus, pConsumer)) {




326,058✔





955



    MND_TMQ_RETURN_CHECK(buildRebInfo(rebSubHash, pConsumer->rebNewTopics, 1, pConsumer));




325,676✔





956



    MND_TMQ_RETURN_CHECK(buildRebInfo(rebSubHash, pConsumer->rebRemovedTopics, 0, pConsumer));




325,676✔





957



  } else {










958



    MND_TMQ_RETURN_CHECK(




382✔





959



        mndSendConsumerMsg(pMnode, pConsumer->consumerId, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, &pMsg->info));










960



  }










961












962



END:




382✔





963



  taosRUnLockLatch(&pConsumer->lock);




1,586,603✔





964



  PRINT_LOG_END




1,586,603✔





965



  return code;




1,586,603✔





966



}










967












968



static int32_t mndCheckConsumer(SRpcMsg *pMsg, SHashObj *rebSubHash) {




15,094,959✔





969



  if (pMsg == NULL || rebSubHash == NULL) {




15,094,959✔





970



    return TSDB_CODE_INVALID_PARA;




×







971



  }










972



  SMnode *        pMnode = pMsg->info.node;




15,094,959✔





973



  SSdb *          pSdb = pMnode->pSdb;




15,094,959✔





974



  SMqConsumerObj *pConsumer = NULL;




15,094,959✔





975



  void *          pIter = NULL;




15,094,959✔





976



  int32_t         code = 0;




15,094,959✔





977



  int32_t         lino = 0;




15,094,959✔





978



  PRINT_LOG_START




15,094,959✔





979












980



  // iterate all consumers, find all modification










981



  while (1) {










982



    pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);




16,681,562✔





983



    if (pIter == NULL) {




16,681,562✔





984



      break;




15,094,959✔





985



    }










986



    MND_TMQ_RETURN_CHECK(checkOneConsumer(pConsumer, pMnode, pMsg, rebSubHash));




1,586,603✔





987



    mndReleaseConsumer(pMnode, pConsumer);




1,586,603✔





988



  }










989



END:




15,094,959✔





990



  PRINT_LOG_END




15,094,959✔





991



  sdbCancelFetch(pSdb, pIter);




15,094,959✔





992



  mndReleaseConsumer(pMnode, pConsumer);




15,094,959✔





993



  return code;




15,094,959✔





994



}










995












996



bool mndRebTryStart() {




15,097,388✔





997



  int32_t old = atomic_val_compare_exchange_32(&mqRebInExecCnt, 0, 1);




15,097,388✔





998



  if (old > 0) mInfo("tmq rebalance counter old val:%d", old) return old == 0;




15,097,388✔





999



}










1000












1001



void mndRebCntInc() {




336,770✔





1002



  int32_t val = atomic_add_fetch_32(&mqRebInExecCnt, 1);




336,770✔





1003



  if (val > 0) mInfo("tmq rebalance cnt inc, value:%d", val)




336,770✔





1004



}




336,770✔





1005












1006



void mndRebCntDec() {




15,434,158✔





1007



  int32_t val = atomic_sub_fetch_32(&mqRebInExecCnt, 1);




15,434,158✔





1008



  if (val > 0) mInfo("tmq rebalance cnt sub, value:%d", val)




15,434,158✔





1009



}




15,434,158✔





1010












1011



static void clearRebOutput(SMqRebOutputObj *rebOutput) {




849,452✔





1012



  if (rebOutput == NULL) {




849,452✔





1013



    return;




419,521✔





1014



  }










1015



  taosArrayDestroy(rebOutput->newConsumers);




429,931✔





1016



  rebOutput->newConsumers = NULL;




429,931✔





1017



  taosArrayDestroy(rebOutput->modifyConsumers);




429,931✔





1018



  rebOutput->modifyConsumers = NULL;




429,931✔





1019



  taosArrayDestroy(rebOutput->removedConsumers);




429,931✔





1020



  rebOutput->removedConsumers = NULL;




429,931✔





1021



  taosArrayDestroy(rebOutput->rebVgs);




429,931✔





1022



  rebOutput->rebVgs = NULL;




429,931✔





1023



  tDeleteSubscribeObj(rebOutput->pSub);




429,931✔





1024



  taosMemoryFree(rebOutput->pSub);




429,931✔





1025



  rebOutput->pSub = NULL;




429,931✔





1026



}










1027












1028



static int32_t initRebOutput(SMqRebOutputObj *rebOutput) {




419,521✔





1029



  if (rebOutput == NULL) {




419,521✔





1030



    return TSDB_CODE_INVALID_PARA;




×







1031



  }










1032



  int32_t code = 0;




419,521✔





1033



  int32_t lino = 0;




419,521✔





1034



  PRINT_LOG_START




419,521✔





1035



  rebOutput->newConsumers = taosArrayInit(0, sizeof(int64_t));




419,521✔





1036



  MND_TMQ_NULL_CHECK(rebOutput->newConsumers);




419,521✔





1037



  rebOutput->removedConsumers = taosArrayInit(0, sizeof(int64_t));




419,521✔





1038



  MND_TMQ_NULL_CHECK(rebOutput->removedConsumers);




419,521✔





1039



  rebOutput->modifyConsumers = taosArrayInit(0, sizeof(int64_t));




419,521✔





1040



  MND_TMQ_NULL_CHECK(rebOutput->modifyConsumers);




419,521✔





1041



  rebOutput->rebVgs = taosArrayInit(0, sizeof(SMqRebOutputVg));




419,521✔





1042



  MND_TMQ_NULL_CHECK(rebOutput->rebVgs);




419,521✔





1043



  rebOutput = NULL;




419,521✔





1044












1045



END:




419,521✔





1046



  PRINT_LOG_END




419,521✔





1047



  clearRebOutput(rebOutput);




419,521✔





1048



  return code;




419,521✔





1049



}










1050












1051



static int32_t buildRebOutput(SMnode *pMnode, SMqRebInputObj *rebInput, SMqRebOutputObj *rebOutput) {




419,521✔





1052



  if (pMnode == NULL || rebInput == NULL || rebOutput == NULL) {




419,521✔





1053



    return TSDB_CODE_INVALID_PARA;




×







1054



  }










1055



  const char *     key = rebInput->pRebInfo->key;




419,521✔





1056



  SMqSubscribeObj *pSub = NULL;




419,521✔





1057



  SMqTopicObj *    pTopic = NULL;




419,521✔





1058



  int32_t          code = mndAcquireSubscribeByKey(pMnode, key, &pSub);




419,521✔





1059



  int32_t          lino = 0;




419,521✔





1060



  PRINT_LOG_START




419,521✔





1061












1062



  if (code != 0) {




419,521✔





1063



    // split sub key and extract topic










1064



    char topic[TSDB_TOPIC_FNAME_LEN] = {0};




153,951✔





1065



    char cgroup[TSDB_CGROUP_LEN] = {0};




153,951✔





1066



    mndSplitSubscribeKey(key, topic, cgroup, true);




153,951✔





1067



    MND_TMQ_RETURN_CHECK(mndAcquireTopic(pMnode, topic, &pTopic));




153,951✔





1068



    taosRLockLatch(&pTopic->lock);




153,951✔





1069



    rebInput->oldConsumerNum = 0;




153,951✔





1070



    MND_TMQ_RETURN_CHECK(mndCreateSubscription(pMnode, pTopic, key, &rebOutput->pSub));




153,951✔





1071



    (void)memcpy(rebOutput->pSub->dbName, pTopic->db, TSDB_DB_FNAME_LEN);




153,951✔





1072



    mInfo("tmq rebalance sub topic:%s has no consumers sub yet", key);




153,951✔





1073



  } else {










1074



    MND_TMQ_RETURN_CHECK(tCloneSubscribeObj(pSub, &rebOutput->pSub));




265,570✔





1075



    rebInput->oldConsumerNum = taosHashGetSize(rebOutput->pSub->consumerHash);




265,570✔





1076












1077



    mInfo("tmq rebalance sub topic:%s has %d consumers sub till now", key,




265,570✔





1078



          taosHashGetSize(rebOutput->pSub->consumerHash));










1079



  }










1080












1081



END:




419,185✔





1082



  PRINT_LOG_END




419,521✔





1083



  if (pTopic != NULL) {




419,521✔





1084



    taosRUnLockLatch(&pTopic->lock);




153,951✔





1085



  }










1086



  mndReleaseTopic(pMnode, pTopic);




419,521✔





1087



  mndReleaseSubscribe(pMnode, pSub);




419,521✔





1088



  return code;




419,521✔





1089



}










1090












1091



static int32_t collectVgs(SMqRebOutputObj *rebOutput, SMqSubscribeObj *pSub) {




2,429✔





1092



  int32_t code = 0;




2,429✔





1093



  int32_t lino = 0;




2,429✔





1094












1095



  void *pIterConsumer = NULL;




2,429✔





1096












1097



  PRINT_LOG_START




2,429✔





1098



  rebOutput->rebVgs = taosArrayInit(0, sizeof(SMqRebOutputVg));




2,429✔





1099



  MND_TMQ_NULL_CHECK(rebOutput->rebVgs);




2,429✔





1100












1101



  SMqConsumerEp *pConsumerEp = NULL;




2,429✔





1102












1103



  while (1) {










1104



    pIterConsumer = taosHashIterate(pSub->consumerHash, pIterConsumer);




4,858✔





1105



    if (pIterConsumer == NULL) break;




4,858✔





1106



    pConsumerEp = (SMqConsumerEp *)pIterConsumer;




2,429✔





1107












1108



    for (int32_t i = 0; i < taosArrayGetSize(pConsumerEp->vgs); i++) {




4,858✔





1109



      int32_t *pVgId = taosArrayGet(pConsumerEp->vgs, i);




2,429✔





1110



      MND_TMQ_NULL_CHECK(pVgId);




2,429✔





1111



      SMqRebOutputVg *vg = taosArrayReserve(rebOutput->rebVgs, 1);




2,429✔





1112



      MND_TMQ_NULL_CHECK(vg);




2,429✔





1113



      vg->vgId = *pVgId;




2,429✔





1114



      vg->oldConsumerId = -1;




2,429✔





1115



      vg->newConsumerId = pConsumerEp->consumerId;




2,429✔





1116



      mDebug("sub:%s reload rebalance vgId:%d remains on consumer:0x%" PRIx64, pSub->key, vg->vgId,




2,429✔





1117



             vg->newConsumerId);










1118



    }










1119



  }










1120



END:




2,429✔





1121



  PRINT_LOG_END




2,429✔





1122



  taosHashCancelIterate(pSub->consumerHash, pIterConsumer);




2,429✔





1123



  return code;




2,429✔





1124



}










1125












1126



static int32_t rebalanceOneSub(SRpcMsg *pMsg, SMqSubscribeObj *pSub) {




10,410✔





1127



  int32_t code = 0;




10,410✔





1128



  int32_t lino = 0;




10,410✔





1129












1130



  SMnode *        pMnode = pMsg->info.node;




10,410✔





1131



  SMqRebOutputObj rebOutput = {0};




10,410✔





1132












1133



  PRINT_LOG_START




10,410✔





1134



  taosRLockLatch(&pSub->lock);




10,410✔





1135












1136



  // topic and cgroup










1137



  char topic[TSDB_TOPIC_FNAME_LEN] = {0};




10,410✔





1138



  char cgroup[TSDB_CGROUP_LEN] = {0};




10,410✔





1139



  mndSplitSubscribeKey(pSub->key, topic, cgroup, true);




10,410✔





1140



  if (taosHashGet(topicsToReload, topic, strlen(topic)) == NULL) {




10,410✔





1141



    mDebug("%s topic:%s no need reload rebalance", __func__, topic);




7,981✔





1142



    goto END;




7,981✔





1143



  }










1144












1145



  rebOutput.pSub = pSub;




2,429✔





1146



  rebOutput.isReload = true;




2,429✔





1147



  MND_TMQ_RETURN_CHECK(collectVgs(&rebOutput, pSub));




2,429✔





1148



  code = mndPersistRebResult(pMnode, pMsg, &rebOutput);




2,429✔





1149



  if (code != 0) {




2,429✔





1150



    mError("%s error,msg:%s", __func__, tstrerror(code))




×







1151



  }










1152












1153



END:




10,410✔





1154



  taosRUnLockLatch(&pSub->lock);




10,410✔





1155



  rebOutput.pSub = NULL;  // avoid double free




10,410✔





1156



  clearRebOutput(&rebOutput);




10,410✔





1157



  PRINT_LOG_END




10,410✔





1158



  return code;




10,410✔





1159



}










1160












1161



static int32_t reloadRebalance(SRpcMsg *pMsg) {




2,429✔





1162



  SMnode *pMnode = pMsg->info.node;




2,429✔





1163












1164



  SSdb *           pSdb = pMnode->pSdb;




2,429✔





1165



  SMqSubscribeObj *pSub = NULL;




2,429✔





1166



  int32_t          code = 0;




2,429✔





1167



  int32_t          lino = 0;




2,429✔





1168












1169



  PRINT_LOG_START




2,429✔





1170



  mInfo("reloadRebalance start, total topics to reload:%d", taosHashGetSize(topicsToReload));




2,429✔





1171



  void *pIter = NULL;




2,429✔





1172



  while (1) {










1173



    pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub);




12,839✔





1174



    if (pIter == NULL) {




12,839✔





1175



      break;




2,429✔





1176



    }










1177












1178



    MND_TMQ_RETURN_CHECK(rebalanceOneSub(pMsg, pSub));




10,410✔





1179



    sdbRelease(pSdb, pSub);




10,410✔





1180



  }










1181



  taosHashClear(topicsToReload);




2,429✔





1182



END:




2,429✔





1183



  sdbCancelFetch(pSdb, pIter);




2,429✔





1184



  sdbRelease(pSdb, pSub);




2,429✔





1185



  PRINT_LOG_END




2,429✔





1186












1187



  return code;




2,429✔





1188



}










1189












1190



static int32_t normalRebalance(SRpcMsg *pMsg) {




15,094,959✔





1191



  int     code = 0;




15,094,959✔





1192



  int32_t lino = 0;




15,094,959✔





1193












1194



  void *  pIter = NULL;




15,094,959✔





1195



  SMnode *pMnode = pMsg->info.node;




15,094,959✔





1196












1197



  PRINT_LOG_START




15,094,959✔





1198



  SHashObj *rebSubHash = taosHashInit(64, MurmurHash3_32, true, HASH_NO_LOCK);




15,094,959✔





1199



  MND_TMQ_NULL_CHECK(rebSubHash);




15,094,959✔





1200












1201



  taosHashSetFreeFp(rebSubHash, freeRebalanceItem);




15,094,959✔





1202












1203



  MND_TMQ_RETURN_CHECK(mndCheckConsumer(pMsg, rebSubHash));




15,094,959✔





1204



  if (taosHashGetSize(rebSubHash) > 0) {




15,094,959✔





1205



    mInfo("tmq rebalance mq rebalance start, total required re-balanced trans:%d", taosHashGetSize(rebSubHash))




369,872✔





1206



  }










1207












1208



  while (1) {




419,521✔





1209



    pIter = taosHashIterate(rebSubHash, pIter);




15,514,480✔





1210



    if (pIter == NULL) {




15,514,480✔





1211



      break;




15,094,959✔





1212



    }










1213












1214



    SMqRebInputObj  rebInput = {0};




419,521✔





1215



    SMqRebOutputObj rebOutput = {0};




419,521✔





1216



    MND_TMQ_RETURN_CHECK(initRebOutput(&rebOutput));




419,521✔





1217



    rebInput.pRebInfo = (SMqRebInfo *)pIter;




419,521✔





1218



    code = buildRebOutput(pMnode, &rebInput, &rebOutput);




419,521✔





1219



    if (code != 0) {




419,521✔





1220



      mError("mq rebalance buildRebOutput, msg:%s", tstrerror(code))




×







1221



    }










1222












1223



    if (code == 0) {




419,521✔





1224



      code = mndDoRebalance(pMnode, &rebInput, &rebOutput);




419,521✔





1225



      if (code != 0) {




419,521✔





1226



        mError("mq rebalance do rebalance error, msg:%s", tstrerror(code))




×







1227



      }










1228



    }










1229












1230



    if (code == 0) {




419,521✔





1231



      code = mndPersistRebResult(pMnode, pMsg, &rebOutput);




419,521✔





1232



      if (code != 0) {




419,521✔





1233



        mError("mq rebalance persist output error, possibly vnode splitted or dropped,msg:%s", tstrerror(code))




88,528✔





1234



      }










1235



    }










1236












1237



    clearRebOutput(&rebOutput);




419,521✔





1238



  }










1239












1240



  if (taosHashGetSize(rebSubHash) > 0) {




15,094,959✔





1241



    mInfo("tmq rebalance mq rebalance completed successfully, wait trans finish")




369,872✔





1242



  }










1243












1244



END:




14,725,087✔





1245



  PRINT_LOG_END




15,094,959✔





1246



  taosHashCancelIterate(rebSubHash, pIter);




15,094,959✔





1247



  taosHashCleanup(rebSubHash);




15,094,959✔





1248



  return code;




15,094,959✔





1249



}










1250












1251



static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {




15,097,388✔





1252



  if (pMsg == NULL) {




15,097,388✔





1253



    return TSDB_CODE_INVALID_PARA;




×







1254



  }










1255



  int     code = 0;




15,097,388✔





1256



  int32_t lino = 0;




15,097,388✔





1257












1258



  void *  pIter = NULL;




15,097,388✔





1259



  SMnode *pMnode = pMsg->info.node;




15,097,388✔





1260



  PRINT_LOG_START;




15,097,388✔





1261



  if (!mndRebTryStart()) {




15,097,388✔






UNCOV

1262



    mInfo("tmq rebalance mq rebalance already in progress, do nothing");




×








UNCOV

1263



    goto END;




×







1264



  }










1265












1266



  if (taosHashGetSize(topicsToReload) > 0) {




15,097,388✔





1267



    code = reloadRebalance(pMsg);




2,429✔





1268



  } else {










1269



    code = normalRebalance(pMsg);




15,094,959✔





1270



  }










1271












1272



  mndRebCntDec();




15,097,388✔





1273












1274



END:




15,097,388✔





1275



  PRINT_LOG_END




15,097,388✔





1276



  TAOS_RETURN(code);




15,097,388✔





1277



}










1278












1279



static int32_t sendDeleteSubToVnode(SMnode *pMnode, SMqSubscribeObj *pSub, STrans *pTrans) {




93,230✔





1280



  if (pMnode == NULL || pSub == NULL || pTrans == NULL) {




93,230✔





1281



    return TSDB_CODE_INVALID_PARA;




×







1282



  }










1283



  void *  pIter = NULL;




93,230✔





1284



  SVgObj *pVgObj = NULL;




93,230✔





1285



  int32_t code = 0;




93,230✔





1286



  int32_t lino = 0;




93,230✔





1287



  PRINT_LOG_START




93,230✔





1288



  while (1) {




654,593✔





1289



    pIter = sdbFetch(pMnode->pSdb, SDB_VGROUP, pIter, (void **)&pVgObj);




747,823✔





1290



    if (pIter == NULL) {




747,823✔





1291



      break;




93,230✔





1292



    }










1293



    if (pVgObj->mountVgId) {




654,593✔





1294



      sdbRelease(pMnode->pSdb, pVgObj);




×







1295



      continue;




×







1296



    }










1297












1298



    if (!mndVgroupInDb(pVgObj, pSub->dbUid)) {




654,593✔





1299



      sdbRelease(pMnode->pSdb, pVgObj);




352,946✔





1300



      continue;




352,946✔





1301



    }










1302



    SMqVDeleteReq *pReq = taosMemoryCalloc(1, sizeof(SMqVDeleteReq));




301,647✔





1303



    MND_TMQ_NULL_CHECK(pReq);




301,647✔





1304



    pReq->head.vgId = htonl(pVgObj->vgId);




301,647✔





1305



    pReq->vgId = pVgObj->vgId;




301,647✔





1306



    pReq->consumerId = -1;




301,647✔





1307



    (void)memcpy(pReq->subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);




301,647✔





1308












1309



    STransAction action = {0};




301,647✔





1310



    action.epSet = mndGetVgroupEpset(pMnode, pVgObj);




301,647✔





1311



    action.pCont = pReq;




301,647✔





1312



    action.contLen = sizeof(SMqVDeleteReq);




301,647✔





1313



    action.msgType = TDMT_VND_TMQ_DELETE_SUB;




301,647✔





1314



    action.acceptableCode = TSDB_CODE_MND_VGROUP_NOT_EXIST;




301,647✔





1315












1316



    sdbRelease(pMnode->pSdb, pVgObj);




301,647✔





1317



    MND_TMQ_RETURN_CHECK(mndTransAppendRedoAction(pTrans, &action));




301,647✔





1318



  }










1319












1320



END:




93,230✔





1321



  PRINT_LOG_END




93,230✔





1322



  sdbRelease(pMnode->pSdb, pVgObj);




93,230✔





1323



  sdbCancelFetch(pMnode->pSdb, pIter);




93,230✔





1324



  return code;




93,230✔





1325



}










1326












1327



static int32_t checkoutOneConsumer(STrans *pTrans, SMqConsumerObj *pConsumer, bool deleteConsumer, char *topicName,




706✔





1328



                                   char *cgroup) {










1329



  int32_t         code = 0;




706✔





1330



  int32_t         lino = 0;




706✔





1331



  SMqConsumerObj *pConsumerNew = NULL;




706✔





1332












1333



  taosRLockLatch(&pConsumer->lock);




706✔





1334












1335



  if (strcmp(cgroup, pConsumer->cgroup) != 0) {




706✔





1336



    goto END;




×







1337



  }










1338












1339



  bool found1 = checkTopic(pConsumer->assignedTopics, topicName);




706✔





1340



  bool found2 = checkTopic(pConsumer->rebRemovedTopics, topicName);




706✔





1341



  bool found3 = checkTopic(pConsumer->rebNewTopics, topicName);




706✔





1342



  if (found1 || found2 || found3) {




706✔





1343



    if (deleteConsumer) {




382✔





1344



      MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup, CONSUMER_CLEAR, NULL, NULL, &pConsumerNew));




382✔





1345



      MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));




382✔





1346



      tDeleteSMqConsumerObj(pConsumerNew);




382✔





1347



      pConsumerNew = NULL;




382✔





1348



    } else {










1349



      mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s", topicName,




×







1350



             pConsumer->consumerId, pConsumer->cgroup);










1351



      code = TSDB_CODE_MND_CGROUP_USED;




×







1352



      goto END;




×







1353



    }










1354



  }










1355












1356



END:




706✔





1357



  tDeleteSMqConsumerObj(pConsumerNew);




706✔





1358



  taosRUnLockLatch(&pConsumer->lock);




706✔





1359



  return code;




706✔





1360



}










1361












1362



static int32_t mndCheckConsumerByGroup(SMnode *pMnode, STrans *pTrans, char *cgroup, char *topic, bool deleteConsumer) {




706✔





1363



  if (pMnode == NULL || pTrans == NULL || cgroup == NULL || topic == NULL) {




706✔





1364



    return TSDB_CODE_INVALID_PARA;




×







1365



  }










1366



  void *          pIter = NULL;




706✔





1367



  SMqConsumerObj *pConsumer = NULL;




706✔





1368



  int             code = 0;




706✔





1369



  int32_t         lino = 0;




706✔





1370



  PRINT_LOG_START




706✔





1371



  while (1) {










1372



    pIter = sdbFetch(pMnode->pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);




1,412✔





1373



    if (pIter == NULL) {




1,412✔





1374



      break;




706✔





1375



    }










1376



    MND_TMQ_RETURN_CHECK(checkoutOneConsumer(pTrans, pConsumer, deleteConsumer, topic, cgroup));




706✔





1377



    sdbRelease(pMnode->pSdb, pConsumer);




706✔





1378



  }










1379












1380



END:




706✔





1381



  sdbRelease(pMnode->pSdb, pConsumer);




706✔





1382



  sdbCancelFetch(pMnode->pSdb, pIter);




706✔





1383



  return code;




706✔





1384



}










1385












1386



static int32_t dropCgroup(SMqSubscribeObj *pSub, SRpcMsg *pMsg, SMDropCgroupReq *dropReq) {




706✔





1387



  int32_t code = 0;




706✔





1388



  int32_t lino = 0;




706✔





1389



  STrans *pTrans = NULL;




706✔





1390



  SMnode *pMnode = pMsg->info.node;




706✔





1391



  PRINT_LOG_START




706✔





1392



  taosRLockLatch(&pSub->lock);




706✔





1393



  if (!dropReq->force && taosHashGetSize(pSub->consumerHash) != 0) {




706✔





1394



    code = TSDB_CODE_MND_CGROUP_USED;




×







1395



    mError("cgroup:%s on topic:%s, failed to drop since %s", dropReq->cgroup, dropReq->topic, tstrerror(code));




×







1396



    goto END;




×







1397



  }










1398












1399



  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pMsg, "drop-cgroup");




706✔





1400



  MND_TMQ_NULL_CHECK(pTrans);




706✔





1401



  mInfo("trans:%d, used to drop cgroup:%s on topic %s", pTrans->id, dropReq->cgroup, dropReq->topic);




706✔





1402



  mndTransSetDbName(pTrans, pSub->dbName, NULL);




706✔





1403



  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));




706✔





1404



  MND_TMQ_RETURN_CHECK(sendDeleteSubToVnode(pMnode, pSub, pTrans));




706✔





1405



  MND_TMQ_RETURN_CHECK(mndCheckConsumerByGroup(pMnode, pTrans, dropReq->cgroup, dropReq->topic, dropReq->force));




706✔





1406



  MND_TMQ_RETURN_CHECK(mndSetDropSubCommitLogs(pMnode, pTrans, pSub));




706✔





1407



  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));




706✔





1408












1409



END:




706✔





1410



  taosRUnLockLatch(&pSub->lock);




706✔





1411



  mndTransDrop(pTrans);




706✔





1412



  PRINT_LOG_END




706✔





1413



  return code;




706✔





1414



}










1415












1416



static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) {




706✔





1417



  if (pMsg == NULL) {




706✔





1418



    return TSDB_CODE_INVALID_PARA;




×







1419



  }










1420



  SMnode *        pMnode = pMsg->info.node;




706✔





1421



  SMDropCgroupReq dropReq = {0};




706✔





1422



  int32_t         code = 0;




706✔





1423



  int32_t         lino = 0;




706✔





1424












1425



  SMqSubscribeObj *pSub = NULL;




706✔





1426












1427



  PRINT_LOG_START




706✔





1428



  MND_TMQ_RETURN_CHECK(tDeserializeSMDropCgroupReq(pMsg->pCont, pMsg->contLen, &dropReq));




706✔





1429



  char key[TSDB_SUBSCRIBE_KEY_LEN] = {0};




706✔





1430



  (void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", dropReq.cgroup, TMQ_SEPARATOR, dropReq.topic);




706✔





1431



  code = mndAcquireSubscribeByKey(pMnode, key, &pSub);




706✔





1432



  if (code != 0) {




706✔





1433



    if (dropReq.igNotExists) {




×







1434



      mInfo("cgroup:%s on topic:%s, not exist, ignore not exist is set", dropReq.cgroup, dropReq.topic);




×







1435



      mndReleaseSubscribe(pMnode, pSub);




×







1436



      return 0;




×







1437



    } else {










1438



      code = TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;




×







1439



      goto END;




×







1440



    }










1441



  }










1442



  MND_TMQ_RETURN_CHECK(dropCgroup(pSub, pMsg, &dropReq));




706✔





1443












1444



END:




706✔





1445



  mndReleaseSubscribe(pMnode, pSub);




706✔





1446



  PRINT_LOG_END




706✔





1447












1448



  if (code != 0) {




706✔





1449



    TAOS_RETURN(code);




×







1450



  }










1451



  TAOS_RETURN(TSDB_CODE_ACTION_IN_PROGRESS);




706✔





1452



}










1453












1454



void mndCleanupSubscribe(SMnode *pMnode) {}




526,880✔





1455












1456



static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *pSub) {




551,061✔





1457



  if (pSub == NULL) {




551,061✔





1458



    return NULL;




×







1459



  }










1460



  int32_t code = 0;




551,061✔





1461



  int32_t lino = 0;




551,061✔





1462



  terrno = TSDB_CODE_OUT_OF_MEMORY;




551,061✔





1463



  void *  buf = NULL;




551,061✔





1464



  int32_t tlen = tEncodeSubscribeObj(NULL, pSub);




551,061✔





1465



  if (tlen <= 0) goto SUB_ENCODE_OVER;




551,061✔





1466



  int32_t size = sizeof(int32_t) + tlen + MND_SUBSCRIBE_RESERVE_SIZE;




551,061✔





1467












1468



  SSdbRaw *pRaw = sdbAllocRaw(SDB_SUBSCRIBE, MND_SUBSCRIBE_VER_NUMBER, size);




551,061✔





1469



  if (pRaw == NULL) goto SUB_ENCODE_OVER;




551,061✔





1470












1471



  buf = taosMemoryMalloc(tlen);




551,061✔





1472



  if (buf == NULL) goto SUB_ENCODE_OVER;




551,061✔





1473












1474



  void *abuf = buf;




551,061✔





1475



  if (tEncodeSubscribeObj(&abuf, pSub) < 0) {




551,061✔





1476



    goto SUB_ENCODE_OVER;




×







1477



  }










1478












1479



  int32_t dataPos = 0;




551,061✔





1480



  SDB_SET_INT32(pRaw, dataPos, tlen, SUB_ENCODE_OVER);




551,061✔





1481



  SDB_SET_BINARY(pRaw, dataPos, buf, tlen, SUB_ENCODE_OVER);




551,061✔





1482



  SDB_SET_RESERVE(pRaw, dataPos, MND_SUBSCRIBE_RESERVE_SIZE, SUB_ENCODE_OVER);




551,061✔





1483



  SDB_SET_DATALEN(pRaw, dataPos, SUB_ENCODE_OVER);




551,061✔





1484












1485



  terrno = TSDB_CODE_SUCCESS;




551,061✔





1486












1487



SUB_ENCODE_OVER:




551,061✔





1488



  taosMemoryFreeClear(buf);




551,061✔





1489



  if (terrno != TSDB_CODE_SUCCESS) {




551,061✔





1490



    mError("subscribe:%s, failed to encode to raw:%p since %s", pSub->key, pRaw, terrstr());




×







1491



    sdbFreeRaw(pRaw);




×







1492



    return NULL;




×







1493



  }










1494












1495



  mDebug("subscribe:%s, encode to raw:%p, row:%p", pSub->key, pRaw, pSub);




551,061✔





1496



  return pRaw;




551,061✔





1497



}










1498












1499



static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw) {




443,475✔





1500



  if (pRaw == NULL) {




443,475✔





1501



    return NULL;




×







1502



  }










1503



  int32_t code = 0;




443,475✔





1504



  int32_t lino = 0;




443,475✔





1505



  terrno = TSDB_CODE_OUT_OF_MEMORY;




443,475✔





1506



  SSdbRow *        pRow = NULL;




443,475✔





1507



  SMqSubscribeObj *pSub = NULL;




443,475✔





1508



  void *           buf = NULL;




443,475✔





1509












1510



  int8_t sver = 0;




443,475✔





1511



  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto SUB_DECODE_OVER;




443,475✔





1512












1513



  if (sver > MND_SUBSCRIBE_VER_NUMBER || sver < 1) {




443,475✔





1514



    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;




×







1515



    goto SUB_DECODE_OVER;




×







1516



  }










1517












1518



  pRow = sdbAllocRow(sizeof(SMqSubscribeObj));




443,475✔





1519



  if (pRow == NULL) goto SUB_DECODE_OVER;




443,475✔





1520












1521



  pSub = sdbGetRowObj(pRow);




443,475✔





1522



  if (pSub == NULL) goto SUB_DECODE_OVER;




443,475✔





1523












1524



  int32_t dataPos = 0;




443,475✔





1525



  int32_t tlen;




443,139✔





1526



  SDB_GET_INT32(pRaw, dataPos, &tlen, SUB_DECODE_OVER);




443,475✔





1527



  buf = taosMemoryMalloc(tlen);




443,475✔





1528



  if (buf == NULL) goto SUB_DECODE_OVER;




443,475✔





1529



  SDB_GET_BINARY(pRaw, dataPos, buf, tlen, SUB_DECODE_OVER);




443,475✔





1530



  SDB_GET_RESERVE(pRaw, dataPos, MND_SUBSCRIBE_RESERVE_SIZE, SUB_DECODE_OVER);




443,475✔





1531












1532



  if (tDecodeSubscribeObj(buf, pSub, sver) == NULL) {




443,475✔





1533



    goto SUB_DECODE_OVER;




×







1534



  }










1535












1536



  terrno = TSDB_CODE_SUCCESS;




443,475✔





1537












1538



SUB_DECODE_OVER:




443,475✔





1539



  taosMemoryFreeClear(buf);




443,475✔





1540



  if (terrno != TSDB_CODE_SUCCESS) {




443,475✔





1541



    mError("subscribe:%s, failed to decode from raw:%p since %s", pSub == NULL ? "null" : pSub->key, pRaw, terrstr());




×







1542



    taosMemoryFreeClear(pRow);




×







1543



    return NULL;




×







1544



  }










1545












1546



  mDebug("subscribe:%s, decode from raw:%p, row:%p", pSub->key, pRaw, pSub);




443,475✔





1547



  return pRow;




443,475✔





1548



}










1549












1550



static int32_t mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *pSub) {




169,441✔





1551



  mDebug("subscribe:%s, perform insert action", pSub != NULL ? pSub->key : "null");




169,441✔





1552



  return 0;




169,441✔





1553



}










1554












1555



static int32_t mndSubActionDelete(SSdb *pSdb, SMqSubscribeObj *pSub) {




443,475✔





1556



  mDebug("subscribe:%s, perform delete action", pSub != NULL ? pSub->key : "null");




443,475✔





1557



  tDeleteSubscribeObj(pSub);




443,475✔





1558



  return 0;




443,475✔





1559



}










1560












1561



static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub) {




180,804✔





1562



  if (pOldSub == NULL || pNewSub == NULL) return TSDB_CODE_INVALID_PARA;




180,804✔





1563



  mDebug("subscribe:%s, perform update action", pOldSub->key);




180,804✔





1564












1565



  taosWLockLatch(&pOldSub->lock);




180,804✔





1566



  TSWAP(pOldSub->consumerHash, pNewSub->consumerHash);




180,804✔





1567



  TSWAP(pOldSub->unassignedVgs, pNewSub->unassignedVgs);




180,804✔





1568



  TSWAP(pOldSub->offsetRows, pNewSub->offsetRows);




180,804✔





1569



  taosWUnLockLatch(&pOldSub->lock);




180,804✔





1570












1571



  return 0;




180,804✔





1572



}










1573












1574



int32_t mndAcquireSubscribeByKey(SMnode *pMnode, const char *key, SMqSubscribeObj **pSub) {




18,455,377✔





1575



  if (pMnode == NULL || key == NULL || pSub == NULL) {




18,455,377✔





1576



    return TSDB_CODE_INVALID_PARA;




×







1577



  }










1578



  SSdb *pSdb = pMnode->pSdb;




18,455,377✔





1579



  *pSub = sdbAcquire(pSdb, SDB_SUBSCRIBE, key);




18,455,377✔





1580



  if (*pSub == NULL) {




18,455,377✔





1581



    return TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;




307,902✔





1582



  }










1583



  return 0;




18,147,475✔





1584



}










1585












1586



int32_t mndGetGroupNumByTopic(SMnode *pMnode, const char *topicName) {




194,536✔





1587



  if (pMnode == NULL || topicName == NULL) return 0;




194,536✔





1588



  int32_t num = 0;




194,536✔





1589



  SSdb *  pSdb = pMnode->pSdb;




194,536✔





1590












1591



  void *           pIter = NULL;




194,536✔





1592



  SMqSubscribeObj *pSub = NULL;




194,536✔





1593



  while (1) {




257,576✔





1594



    pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub);




452,112✔





1595



    if (pIter == NULL) break;




452,112✔





1596












1597



    char topic[TSDB_TOPIC_FNAME_LEN] = {0};




257,576✔





1598



    char cgroup[TSDB_CGROUP_LEN] = {0};




257,576✔





1599



    taosRLockLatch(&pSub->lock);




257,576✔





1600



    mndSplitSubscribeKey(pSub->key, topic, cgroup, true);




257,576✔





1601



    taosRUnLockLatch(&pSub->lock);




257,576✔





1602



    if (strcmp(topic, topicName) != 0) {




257,576✔





1603



      sdbRelease(pSdb, pSub);




142,030✔





1604



      continue;




142,030✔





1605



    }










1606












1607



    num++;




115,546✔





1608



    sdbRelease(pSdb, pSub);




115,546✔





1609



  }










1610












1611



  return num;




194,536✔





1612



}










1613












1614



void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub) {




18,301,426✔





1615



  if (pMnode == NULL || pSub == NULL) return;




18,301,426✔





1616



  SSdb *pSdb = pMnode->pSdb;




18,147,475✔





1617



  sdbRelease(pSdb, pSub);




18,147,475✔





1618



}










1619












1620



int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {




93,230✔





1621



  if (pMnode == NULL || pTrans == NULL || pSub == NULL) return TSDB_CODE_INVALID_PARA;




93,230✔





1622



  int32_t code = 0;




93,230✔





1623



  int32_t lino = 0;




93,230✔





1624



  PRINT_LOG_START




93,230✔





1625



  SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);




93,230✔





1626



  MND_TMQ_NULL_CHECK(pCommitRaw);




93,230✔





1627



  code = mndTransAppendCommitlog(pTrans, pCommitRaw);




93,230✔





1628



  if (code != 0) {




93,230✔





1629



    sdbFreeRaw(pCommitRaw);




×







1630



    goto END;




×







1631



  }










1632



  code = sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);




93,230✔





1633



END:




93,230✔





1634



  PRINT_LOG_END




93,230✔





1635



  return code;




93,230✔





1636



}










1637












1638



static int32_t dropOneSub(SMqSubscribeObj *pSub, SMnode *pMnode, STrans *pTrans, const char *topicName, bool force) {




177,861✔





1639



  int32_t code = 0;




177,861✔





1640



  int32_t lino = 0;




177,861✔





1641



  char    topic[TSDB_TOPIC_FNAME_LEN] = {0};




177,861✔





1642



  char    cgroup[TSDB_CGROUP_LEN] = {0};




177,861✔





1643



  taosRLockLatch(&pSub->lock);




177,861✔





1644



  mndSplitSubscribeKey(pSub->key, topic, cgroup, true);




177,861✔





1645



  if (strcmp(topic, topicName) != 0) {




177,861✔





1646



    goto END;




85,337✔





1647



  }










1648












1649



  // iter all vnode to delete handle










1650



  if (!force && taosHashGetSize(pSub->consumerHash) != 0) {




92,524✔





1651



    code = TSDB_CODE_MND_IN_REBALANCE;




×







1652



    goto END;




×







1653



  }










1654












1655



  MND_TMQ_RETURN_CHECK(sendDeleteSubToVnode(pMnode, pSub, pTrans));




92,524✔





1656



  MND_TMQ_RETURN_CHECK(mndSetDropSubCommitLogs(pMnode, pTrans, pSub));




92,524✔





1657












1658



END:




177,861✔





1659



  taosRUnLockLatch(&pSub->lock);




177,861✔





1660












1661



  return code;




177,861✔





1662



}










1663












1664



int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName, bool force) {




121,141✔





1665



  if (pMnode == NULL || pTrans == NULL || topicName == NULL) return TSDB_CODE_INVALID_PARA;




121,141✔





1666



  SSdb *           pSdb = pMnode->pSdb;




121,141✔





1667



  int32_t          code = 0;




121,141✔





1668



  int32_t          lino = 0;




121,141✔





1669



  void *           pIter = NULL;




121,141✔





1670



  SMqSubscribeObj *pSub = NULL;




121,141✔





1671



  PRINT_LOG_START




121,141✔





1672



  while (1) {










1673



    pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub);




299,002✔





1674



    if (pIter == NULL) break;




299,002✔





1675












1676



    MND_TMQ_RETURN_CHECK(dropOneSub(pSub, pMnode, pTrans, topicName, force));




177,861✔





1677



    sdbRelease(pSdb, pSub);




177,861✔





1678



  }










1679












1680



END:




121,141✔





1681



  PRINT_LOG_END




121,141✔





1682



  sdbRelease(pSdb, pSub);




121,141✔





1683



  sdbCancelFetch(pSdb, pIter);




121,141✔





1684












1685



  TAOS_RETURN(code);




121,141✔





1686



}










1687












1688



static int32_t buildResult(SSDataBlock *pBlock, int32_t *numOfRows, int64_t consumerId, const char *user,




107,194✔





1689



                           const char *fqdn, const char *topic, const char *cgroup, SArray *vgs, SArray *offsetRows) {










1690



  if (pBlock == NULL || numOfRows == NULL || topic == NULL || cgroup == NULL) {




107,194✔





1691



    return TSDB_CODE_INVALID_PARA;




×







1692



  }










1693



  int32_t code = 0;




107,194✔





1694



  int32_t lino = 0;




107,194✔





1695



  PRINT_LOG_START




107,194✔





1696



  int32_t sz = taosArrayGetSize(vgs);




107,194✔





1697



  for (int32_t j = 0; j < sz; j++) {




227,491✔





1698



    int32_t *pVgId = taosArrayGet(vgs, j);




120,297✔





1699



    MND_TMQ_NULL_CHECK(pVgId);




120,297✔





1700












1701



    SColumnInfoData *pColInfo = NULL;




120,297✔





1702



    int32_t          cols = 0;




120,297✔





1703












1704



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




120,297✔





1705



    MND_TMQ_NULL_CHECK(pColInfo);




120,297✔





1706



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, (const char *)topic, false));




120,297✔





1707












1708



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




120,297✔





1709



    MND_TMQ_NULL_CHECK(pColInfo);




120,297✔





1710



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, (const char *)cgroup, false));




120,297✔





1711












1712



    // vg id










1713



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




120,297✔





1714



    MND_TMQ_NULL_CHECK(pColInfo);




120,297✔





1715



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, (const char *)pVgId, false));




120,297✔





1716












1717



    // consumer id










1718



    char consumerIdHex[TSDB_CONSUMER_ID_LEN] = {0};




120,297✔





1719



    (void)snprintf(varDataVal(consumerIdHex), TSDB_CONSUMER_ID_LEN - VARSTR_HEADER_SIZE, "0x%" PRIx64, consumerId);




120,297✔





1720



    varDataSetLen(consumerIdHex, strlen(varDataVal(consumerIdHex)));




120,297✔





1721












1722



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




120,297✔





1723



    MND_TMQ_NULL_CHECK(pColInfo);




120,297✔





1724



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, (const char *)consumerIdHex, consumerId == -1));




120,297✔





1725












1726



    char userStr[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};




120,297✔





1727



    if (user) STR_TO_VARSTR(userStr, user);




120,297✔





1728



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




120,297✔





1729



    MND_TMQ_NULL_CHECK(pColInfo);




120,297✔





1730



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, userStr, user == NULL));




120,297✔





1731












1732



    char fqdnStr[TSDB_FQDN_LEN + VARSTR_HEADER_SIZE] = {0};




120,297✔





1733



    if (fqdn) STR_TO_VARSTR(fqdnStr, fqdn);




120,297✔





1734



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




120,297✔





1735



    MND_TMQ_NULL_CHECK(pColInfo);




120,297✔





1736



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, fqdnStr, fqdn == NULL));




120,297✔





1737












1738



    mInfo("mnd show subscriptions: topic %s, consumer:0x%" PRIx64 " cgroup %s vgid %d", varDataVal(topic), consumerId,




120,297✔





1739



          varDataVal(cgroup), *pVgId);










1740












1741



    // offset










1742



    OffsetRows *data = NULL;




120,297✔





1743



    for (int i = 0; i < taosArrayGetSize(offsetRows); i++) {




416,075✔





1744



      OffsetRows *tmp = taosArrayGet(offsetRows, i);




295,778✔





1745



      MND_TMQ_NULL_CHECK(tmp);




295,778✔





1746



      if (tmp->vgId != *pVgId) {




295,778✔





1747



        // mInfo("mnd show subscriptions: do not find vgId:%d, %d in offsetRows", tmp->vgId, *pVgId);










1748



        continue;




216,568✔





1749



      }










1750



      data = tmp;




79,210✔





1751



    }










1752



    if (data) {




120,297✔





1753



      // vg id










1754



      char buf[TSDB_OFFSET_LEN * 2 + VARSTR_HEADER_SIZE] = {0};




79,210✔





1755



      (void)tFormatOffset(varDataVal(buf), TSDB_OFFSET_LEN, &data->offset);




79,210✔





1756



      (void)snprintf(varDataVal(buf) + strlen(varDataVal(buf)),




158,420✔





1757



                     sizeof(buf) - VARSTR_HEADER_SIZE - strlen(varDataVal(buf)), "/%" PRId64, data->ever);




79,210✔





1758



      varDataSetLen(buf, strlen(varDataVal(buf)));




79,210✔





1759



      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




79,210✔





1760



      MND_TMQ_NULL_CHECK(pColInfo);




79,210✔





1761



      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, (const char *)buf, false));




79,210✔





1762



      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




79,210✔





1763



      MND_TMQ_NULL_CHECK(pColInfo);




79,210✔





1764



      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, *numOfRows, (const char *)&data->rows, false));




79,210✔





1765



    } else {










1766



      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




41,087✔





1767



      MND_TMQ_NULL_CHECK(pColInfo);




41,087✔





1768



      colDataSetNULL(pColInfo, *numOfRows);




41,087✔





1769



      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




41,087✔





1770



      MND_TMQ_NULL_CHECK(pColInfo);




41,087✔





1771



      colDataSetNULL(pColInfo, *numOfRows);




41,087✔





1772



      mInfo("mnd show subscriptions: do not find vgId:%d in offsetRows", *pVgId);




41,087✔





1773



    }










1774



    (*numOfRows)++;




120,297✔





1775



  }










1776












1777



END:




107,194✔





1778



  PRINT_LOG_END




107,194✔





1779



  return code;




107,194✔





1780



}










1781












1782



static int32_t retrieveSub(SRpcMsg *pReq, SMqSubscribeObj *pSub, SUserObj *pOperUser, bool showAll, SSDataBlock *pBlock,




64,651✔





1783



                           int32_t *numOfRows, int32_t rowsCapacity) {










1784



  int32_t        code = 0;




64,651✔





1785



  int32_t        lino = 0;




64,651✔





1786



  SMnode        *pMnode = pReq->info.node;




64,651✔





1787



  SSdb          *pSdb = pMnode->pSdb;




64,651✔





1788



  SMqConsumerEp *pConsumerEp = NULL;




64,651✔





1789



  SMqTopicObj   *pTopic = NULL;




64,651✔





1790



  void          *pIter = NULL;




64,651✔





1791



  bool           showTopic = false;




64,651✔





1792



  PRINT_LOG_START




64,651✔





1793












1794



  taosRLockLatch(&pSub->lock);




64,651✔





1795



  if (*numOfRows + pSub->vgNum > rowsCapacity) {




64,651✔





1796



    MND_TMQ_RETURN_CHECK(blockDataEnsureCapacity(pBlock, *numOfRows + pSub->vgNum));




×







1797



  }










1798












1799



  // topic and cgroup










1800



  char topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};




64,651✔





1801



  char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};




64,651✔





1802



  mndSplitSubscribeKey(pSub->key, varDataVal(topic), varDataVal(cgroup), false);




64,651✔





1803



  varDataSetLen(topic, strlen(varDataVal(topic)));




64,651✔





1804



  varDataSetLen(cgroup, strlen(varDataVal(cgroup)));




64,651✔





1805












1806



  if (!showAll) {




64,651✔





1807



    char topicFName[TSDB_TOPIC_FNAME_LEN + 1] = {0};




344✔





1808



    (void)snprintf(topicFName, sizeof(topicFName), "%d.%s", pOperUser->acctId, varDataVal(topic));




344✔





1809



    (void)mndAcquireTopic(pMnode, topicFName, &pTopic);




344✔





1810



    if (pTopic) {




344✔





1811



      SName name = {0};  // 1.topic1




344✔





1812



      if (0 == tNameFromString(&name, pTopic->name, T_NAME_ACCT | T_NAME_DB)) {




344✔





1813



        if (0 == mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_SUBSCRIPTION_SHOW, PRIV_OBJ_TOPIC, pTopic->ownerId,




344✔





1814



                                          pTopic->db, name.dbname)) {




344✔





1815



          showTopic = true;




172✔





1816



        }










1817



      }










1818



    }










1819



  }










1820












1821



  while (1) {




42,715✔





1822



    pIter = taosHashIterate(pSub->consumerHash, pIter);




107,366✔





1823



    if (pIter == NULL) break;




107,366✔





1824



    pConsumerEp = (SMqConsumerEp *)pIter;




42,715✔





1825












1826



    char           *user = NULL;




42,715✔





1827



    char           *fqdn = NULL;




42,715✔





1828



    bool            subscribeOwner = false;




42,715✔





1829



    SMqConsumerObj *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &pConsumerEp->consumerId);




42,715✔





1830



    if (pConsumer != NULL) {




42,715✔





1831



      user = pConsumer->user;




42,715✔





1832



      fqdn = pConsumer->fqdn;




42,715✔





1833



      if (strncmp(pConsumer->user, pOperUser->name, TSDB_USER_LEN) == 0) {




42,715✔





1834



        subscribeOwner = true;




42,371✔





1835



      }










1836



      sdbRelease(pSdb, pConsumer);




42,715✔





1837



    }










1838



    if (!showAll && !showTopic && !subscribeOwner) {




42,715✔





1839



      continue;




172✔





1840



    }










1841



    MND_TMQ_RETURN_CHECK(buildResult(pBlock, numOfRows, pConsumerEp->consumerId, user, fqdn, topic, cgroup,




42,543✔





1842



                                     pConsumerEp->vgs, pConsumerEp->offsetRows));










1843



  }










1844












1845



  MND_TMQ_RETURN_CHECK(




64,651✔





1846



      buildResult(pBlock, numOfRows, -1, NULL, NULL, topic, cgroup, pSub->unassignedVgs, pSub->offsetRows));










1847












1848



  pBlock->info.rows = *numOfRows;




64,651✔





1849












1850



END:




64,651✔





1851



  mndReleaseTopic(pMnode, pTopic);




64,651✔





1852



  taosRUnLockLatch(&pSub->lock);




64,651✔





1853



  taosHashCancelIterate(pSub->consumerHash, pIter);




64,651✔





1854



  PRINT_LOG_END




64,651✔





1855



  return code;




64,651✔





1856



}










1857












1858



int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {




27,550✔





1859



  if (pReq == NULL || pShow == NULL || pBlock == NULL) {




27,550✔





1860



    return TSDB_CODE_INVALID_PARA;




×







1861



  }










1862



  SMnode          *pMnode = pReq->info.node;




27,550✔





1863



  SSdb            *pSdb = pMnode->pSdb;




27,550✔





1864



  int32_t          numOfRows = 0;




27,550✔





1865



  SMqSubscribeObj *pSub = NULL;




27,550✔





1866



  SUserObj        *pOperUser = NULL;




27,550✔





1867



  int32_t          code = 0;




27,550✔





1868



  int32_t          lino = 0;




27,550✔





1869



  bool             showAll = false;




27,550✔





1870



  char             objFName[TSDB_OBJ_FNAME_LEN + 1] = {0};




27,550✔





1871












1872



  mInfo("mnd show subscriptions begin");




27,550✔





1873



  MND_TMQ_RETURN_CHECK(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser));




27,550✔





1874



  (void)snprintf(objFName, sizeof(objFName), "%d.*", pOperUser->acctId);




27,550✔





1875



  showAll = (0 == mndCheckSysObjPrivilege(pMnode, pOperUser, RPC_MSG_TOKEN(pReq), PRIV_SUBSCRIPTION_SHOW,




27,550✔





1876



                                          PRIV_OBJ_TOPIC, 0, objFName, "*"));










1877












1878



  while (numOfRows < rowsCapacity) {




92,201✔





1879



    pShow->pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pShow->pIter, (void **)&pSub);




92,201✔





1880



    if (pShow->pIter == NULL) {




92,201✔





1881



      break;




27,550✔





1882



    }










1883












1884



    MND_TMQ_RETURN_CHECK(retrieveSub(pReq, pSub, pOperUser, showAll, pBlock, &numOfRows, rowsCapacity));




64,651✔





1885












1886



    sdbRelease(pSdb, pSub);




64,651✔





1887



    pSub = NULL;




64,651✔





1888



  }










1889



  mInfo("mnd end show subscriptions");




27,550✔





1890



  pShow->numOfRows += numOfRows;




27,550✔





1891












1892



END:




27,550✔





1893



  sdbCancelFetch(pSdb, pShow->pIter);




27,550✔





1894



  sdbRelease(pSdb, pSub);




27,550✔





1895



  mndReleaseUser(pMnode, pOperUser);




27,550✔





1896












1897



  if (code != 0) {




27,550✔





1898



    mError("mnd show subscriptions failed, msg:%s", tstrerror(code));




×







1899



    TAOS_RETURN(code);




×







1900



  } else {










1901



    mDebug("mnd show subscriptions success, rows:%d", numOfRows);




27,550✔





1902



    return numOfRows;




27,550✔





1903



  }










1904



}










1905












1906



void mndCancelGetNextSubscribe(SMnode *pMnode, void *pIter) {




×







1907



  if (pMnode == NULL) {




×







1908



    return;




×







1909



  }










1910



  SSdb *pSdb = pMnode->pSdb;




×







1911



  sdbCancelFetchByType(pSdb, pIter, SDB_SUBSCRIBE);




×







1912



}
























    

  • 

    •