taosdata / TDengine / #4885

coverage: 65.258% (+4.6%) from 60.617%
#4885

push

travis-ci

web-flow 
feat(tmq): [TS-6379]remove limition for table operation in tmq  (#33834)

872 of 1074 new or added lines in 16 files covered. (81.19%)

659 existing lines in 92 files now uncovered.

177890 of 272597 relevant lines covered (65.26%)

103732965.73 hits per line

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

88.58
/source/dnode/mnode/impl/src/mndConsumer.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 "mndConsumer.h"










18



#include "mndDb.h"










19



#include "mndPrivilege.h"










20



#include "mndShow.h"










21



#include "mndSubscribe.h"










22



#include "mndTopic.h"










23



#include "mndTrans.h"










24



#include "mndVgroup.h"










25



#include "tcompare.h"










26



#include "tname.h"










27












28



#define MND_CONSUMER_VER_NUMBER   3










29



#define MND_CONSUMER_RESERVE_SIZE 64










30












31



#define MND_MAX_GROUP_PER_TOPIC 100










32












33



static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer);










34



static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer);










35



static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer);










36



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










37



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










38












39



static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg);










40



static int32_t mndProcessAskEpReq(SRpcMsg *pMsg);










41



static int32_t mndProcessMqHbReq(SRpcMsg *pMsg);










42



static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg);










43












44



int32_t mndInitConsumer(SMnode *pMnode) {




504,988✔





45



  SSdbTable table = {




504,988✔





46



      .sdbType = SDB_CONSUMER,










47



      .keyType = SDB_KEY_INT64,










48



      .encodeFp = (SdbEncodeFp)mndConsumerActionEncode,










49



      .decodeFp = (SdbDecodeFp)mndConsumerActionDecode,










50



      .insertFp = (SdbInsertFp)mndConsumerActionInsert,










51



      .updateFp = (SdbUpdateFp)mndConsumerActionUpdate,










52



      .deleteFp = (SdbDeleteFp)mndConsumerActionDelete,










53



  };










54












55



  if (pMnode == NULL){




504,988✔





56



    return TSDB_CODE_INVALID_PARA;




×







57



  }










58



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_SUBSCRIBE, mndProcessSubscribeReq);




504,988✔





59



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq);




504,988✔





60



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq);




504,988✔





61



  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, mndProcessConsumerClearMsg);




504,988✔





62












63



  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer);




504,988✔





64



  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndCancelGetNextConsumer);




504,988✔





65












66



  return sdbSetTable(pMnode->pSdb, table);




504,988✔





67



}










68












69



void mndCleanupConsumer(SMnode *pMnode) {}




504,870✔





70












71



int32_t mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo *info) {




38,143✔





72



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




38,143✔





73



    return TSDB_CODE_INVALID_PARA;




×







74



  }










75



  int32_t code = 0;




38,143✔






76



  int32_t lino = 0;




38,143✔






77



  PRINT_LOG_START




38,143✔





78



  void   *msg  = rpcMallocCont(sizeof(int64_t));




38,143✔





79



  MND_TMQ_NULL_CHECK(msg);




38,143✔





80












81



  *(int64_t*)msg = consumerId;




38,143✔





82



  SRpcMsg rpcMsg = {




38,143✔





83



      .msgType = msgType,










84



      .pCont = msg,










85



      .contLen = sizeof(int64_t),










86



      .info = *info,










87



  };










88












89



  mInfo("mndSendConsumerMsg type:%d consumer:0x%" PRIx64, msgType, consumerId);




38,143✔





90



  MND_TMQ_RETURN_CHECK(tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg));




38,143✔





91












92



END:




38,143✔






93



  PRINT_LOG_END




38,143✔






94



  return code;




38,143✔





95



}










96













97



static int32_t validateOneTopic(STrans* pTrans,char *pOneTopic, SCMSubscribeReq *subscribe, SMnode *pMnode, const char *pUser) {




54,680✔






98



  int32_t      code = 0;




54,680✔






99



  int32_t lino = 0;




54,680✔






100



  SMqTopicObj *pTopic = NULL;




54,680✔





101













102



  PRINT_LOG_START




54,680✔






103



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




54,680✔






104



  taosRLockLatch(&pTopic->lock);




54,617✔





105













106



  MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pUser, MND_OPER_SUBSCRIBE, pTopic));




54,617✔






107



  MND_TMQ_RETURN_CHECK(grantCheckExpire(TSDB_GRANT_SUBSCRIPTION));




54,489✔





108













109



  if (subscribe->enableReplay) {




54,489✔






110



    if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) {




210✔






111



      code = TSDB_CODE_TMQ_REPLAY_NOT_SUPPORT;




60✔






112



      goto END;




60✔





113



    } 











114



    if (pTopic->stbName[0] != 0) {




150✔






115



      SDbObj *pDb = mndAcquireDb(pMnode, pTopic->db);




120✔






116



      if (pDb == NULL) {




120✔






NEW


117



        code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×








NEW


118



        goto END;




×







119



      }











120



      if (pDb->cfg.numOfVgroups != 1) {




120✔






121



        mndReleaseDb(pMnode, pDb);




30✔






122



        code = TSDB_CODE_TMQ_REPLAY_NEED_ONE_VGROUP;




30✔






123



        goto END;




30✔





124



      }











125



      mndReleaseDb(pMnode, pDb);




90✔





126



    }










127



  }











128



  char  key[TSDB_CONSUMER_ID_LEN] = {0};




54,399✔






129



  (void)snprintf(key, TSDB_CONSUMER_ID_LEN, "%"PRIx64, subscribe->consumerId);




54,399✔






130



  mndTransSetDbName(pTrans, pTopic->db, key);




54,399✔






131



  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));




54,399✔





132













133



END:




54,680✔






134



  PRINT_LOG_END




54,680✔






135



  if (pTopic != NULL) {




54,680✔






136



    taosRUnLockLatch(&pTopic->lock);




54,617✔





137



  }











138



  mndReleaseTopic(pMnode, pTopic);




54,680✔





139



  return code;




54,680✔





140



}










141












142



static int32_t validateTopics(STrans* pTrans, SCMSubscribeReq *subscribe, SMnode *pMnode, const char *pUser) {




94,831✔





143



  if (pTrans == NULL || subscribe == NULL || pMnode == NULL || pUser == NULL) {




94,831✔





144



    return TSDB_CODE_INVALID_PARA;




×







145



  }










146



  int32_t      code = 0;




94,831✔






147



  int32_t lino = 0;




94,831✔





148













149



  PRINT_LOG_START




94,831✔





150



  int32_t numOfTopics = taosArrayGetSize(subscribe->topicNames);




94,831✔





151



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




149,230✔





152



    char *pOneTopic = taosArrayGetP(subscribe->topicNames, i);




54,680✔






153



    MND_TMQ_RETURN_CHECK(validateOneTopic(pTrans, pOneTopic, subscribe, pMnode, pUser));




54,680✔





154



  }










155












156



END:




94,550✔






157



  PRINT_LOG_END




94,831✔





158



  return code;




94,831✔





159



}










160












161



static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {




38,143✔





162



  if (pMsg == NULL || pMsg->pCont == NULL) {




38,143✔





163



    return TSDB_CODE_INVALID_PARA;




×







164



  }










165



  int32_t              code = 0;




38,143✔






166



  int32_t              lino = 0;




38,143✔





167



  SMnode              *pMnode = pMsg->info.node;




38,143✔





168



  SMqConsumerClearMsg *pClearMsg = pMsg->pCont;




38,143✔





169



  SMqConsumerObj      *pConsumerNew = NULL;




38,143✔





170



  STrans              *pTrans = NULL;




38,143✔





171



  SMqConsumerObj      *pConsumer = NULL;




38,143✔





172












173



  MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, pClearMsg->consumerId, &pConsumer));




38,143✔






174



  taosRLockLatch(&pConsumer->lock);




37,853✔





175



  mInfo("consumer:0x%" PRIx64 " needs to be cleared, status %s", pClearMsg->consumerId,




37,853✔





176



        mndConsumerStatusName(pConsumer->status));










177













178



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




37,853✔





179












180



  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "clear-csm");




37,853✔





181



  MND_TMQ_NULL_CHECK(pTrans);




37,853✔





182



  MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));




37,853✔





183



  code = mndTransPrepare(pMnode, pTrans);




37,853✔





184












185



END:




38,143✔






186



  PRINT_LOG_END




38,143✔






187



  if (pConsumer != NULL) {




38,143✔






188



    taosRUnLockLatch(&pConsumer->lock);




37,853✔





189



  }










190



  mndReleaseConsumer(pMnode, pConsumer);




38,143✔





191



  tDeleteSMqConsumerObj(pConsumerNew);




38,143✔





192



  mndTransDrop(pTrans);




38,143✔





193



  return code;




38,143✔





194



}










195













196



static void checkOnePrivilege(const char* topic, SMnode *pMnode, SMqHbRsp *rsp, const char *user) {




234,019✔






197



  int32_t code = 0;




234,019✔






198



  int32_t lino = 0;




234,019✔






199



  SMqTopicObj *pTopic = NULL;




234,019✔






200



  PRINT_LOG_START




234,019✔






201



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




234,019✔






202



  taosRLockLatch(&pTopic->lock);




234,019✔






203



  STopicPrivilege *data = taosArrayReserve(rsp->topicPrivileges, 1);




234,019✔






204



  MND_TMQ_NULL_CHECK(data);




234,019✔






205



  tstrncpy(data->topic, topic, TSDB_TOPIC_FNAME_LEN);




234,019✔






206



  if (mndCheckTopicPrivilege(pMnode, user, MND_OPER_SUBSCRIBE, pTopic) != 0 ||




467,654✔






207



      grantCheckExpire(TSDB_GRANT_SUBSCRIPTION) < 0) {




233,635✔






208



    data->noPrivilege = 1;




384✔





209



  } else {











210



    data->noPrivilege = 0;




233,635✔





211



  }










212













213



END:




234,019✔






214



  PRINT_LOG_END




234,019✔






215



  if (pTopic != NULL) {




234,019✔






216



    taosRUnLockLatch(&pTopic->lock);




234,019✔





217



  }











218



  mndReleaseTopic(pMnode, pTopic);




234,019✔






219



}




234,019✔





220












221



static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRsp *rsp, char *user) {




267,747✔





222



  if (pMnode == NULL || pConsumer == NULL || rsp == NULL || user == NULL) {




267,747✔





223



    return TSDB_CODE_INVALID_PARA;




×







224



  }










225



  int32_t code = 0;




267,747✔






226



  int32_t lino = 0;




267,747✔






227



  PRINT_LOG_START




267,747✔





228












229



  rsp->topicPrivileges = taosArrayInit(taosArrayGetSize(pConsumer->currentTopics), sizeof(STopicPrivilege));




267,747✔





230



  MND_TMQ_NULL_CHECK(rsp->topicPrivileges);




267,747✔





231



  for (int32_t i = 0; i < taosArrayGetSize(pConsumer->currentTopics); i++) {




501,766✔





232



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




234,019✔






233



    checkOnePrivilege(topic, pMnode, rsp, user);




234,019✔





234



  }










235












236



END:




267,747✔






237



  PRINT_LOG_END




267,747✔





238



  return code;




267,747✔





239



}










240












241



static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pConsumer){




267,747✔





242



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




267,747✔





243



    return;




×







244



  }










245



  for (int i = 0; i < taosArrayGetSize(req->topics); i++) {




501,377✔





246



    TopicOffsetRows *data = taosArrayGet(req->topics, i);




233,630✔





247



    if (data == NULL){




233,630✔





248



      continue;




×







249



    }










250



    mInfo("heartbeat report offset rows.%s:%s", pConsumer->cgroup, data->topicName);




233,630✔





251












252



    SMqSubscribeObj *pSub = NULL;




233,630✔





253



    char  key[TSDB_SUBSCRIBE_KEY_LEN] = {0};




233,630✔





254



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




233,630✔





255



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




233,630✔





256



    if (code != 0) {




233,630✔





257



      mError("failed to acquire subscribe by key:%s, code:%d", key, code);




×







258



      continue;




×







259



    }










260



    taosWLockLatch(&pSub->lock);




233,630✔





261



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




233,630✔





262



    if (pConsumerEp) {




233,630✔





263



      taosArrayDestroy(pConsumerEp->offsetRows);




228,207✔





264



      pConsumerEp->offsetRows = data->offsetRows;




228,207✔





265



      data->offsetRows = NULL;




228,207✔





266



    }










267



    taosWUnLockLatch(&pSub->lock);




233,630✔





268












269



    mndReleaseSubscribe(pMnode, pSub);




233,630✔





270



  }










271



}










272












273



static int32_t buildMqHbRsp(SRpcMsg *pMsg, SMqHbRsp *rsp){




267,747✔





274



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




267,747✔





275



    return TSDB_CODE_INVALID_PARA;




×







276



  }










277



  int32_t tlen = tSerializeSMqHbRsp(NULL, 0, rsp);




267,747✔





278



  if (tlen <= 0){




267,747✔





279



    return TSDB_CODE_TMQ_INVALID_MSG;




×







280



  }










281



  void   *buf = rpcMallocCont(tlen);




267,747✔





282



  if (buf == NULL) {




267,747✔





283



    return terrno;




×







284



  }










285












286



  if(tSerializeSMqHbRsp(buf, tlen, rsp) <= 0){




267,747✔





287



    rpcFreeCont(buf);




×







288



    return TSDB_CODE_TMQ_INVALID_MSG;




×







289



  }










290



  pMsg->info.rsp = buf;




267,747✔





291



  pMsg->info.rspLen = tlen;




267,747✔





292



  return 0;




267,747✔





293



}










294












295



static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {




270,871✔





296



  if (pMsg == NULL) {




270,871✔





297



    return TSDB_CODE_INVALID_PARA;




×







298



  }










299



  int32_t         code = 0;




270,871✔






300



  int32_t         lino = 0;




270,871✔





301



  SMnode         *pMnode = pMsg->info.node;




270,871✔





302



  SMqHbReq        req = {0};




270,871✔





303



  SMqHbRsp        rsp = {0};




270,871✔





304



  SMqConsumerObj *pConsumer = NULL;




270,871✔






305



  PRINT_LOG_START




270,871✔





306












307



  MND_TMQ_RETURN_CHECK(tDeserializeSMqHbReq(pMsg->pCont, pMsg->contLen, &req));




270,871✔





308



  int64_t consumerId = req.consumerId;




270,871✔





309



  MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));




270,871✔






310



  taosWLockLatch(&pConsumer->lock);




267,747✔





311



  MND_TMQ_RETURN_CHECK(checkPrivilege(pMnode, pConsumer, &rsp, pMsg->info.conn.user));




267,747✔





312



  atomic_store_32(&pConsumer->hbStatus, 0);




267,747✔





313



  mDebug("consumer:0x%" PRIx64 " receive hb pollFlag:%d pollStatus:%d", consumerId, req.pollFlag, pConsumer->pollStatus);




267,747✔





314



  if (req.pollFlag == 1){




267,747✔





315



    atomic_store_32(&pConsumer->pollStatus, 0);




136,769✔





316



    pConsumer->pollTime = taosGetTimestampMs();




273,538✔





317



  }










318












319



  storeOffsetRows(pMnode, &req, pConsumer);




267,747✔





320



  rsp.debugFlag = tqClientDebugFlag;




267,747✔





321



  code = buildMqHbRsp(pMsg, &rsp);




267,747✔





322












323



END:




270,871✔






324



  if (pConsumer != NULL) {




270,871✔






325



    taosWUnLockLatch(&pConsumer->lock);




267,747✔





326



  }










327



  tDestroySMqHbRsp(&rsp);




270,871✔





328



  mndReleaseConsumer(pMnode, pConsumer);




270,871✔





329



  tDestroySMqHbReq(&req);




270,871✔






330



  PRINT_LOG_END




270,871✔






331



  return code;




270,871✔





332



}










333













334



static int32_t processEachTopicEp(SMnode *pMnode, SMqConsumerObj *pConsumer, char *topic, SMqAskEpRsp *rsp, int32_t epoch) {




94,528✔






335



  int32_t         code = 0;




94,528✔






336



  int32_t         lino = 0;




94,528✔






337



  SMqSubscribeObj *pSub = NULL;




94,528✔






338



  SMqSubTopicEp topicEp = {0};




94,528✔






339



  char  key[TSDB_SUBSCRIBE_KEY_LEN] = {0};




94,528✔






340



  PRINT_LOG_START




94,528✔






341



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




94,528✔






342



  if(mndAcquireSubscribeByKey(pMnode, key, &pSub) != 0) {




94,528✔






NEW


343



    mWarn("%s failed to acquire subscribe by key:%s", __func__, key);




×








NEW


344



    goto END;




×







345



  }











346



  tstrncpy(topicEp.topic, topic, TSDB_TOPIC_FNAME_LEN);




94,528✔





347













348



  taosWLockLatch(&pSub->lock);




94,528✔





349



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











350



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




94,528✔






351



  MND_TMQ_NULL_CHECK(pConsumerEp);




94,528✔






352



  int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs);




94,528✔






353



  topicEp.vgs = taosArrayInit(vgNum, sizeof(SMqSubVgEp));




94,528✔






354



  MND_TMQ_NULL_CHECK(topicEp.vgs);




94,528✔





355













356



  tstrncpy(topicEp.db, pSub->dbName, TSDB_DB_FNAME_LEN);




94,528✔






357



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




332,967✔






358



    SMqVgEp *pVgEp = taosArrayGet(pConsumerEp->vgs, j);




238,439✔






359



    if (pVgEp == NULL) {




238,439✔






NEW


360



      continue;




×







361



    }











362



    if (epoch == -1) {




238,439✔






363



      SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId);




39,321✔






364



      if (pVgroup) {




39,321✔






365



        pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);




2,439✔






366



        mndReleaseVgroup(pMnode, pVgroup);




2,439✔





367



      }










368



    }











369



    SMqSubVgEp vgEp = {.epSet = pVgEp->epSet, .vgId = pVgEp->vgId, .offset = -1};




238,439✔






370



    MND_TMQ_NULL_CHECK(taosArrayPush(topicEp.vgs, &vgEp));




476,878✔





371



  }











372



  MND_TMQ_NULL_CHECK(taosArrayPush(rsp->topics, &topicEp));




189,056✔






373



  topicEp.vgs = NULL;




94,528✔





374













375



END:




94,528✔






376



  if (pSub != NULL) {




94,528✔






377



    taosWUnLockLatch(&pSub->lock);




94,528✔





378



  }











379



  taosArrayDestroy(topicEp.vgs);




94,528✔






380



  mndReleaseSubscribe(pMnode, pSub);




94,528✔






381



  PRINT_LOG_END




94,528✔





382



  return code;




94,528✔





383



}










384












385



static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t epoch, SMqAskEpRsp *rsp){




134,849✔





386



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




134,849✔





387



    return TSDB_CODE_INVALID_PARA;




×







388



  }











389



  int32_t code = 0;




134,849✔






390



  int32_t lino = 0;




134,849✔






391



  PRINT_LOG_START




134,849✔





392












393



  int32_t numOfTopics = taosArrayGetSize(pConsumer->currentTopics);




134,849✔





394



  rsp->topics = taosArrayInit(numOfTopics, sizeof(SMqSubTopicEp));




134,849✔






395



  MND_TMQ_NULL_CHECK(rsp->topics);




134,849✔





396












397



  // handle all topics subscribed by this consumer










398



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




229,377✔





399



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




94,528✔






400



    MND_TMQ_RETURN_CHECK(processEachTopicEp(pMnode, pConsumer, topic, rsp, epoch));




94,528✔





401



  }










402













403



END:




134,849✔






404



  PRINT_LOG_END




134,849✔






405



  return code;




134,849✔





406



}










407












408



static int32_t buildAskEpRsp(SRpcMsg *pMsg, SMqAskEpRsp *rsp, int32_t serverEpoch, int64_t consumerId){




636,172✔





409



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




636,172✔





410



    return TSDB_CODE_INVALID_PARA;




×







411



  }










412



  int32_t code = 0;




636,278✔





413



  // encode rsp










414



  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqAskEpRsp(NULL, rsp);




636,172✔





415



  void   *buf = rpcMallocCont(tlen);




636,172✔





416



  if (buf == NULL) {




636,278✔





417



    return terrno;




×







418



  }










419












420



  SMqRspHead *pHead = buf;




636,278✔





421












422



  pHead->mqMsgType = TMQ_MSG_TYPE__EP_RSP;




636,278✔





423



  pHead->epoch = serverEpoch;




636,278✔





424



  pHead->consumerId = consumerId;




636,278✔





425



  pHead->walsver = 0;




636,278✔





426



  pHead->walever = 0;




636,278✔





427












428



  void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));




636,278✔





429



  if (tEncodeSMqAskEpRsp(&abuf, rsp) < 0) {




636,278✔





430



    rpcFreeCont(buf);




×







431



    return TSDB_CODE_TSC_INTERNAL_ERROR;




×







432



  }










433












434



  // send rsp










435



  pMsg->info.rsp = buf;




636,278✔





436



  pMsg->info.rspLen = tlen;




636,278✔





437



  return code;




636,384✔





438



}










439












440



static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {




933,305✔





441



  if (pMsg == NULL) {




933,305✔





442



    return TSDB_CODE_INVALID_PARA;




×







443



  }










444



  SMnode     *pMnode = pMsg->info.node;




933,305✔





445



  SMqAskEpReq req = {0};




933,517✔





446



  SMqAskEpRsp rsp = {0};




933,517✔





447



  int32_t     code = 0;




933,517✔






448



  int32_t     lino = 0;




933,517✔





449



  SMqConsumerObj *pConsumer = NULL;




933,517✔





450



  PRINT_LOG_START




933,517✔





451












452



  MND_TMQ_RETURN_CHECK(tDeserializeSMqAskEpReq(pMsg->pCont, pMsg->contLen, &req));




933,623✔





453



  int64_t consumerId = req.consumerId;




933,623✔





454



  MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));




933,623✔






455



  taosRLockLatch(&pConsumer->lock);




930,198✔





456



  if (strncmp(req.cgroup, pConsumer->cgroup, tListLen(pConsumer->cgroup)) != 0) {




930,304✔





457



    mError("consumer:0x%" PRIx64 " group:%s not consistent with data in sdb, saved cgroup:%s", consumerId, req.cgroup,




×







458



           pConsumer->cgroup);










459



    code = TSDB_CODE_MND_CONSUMER_NOT_EXIST;




×







460



    goto END;




×







461



  }










462












463



  // 1. check consumer status










464



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




930,092✔





465



  if (status != MQ_CONSUMER_STATUS_READY) {




930,198✔





466



    mInfo("consumer:0x%" PRIx64 " not ready, status: %s", consumerId, mndConsumerStatusName(status));




293,920✔





467



    code = TSDB_CODE_MND_CONSUMER_NOT_READY;




293,920✔





468



    goto END;




293,920✔





469



  }










470












471



  int32_t epoch = req.epoch;




636,278✔





472



  int32_t serverEpoch = atomic_load_32(&pConsumer->epoch);




636,278✔





473












474



  // 2. check epoch, only send ep info when epochs do not match










475



  if (epoch != serverEpoch) {




636,278✔





476



    mInfo("process ask ep, consumer:0x%" PRIx64 "(epoch %d) update with server epoch %d",




134,849✔





477



          consumerId, epoch, serverEpoch);










478



    MND_TMQ_RETURN_CHECK(addEpSetInfo(pMnode, pConsumer, epoch, &rsp));




134,849✔





479



  }










480












481



  code = buildAskEpRsp(pMsg, &rsp, serverEpoch, consumerId);




636,278✔





482












483



END:




933,623✔






484



  if (pConsumer != NULL) {




933,623✔






485



    taosRUnLockLatch(&pConsumer->lock);




930,304✔





486



  }










487



  tDeleteSMqAskEpRsp(&rsp);










488



  mndReleaseConsumer(pMnode, pConsumer);




933,517✔






489



  PRINT_LOG_END




933,623✔





490



  return code;




933,623✔





491



}










492












493



int32_t mndSetConsumerDropLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {




38,373✔





494



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




38,373✔





495



    return TSDB_CODE_INVALID_PARA;




×







496



  }










497



  int32_t  code = 0;




38,373✔






498



  int32_t lino = 0;




38,373✔






499



  PRINT_LOG_START




38,373✔





500



  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);




38,373✔





501



  MND_TMQ_NULL_CHECK(pCommitRaw);




38,373✔





502



  code = mndTransAppendCommitlog(pTrans, pCommitRaw);




38,373✔





503



  if (code != 0) {




38,373✔





504



    sdbFreeRaw(pCommitRaw);




×







505



    goto END;




×







506



  }










507



  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));




38,373✔





508












509



END:




38,373✔






510



  PRINT_LOG_END




38,373✔





511



  return code;




38,373✔





512



}










513












514



int32_t mndSetConsumerCommitLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {




192,892✔





515



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




192,892✔





516



    return TSDB_CODE_INVALID_PARA;




×







517



  }










518



  int32_t  code = 0;




192,892✔






519



  int32_t lino = 0;




192,892✔






520



  PRINT_LOG_START




192,892✔





521



  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);




192,892✔





522



  MND_TMQ_NULL_CHECK(pCommitRaw);




192,892✔





523



  code = mndTransAppendCommitlog(pTrans, pCommitRaw);




192,892✔





524



  if (code != 0) {




192,892✔





525



    sdbFreeRaw(pCommitRaw);




×







526



    goto END;




×







527



  }










528



  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));




192,892✔





529



END:




192,892✔





530












531



  return code;




192,892✔





532



}










533












534



static void freeItem(void *param) {




62✔





535



  if (param == NULL) {




62✔





536



    return;




×







537



  }










538



  void *pItem = *(void **)param;




62✔





539



  if (pItem != NULL) {




62✔





540



    taosMemoryFree(pItem);




62✔





541



  }










542



}










543












544



#define ADD_TOPIC_TO_ARRAY(element, array) \










545



char *newTopicCopy = taosStrdup(element); \










546



MND_TMQ_NULL_CHECK(newTopicCopy);\










547



if (taosArrayPush(pConsumerNew->array, &newTopicCopy) == NULL){\










548



  taosMemoryFree(newTopicCopy);\










549



  code = terrno;\










550



  goto END;\










551



}










552












553



static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerObj *pConsumerNew){




41,317✔





554



  if (pExistedConsumer == NULL || pConsumerNew == NULL) {




41,317✔





555



    return TSDB_CODE_INVALID_PARA;




×







556



  }










557



  int32_t code = 0;




41,317✔






558



  int32_t lino = 0;




41,317✔






559



  PRINT_LOG_START




41,317✔






560



  taosRLockLatch(&pExistedConsumer->lock);




41,317✔





561












562



  pConsumerNew->rebNewTopics = taosArrayInit(0, sizeof(void *));




41,317✔





563



  MND_TMQ_NULL_CHECK(pConsumerNew->rebNewTopics);




41,317✔





564



  pConsumerNew->rebRemovedTopics = taosArrayInit(0, sizeof(void *));




41,317✔





565



  MND_TMQ_NULL_CHECK(pConsumerNew->rebRemovedTopics);




41,317✔





566












567



  int32_t newTopicNum = taosArrayGetSize(pConsumerNew->assignedTopics);




41,317✔





568



  int32_t oldTopicNum = taosArrayGetSize(pExistedConsumer->currentTopics);




41,317✔





569



  int32_t i = 0, j = 0;




41,317✔





570



  while (i < oldTopicNum || j < newTopicNum) {




83,166✔





571



    if (i >= oldTopicNum) {




41,849✔





572



      void* tmp = taosArrayGetP(pConsumerNew->assignedTopics, j);




285✔





573



      MND_TMQ_NULL_CHECK(tmp);




285✔





574



      ADD_TOPIC_TO_ARRAY(tmp, rebNewTopics);




570✔





575



      j++;




285✔





576



      continue;




285✔





577



    } else if (j >= newTopicNum) {




41,564✔





578



      void* tmp = taosArrayGetP(pExistedConsumer->currentTopics, i);




41,215✔





579



      MND_TMQ_NULL_CHECK(tmp);




41,215✔





580



      ADD_TOPIC_TO_ARRAY(tmp, rebRemovedTopics);




82,430✔





581



      i++;




41,215✔





582



      continue;




41,215✔





583



    } else {










584



      char *oldTopic = taosArrayGetP(pExistedConsumer->currentTopics, i);




349✔





585



      MND_TMQ_NULL_CHECK(oldTopic);




349✔





586



      char *newTopic = taosArrayGetP(pConsumerNew->assignedTopics, j);




349✔





587



      MND_TMQ_NULL_CHECK(newTopic);




349✔





588



      int   comp = strcmp(oldTopic, newTopic);




349✔





589



      if (comp == 0) {




349✔





590



        i++;




349✔





591



        j++;




349✔





592



        continue;




349✔





593



      } else if (comp < 0) {




×







594



        ADD_TOPIC_TO_ARRAY(oldTopic, rebRemovedTopics);




×







595



        i++;




×







596



        continue;




×







597



      } else {




×







598



        ADD_TOPIC_TO_ARRAY(newTopic, rebNewTopics);




×







599



        j++;




×







600



        continue;




×







601



      }










602



    }










603



  }










604



  // no topics need to be rebalanced










605



  if (taosArrayGetSize(pConsumerNew->rebNewTopics) == 0 && taosArrayGetSize(pConsumerNew->rebRemovedTopics) == 0) {




41,317✔





606



    code = TSDB_CODE_TMQ_NO_NEED_REBALANCE;




349✔





607



  }










608












609



END:




40,968✔






610



  taosRUnLockLatch(&pExistedConsumer->lock);




41,317✔






611



  PRINT_LOG_END




41,317✔





612



  return code;




41,317✔





613



}










614












615



static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){




94,937✔





616



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




94,937✔





617



    return TSDB_CODE_INVALID_PARA;




×







618



  }










619



  taosArraySort(pTopicList, taosArrayCompareString);




94,937✔





620



  taosArrayRemoveDuplicate(pTopicList, taosArrayCompareString, freeItem);




94,937✔





621












622



  int32_t newTopicNum = taosArrayGetSize(pTopicList);




94,937✔





623



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




149,617✔





624



    int32_t gNum = mndGetGroupNumByTopic(pMnode, (const char *)taosArrayGetP(pTopicList, i));




54,786✔





625



    if (gNum >= MND_MAX_GROUP_PER_TOPIC) {




54,786✔





626



      return TSDB_CODE_TMQ_GROUP_OUT_OF_RANGE;




106✔





627



    }










628



  }










629



  return 0;




94,831✔





630



}










631












632



static int32_t buildSubConsumer(SMnode *pMnode, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer){




94,550✔





633



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




94,550✔





634



    return TSDB_CODE_INVALID_PARA;




×







635



  }










636



  int64_t         consumerId = subscribe->consumerId;




94,550✔





637



  char           *cgroup     = subscribe->cgroup;




94,550✔





638



  SMqConsumerObj *pConsumerNew     = NULL;




94,550✔





639



  SMqConsumerObj *pExistedConsumer = NULL;




94,550✔






640



  int32_t lino = 0;




94,550✔






641



  PRINT_LOG_START




94,550✔





642



  int32_t code = mndAcquireConsumer(pMnode, consumerId, &pExistedConsumer);




94,550✔





643



  if (code != 0) {




94,550✔





644



    mInfo("receive tmq subscribe request from new consumer:0x%" PRIx64




53,233✔





645



              ",cgroup:%s, numOfTopics:%d", consumerId,










646



          subscribe->cgroup, (int32_t)taosArrayGetSize(subscribe->topicNames));










647












648



    MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_INSERT_SUB, NULL, subscribe, &pConsumerNew));




53,233✔





649



  } else {










650



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




41,317✔





651












652



    mInfo("receive tmq subscribe request from existed consumer:0x%" PRIx64




41,317✔





653



              ",cgroup:%s, current status:%d(%s), subscribe topic num: %d",










654



          consumerId, subscribe->cgroup, status, mndConsumerStatusName(status),










655



          (int32_t)taosArrayGetSize(subscribe->topicNames));










656












657



    if (status != MQ_CONSUMER_STATUS_READY) {




41,317✔





658



      code = TSDB_CODE_MND_CONSUMER_NOT_READY;




×







659



      goto END;




×







660



    }










661



    MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_UPDATE_SUB, NULL, subscribe, &pConsumerNew));




41,317✔





662



    MND_TMQ_RETURN_CHECK(getTopicAddDelete(pExistedConsumer, pConsumerNew));




41,317✔





663



  }










664



  if (ppConsumer){




94,201✔





665



    *ppConsumer = pConsumerNew;




94,201✔






666



    pConsumerNew = NULL;




94,201✔





667



  }










668












669



END:




94,532✔






670



  PRINT_LOG_END




94,550✔





671



  mndReleaseConsumer(pMnode, pExistedConsumer);




94,550✔





672



  tDeleteSMqConsumerObj(pConsumerNew);




94,550✔





673



  return code;




94,550✔





674



}










675












676



int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {




131,678✔





677



  if (pMsg == NULL) {




131,678✔





678



    return TSDB_CODE_INVALID_PARA;




×







679



  }










680



  SMnode *pMnode = pMsg->info.node;




131,678✔





681



  char   *msgStr = pMsg->pCont;




131,678✔





682



  int32_t code = 0;




131,678✔






683



  int32_t lino = 0;




131,678✔





684



  SMqConsumerObj *pConsumerNew = NULL;




131,678✔





685



  STrans         *pTrans = NULL;




131,678✔





686












687



  PRINT_LOG_START




131,678✔





688



  SCMSubscribeReq subscribe = {0};




131,678✔





689



  MND_TMQ_RETURN_CHECK(tDeserializeSCMSubscribeReq(msgStr, &subscribe, pMsg->contLen));




263,356✔





690



  bool unSubscribe = (taosArrayGetSize(subscribe.topicNames) == 0);




131,678✔





691



  if(unSubscribe){




131,678✔





692



    SMqConsumerObj *pConsumerTmp = NULL;




77,424✔





693



    MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, subscribe.consumerId, &pConsumerTmp));




77,424✔






694



    taosRLockLatch(&pConsumerTmp->lock);




75,630✔






695



    size_t topicNum = taosArrayGetSize(pConsumerTmp->assignedTopics);




75,630✔






696



    taosRUnLockLatch(&pConsumerTmp->lock);




75,630✔






697



    mndReleaseConsumer(pMnode, pConsumerTmp);




75,630✔






698



    if (topicNum == 0){




75,630✔





699



      goto END;




34,947✔





700



    }










701



  }










702



  MND_TMQ_RETURN_CHECK(checkAndSortTopic(pMnode, subscribe.topicNames));




94,937✔





703



  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY,




94,831✔





704



                          (unSubscribe ? TRN_CONFLICT_NOTHING :TRN_CONFLICT_DB_INSIDE),










705



                          pMsg, "subscribe");










706



  MND_TMQ_NULL_CHECK(pTrans);




94,831✔





707












708



  MND_TMQ_RETURN_CHECK(validateTopics(pTrans, &subscribe, pMnode, pMsg->info.conn.user));




94,831✔





709



  MND_TMQ_RETURN_CHECK(buildSubConsumer(pMnode, &subscribe, &pConsumerNew));




94,550✔





710



  MND_TMQ_RETURN_CHECK(mndSetConsumerCommitLogs(pTrans, pConsumerNew));




94,201✔





711



  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));




94,201✔





712



  code = TSDB_CODE_ACTION_IN_PROGRESS;




94,201✔





713












714



END:




131,678✔





715



  mndTransDrop(pTrans);




131,678✔





716



  tDeleteSMqConsumerObj(pConsumerNew);




131,678✔





717



  taosArrayDestroyP(subscribe.topicNames, NULL);




131,678✔





718



  code = (code == TSDB_CODE_TMQ_NO_NEED_REBALANCE || code == TSDB_CODE_MND_CONSUMER_NOT_EXIST) ? 0 : code;




131,678✔






719



  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS){




131,678✔






720



    mError("tmq subscribe request from consumer:0x%" PRIx64 " failed, code:%d", subscribe.consumerId, code);




387✔





721



  } else {











722



    mInfo("tmq subscribe request from consumer:0x%" PRIx64 " processed, code:%d", subscribe.consumerId, code);




131,291✔





723



  }










724



  return code;




131,678✔





725



}










726












727



SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {




267,944✔





728



  if (pConsumer == NULL) {




267,944✔





729



    return NULL;




×







730



  }










731



  int32_t code = 0;




267,944✔





732



  int32_t lino = 0;




267,944✔





733



  terrno = TSDB_CODE_OUT_OF_MEMORY;




267,944✔





734












735



  void   *buf = NULL;




267,944✔





736



  int32_t tlen = tEncodeSMqConsumerObj(NULL, pConsumer);




267,944✔





737



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




267,944✔





738












739



  SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);




267,944✔





740



  if (pRaw == NULL) goto CM_ENCODE_OVER;




267,944✔





741












742



  buf = taosMemoryMalloc(tlen);




267,944✔





743



  if (buf == NULL) goto CM_ENCODE_OVER;




267,944✔





744












745



  void *abuf = buf;




267,944✔





746



  if(tEncodeSMqConsumerObj(&abuf, pConsumer) < 0){




267,944✔





747



    goto CM_ENCODE_OVER;




×







748



  }










749












750



  int32_t dataPos = 0;




267,944✔





751



  SDB_SET_INT32(pRaw, dataPos, tlen, CM_ENCODE_OVER);




267,944✔





752



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




267,944✔





753



  SDB_SET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_ENCODE_OVER);




267,944✔





754



  SDB_SET_DATALEN(pRaw, dataPos, CM_ENCODE_OVER);




267,944✔





755












756



  terrno = TSDB_CODE_SUCCESS;




267,944✔





757












758



CM_ENCODE_OVER:




267,944✔





759



  taosMemoryFreeClear(buf);




267,944✔





760



  if (terrno != 0) {




267,944✔





761



    mError("consumer:0x%" PRIx64 " failed to encode to raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());




×







762



    sdbFreeRaw(pRaw);




×







763



    return NULL;




×







764



  }










765












766



  mTrace("consumer:0x%" PRIx64 ", encode to raw:%p, row:%p", pConsumer->consumerId, pRaw, pConsumer);




267,944✔





767



  return pRaw;




267,944✔





768



}










769












770



SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {




232,445✔





771



  if (pRaw == NULL) {




232,445✔





772



    return NULL;




×







773



  }










774



  int32_t         code = 0;




232,445✔





775



  int32_t         lino = 0;




232,445✔





776



  SSdbRow        *pRow = NULL;




232,445✔





777



  SMqConsumerObj *pConsumer = NULL;




232,445✔





778



  void           *buf = NULL;




232,445✔





779












780



  terrno = 0;




232,445✔





781



  int8_t sver = 0;




232,445✔





782



  if (sdbGetRawSoftVer(pRaw, &sver) != 0) {




232,445✔





783



    goto CM_DECODE_OVER;




×







784



  }










785












786



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




232,445✔





787



    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;




×







788



    goto CM_DECODE_OVER;




×







789



  }










790












791



  pRow = sdbAllocRow(sizeof(SMqConsumerObj));




232,445✔





792



  if (pRow == NULL) {




232,445✔





793



    goto CM_DECODE_OVER;




×







794



  }










795












796



  pConsumer = sdbGetRowObj(pRow);




232,445✔





797



  if (pConsumer == NULL) {




232,445✔





798



    goto CM_DECODE_OVER;




×







799



  }










800












801



  int32_t dataPos = 0;




232,445✔





802



  int32_t len;




232,403✔





803



  SDB_GET_INT32(pRaw, dataPos, &len, CM_DECODE_OVER);




232,445✔





804



  buf = taosMemoryMalloc(len);




232,445✔





805



  if (buf == NULL) {




232,445✔





806



    terrno = TSDB_CODE_OUT_OF_MEMORY;




×







807



    goto CM_DECODE_OVER;




×







808



  }










809












810



  SDB_GET_BINARY(pRaw, dataPos, buf, len, CM_DECODE_OVER);




232,445✔





811



  SDB_GET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_DECODE_OVER);




232,445✔





812












813



  if (tDecodeSMqConsumerObj(buf, pConsumer, sver) == NULL) {




232,445✔





814



    terrno = TSDB_CODE_OUT_OF_MEMORY;  // TODO set correct error code




×







815



    goto CM_DECODE_OVER;




×







816



  }










817












818



  tmsgUpdateDnodeEpSet(&pConsumer->ep);




232,445✔





819












820



CM_DECODE_OVER:




232,445✔





821



  taosMemoryFreeClear(buf);




232,445✔





822



  if (terrno != TSDB_CODE_SUCCESS) {




232,445✔





823



    mError("consumer:0x%" PRIx64 " failed to decode from raw:%p since %s",




×







824



           pConsumer == NULL ? 0 : pConsumer->consumerId, pRaw, terrstr());










825



    taosMemoryFreeClear(pRow);




×







826



  }










827












828



  return pRow;




232,445✔





829



}










830












831



static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {




54,089✔





832



  if (pConsumer == NULL) {




54,089✔





833



    return TSDB_CODE_INVALID_PARA;




×







834



  }










835



  mInfo("consumer:0x%" PRIx64 " sub insert, cgroup:%s status:%d(%s) epoch:%d", pConsumer->consumerId, pConsumer->cgroup,




54,089✔





836



        pConsumer->status, mndConsumerStatusName(pConsumer->status), pConsumer->epoch);










837



  pConsumer->subscribeTime = pConsumer->createTime;




54,089✔





838



  return 0;




54,089✔





839



}










840












841



static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {




232,445✔





842



  if (pConsumer == NULL) {




232,445✔





843



    return TSDB_CODE_INVALID_PARA;




×







844



  }










845



  mInfo("consumer:0x%" PRIx64 " perform delete action, status:(%d)%s", pConsumer->consumerId, pConsumer->status,




232,445✔





846



        mndConsumerStatusName(pConsumer->status));










847



  tClearSMqConsumerObj(pConsumer);




232,445✔





848



  return 0;




232,445✔





849



}










850












851



// remove from topic list










852



static void removeFromTopicList(SArray *topicList, const char *pTopic, int64_t consumerId, char *type) {




139,696✔





853



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




139,696✔





854



    return;




×







855



  }










856



  int32_t size = taosArrayGetSize(topicList);




139,696✔





857



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




140,599✔





858



    char *p = taosArrayGetP(topicList, i);




139,094✔





859



    if (strcmp(pTopic, p) == 0) {




139,094✔





860



      taosArrayRemove(topicList, i);




138,191✔





861



      taosMemoryFree(p);




138,191✔





862












863



      mInfo("tmq rebalance consumer:0x%" PRIx64 " remove topic:%s in the %s topic list, remain newTopics:%d",




138,191✔





864



            consumerId, pTopic, type, (int)taosArrayGetSize(topicList));










865



      break;




138,191✔





866



    }










867



  }










868



}










869












870



static bool existInCurrentTopicList(const SMqConsumerObj *pConsumer, const char *pTopic) {




54,130✔





871



  if (pConsumer == NULL || pTopic == NULL) {




54,130✔





872



    return false;




×







873



  }










874



  bool    existing = false;




54,130✔





875



  int32_t size = taosArrayGetSize(pConsumer->currentTopics);




54,130✔





876



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




54,679✔





877



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




549✔





878



    if (topic && strcmp(topic, pTopic) == 0) {




549✔





879



      existing = true;




×







880



      break;




×







881



    }










882



  }










883












884



  return existing;




54,130✔





885



}










886












887



static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer) {




139,915✔





888



  if (pOldConsumer == NULL || pNewConsumer == NULL) {




139,915✔





889



    return TSDB_CODE_INVALID_PARA;




×







890



  }











891



  int32_t lino = 0;




139,915✔






892



  int32_t code = 0;




139,915✔






893



  char *pNewTopic = NULL;




139,915✔






894



  PRINT_LOG_START




139,915✔






895



  taosWLockLatch(&pOldConsumer->lock);




139,915✔





896



  mInfo("consumer:0x%" PRIx64 " perform update action, update type:%d, subscribe-time:%" PRId64 ", createTime:%" PRId64,




139,915✔





897



        pOldConsumer->consumerId, pNewConsumer->updateType, pOldConsumer->subscribeTime, pOldConsumer->createTime);










898












899



  if (pNewConsumer->updateType == CONSUMER_UPDATE_SUB) {




139,915✔





900



    TSWAP(pOldConsumer->rebNewTopics, pNewConsumer->rebNewTopics);




41,036✔





901



    TSWAP(pOldConsumer->rebRemovedTopics, pNewConsumer->rebRemovedTopics);




41,036✔





902



    TSWAP(pOldConsumer->assignedTopics, pNewConsumer->assignedTopics);




41,036✔





903












904



    pOldConsumer->subscribeTime = taosGetTimestampMs();




41,036✔





905



    pOldConsumer->status = MQ_CONSUMER_STATUS_REBALANCE;




41,036✔





906



    mInfo("consumer:0x%" PRIx64 " subscribe update, modify existed consumer", pOldConsumer->consumerId);




41,036✔





907



  } else if (pNewConsumer->updateType == CONSUMER_UPDATE_REB) {




98,879✔





908



    (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);




1,966✔






909



    pOldConsumer->status = MQ_CONSUMER_STATUS_READY;




1,966✔





910



    pOldConsumer->rebalanceTime = taosGetTimestampMs();




1,966✔





911



    mInfo("tmq rebalance consumer:0x%" PRIx64 " rebalance update, only rebalance time", pOldConsumer->consumerId);




1,966✔





912



  } else if (pNewConsumer->updateType == CONSUMER_ADD_REB) {




96,913✔





913



    void *tmp = taosArrayGetP(pNewConsumer->rebNewTopics, 0);




54,130✔






914



    MND_TMQ_NULL_CHECK(tmp);




54,130✔





915



    char *pNewTopic = taosStrdup(tmp);




54,130✔






916



    MND_TMQ_NULL_CHECK(pNewTopic);




54,130✔





917



    removeFromTopicList(pOldConsumer->rebNewTopics, pNewTopic, pOldConsumer->consumerId, "new");




54,130✔





918



    bool existing = existInCurrentTopicList(pOldConsumer, pNewTopic);




54,130✔





919



    if (existing) {




54,130✔





920



      mError("tmq rebalance consumer:0x%" PRIx64 " add new topic:%s should not in currentTopics", pOldConsumer->consumerId, pNewTopic);




×







921



    } else {











922



      MND_TMQ_NULL_CHECK(taosArrayPush(pOldConsumer->currentTopics, &pNewTopic));




108,260✔






923



      pNewTopic = NULL;




54,130✔





924



      taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString);




54,130✔





925



    }










926












927



    int32_t status = pOldConsumer->status;




54,130✔





928



    if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {




54,130✔





929



      pOldConsumer->status = MQ_CONSUMER_STATUS_READY;




53,598✔





930



    }










931












932



    pOldConsumer->rebalanceTime = taosGetTimestampMs();




54,130✔





933



    (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);




54,130✔





934












935



    mInfo("tmq rebalance consumer:0x%" PRIx64 " rebalance update add, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64




54,130✔





936



          ", current topics:%d, newTopics:%d, removeTopics:%d",










937



          pOldConsumer->consumerId, status, mndConsumerStatusName(status), pOldConsumer->status,










938



          mndConsumerStatusName(pOldConsumer->status), pOldConsumer->epoch, pOldConsumer->rebalanceTime,










939



          (int)taosArrayGetSize(pOldConsumer->currentTopics), (int)taosArrayGetSize(pOldConsumer->rebNewTopics),










940



          (int)taosArrayGetSize(pOldConsumer->rebRemovedTopics));










941












942



  } else if (pNewConsumer->updateType == CONSUMER_REMOVE_REB) {




42,783✔





943



    char *topic = taosArrayGetP(pNewConsumer->rebRemovedTopics, 0);




42,783✔






944



    MND_TMQ_NULL_CHECK(topic);




42,783✔





945



    removeFromTopicList(pOldConsumer->rebRemovedTopics, topic, pOldConsumer->consumerId, "remove");




42,783✔





946



    removeFromTopicList(pOldConsumer->currentTopics, topic, pOldConsumer->consumerId, "current");




42,783✔





947












948



    int32_t status = pOldConsumer->status;




42,783✔





949



    if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {




42,783✔





950



      pOldConsumer->status = MQ_CONSUMER_STATUS_READY;




42,251✔





951



    }










952



    pOldConsumer->rebalanceTime = taosGetTimestampMs();




42,783✔





953



    (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);




42,783✔





954












955



    mInfo("tmq rebalanceconsumer:0x%" PRIx64 " rebalance update remove, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64




42,783✔





956



          ", current topics:%d, newTopics:%d, removeTopics:%d",










957



          pOldConsumer->consumerId, status, mndConsumerStatusName(status), pOldConsumer->status,










958



          mndConsumerStatusName(pOldConsumer->status), pOldConsumer->epoch, pOldConsumer->rebalanceTime,










959



          (int)taosArrayGetSize(pOldConsumer->currentTopics), (int)taosArrayGetSize(pOldConsumer->rebNewTopics),










960



          (int)taosArrayGetSize(pOldConsumer->rebRemovedTopics));










961



  }










962













NEW


963



END:




×








964



  taosMemoryFree(pNewTopic);




139,915✔





965



  taosWUnLockLatch(&pOldConsumer->lock);




139,915✔





966













967



  PRINT_LOG_END




139,915✔





968













969



  return code;




139,915✔





970



}










971












972



int32_t mndAcquireConsumer(SMnode *pMnode, int64_t consumerId, SMqConsumerObj** pConsumer) {




1,414,611✔





973



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




1,414,611✔





974



    return TSDB_CODE_INVALID_PARA;




×







975



  }










976



  SSdb           *pSdb = pMnode->pSdb;




1,414,611✔





977



  *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);




1,414,611✔





978



  if (*pConsumer == NULL) {




1,414,505✔





979



    return TSDB_CODE_MND_CONSUMER_NOT_EXIST;




61,760✔





980



  }










981



  return 0;




1,352,851✔





982



}










983












984



void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {




12,072,248✔





985



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




12,072,248✔





986



    return;




10,338,322✔





987



  }










988



  SSdb *pSdb = pMnode->pSdb;




1,733,926✔





989



  sdbRelease(pSdb, pConsumer);




1,733,926✔





990



}










991













992



static int32_t buildResult(SMqConsumerObj *pConsumer, SShowObj *pShow, SSDataBlock *pBlock, int32_t numOfRows, const char* showTopic, bool hasTopic) {




21,257✔






993



  int32_t         code = 0;




21,257✔






994



  int32_t         lino = 0;




21,257✔






995



  SColumnInfoData *pColInfo = NULL;




21,257✔






996



  int32_t          cols = 0;




21,257✔






997



  char * parasStr = NULL;




21,257✔






998



  char           *status = NULL;




21,257✔





999













1000



  PRINT_LOG_START




21,257✔





1001



  // consumer id











1002



  char consumerIdHex[TSDB_CONSUMER_ID_LEN + VARSTR_HEADER_SIZE] = {0};




21,257✔






1003



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




21,257✔






1004



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




21,257✔





1005













1006



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




21,257✔






1007



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1008



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




21,257✔





1009












1010



  // consumer group











1011



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




21,257✔






1012



  STR_TO_VARSTR(cgroup, pConsumer->cgroup);




21,257✔





1013













1014



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




21,257✔






1015



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1016



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




21,257✔





1017












1018



  // client id











1019



  char clientId[TSDB_CLIENT_ID_LEN + VARSTR_HEADER_SIZE] = {0};




21,257✔






1020



  STR_TO_VARSTR(clientId, pConsumer->clientId);




21,257✔





1021













1022



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




21,257✔






1023



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1024



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




21,257✔





1025












1026



  // user











1027



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




21,257✔






1028



  STR_TO_VARSTR(user, pConsumer->user);




21,257✔





1029













1030



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




21,257✔






1031



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1032



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




21,257✔





1033












1034



  // fqdn











1035



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




21,257✔






1036



  STR_TO_VARSTR(fqdn, pConsumer->fqdn);




21,257✔





1037













1038



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




21,257✔






1039



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1040



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




21,257✔





1041












1042



  // status











1043



  const char *pStatusName = mndConsumerStatusName(pConsumer->status);




21,257✔






1044



  status = taosMemoryCalloc(1, pShow->pMeta->pSchemas[cols].bytes);




21,257✔






1045



  MND_TMQ_NULL_CHECK(status);




21,257✔






1046



  STR_TO_VARSTR(status, pStatusName);




21,257✔





1047













1048



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




21,257✔






1049



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1050



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




21,257✔





1051












1052



  // one subscribed topic











1053



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




21,257✔






1054



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1055



  if (hasTopic) {




21,257✔






1056



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




21,257✔






1057



    mndTopicGetShowName(showTopic, topic + VARSTR_HEADER_SIZE);




21,257✔






1058



    *(VarDataLenT *)(topic) = strlen(topic + VARSTR_HEADER_SIZE);




21,257✔






1059



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




21,257✔





1060



  } else {











NEW


1061



    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, NULL, true));




×







1062



  }










1063












1064



  // up time











1065



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




21,257✔






1066



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1067



  MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->createTime, false));




21,257✔





1068












1069



  // subscribe time











1070



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




21,257✔






1071



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1072



  MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->subscribeTime, false));




21,257✔





1073












1074



  // rebalance time











1075



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




21,257✔






1076



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1077



  MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->rebalanceTime, pConsumer->rebalanceTime == 0));




21,257✔





1078













1079



  char         buf[TSDB_OFFSET_LEN] = {0};




21,257✔






1080



  STqOffsetVal pVal = {.type = pConsumer->resetOffsetCfg};




21,257✔






1081



  tFormatOffset(buf, TSDB_OFFSET_LEN, &pVal);




21,257✔





1082













1083



  parasStr = taosMemoryCalloc(1, pShow->pMeta->pSchemas[cols].bytes);




21,257✔






1084



  MND_TMQ_NULL_CHECK(parasStr);




21,257✔






1085



  (void)snprintf(varDataVal(parasStr), pShow->pMeta->pSchemas[cols].bytes - VARSTR_HEADER_SIZE, "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName,




42,514✔






1086



          pConsumer->autoCommit, pConsumer->autoCommitInterval, buf);




21,257✔






1087



  varDataSetLen(parasStr, strlen(varDataVal(parasStr)));




21,257✔





1088













1089



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




21,257✔






1090



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1091



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




21,257✔





1092












1093



  // rebalance time











1094



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




21,257✔






1095



  MND_TMQ_NULL_CHECK(pColInfo);




21,257✔






1096



  MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->pollTime, pConsumer->pollTime == 0));




21,257✔





1097













1098



END:




21,257✔






1099



  PRINT_LOG_END




21,257✔






1100



  taosMemoryFreeClear(status);




21,257✔






1101



  taosMemoryFreeClear(parasStr);




21,257✔






1102



  return code;




21,257✔





1103



}










1104













1105



static int32_t retrieveOneConsumer(SMqConsumerObj *pConsumer, int32_t* numOfRows, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {




21,736✔






1106



  int32_t         code = 0;




21,736✔






1107



  int32_t         lino = 0;




21,736✔






1108



  PRINT_LOG_START




21,736✔






1109



  taosRLockLatch(&pConsumer->lock);




21,736✔






1110



  mDebug("showing consumer:0x%" PRIx64, pConsumer->consumerId);




21,736✔






1111



  if (taosArrayGetSize(pConsumer->assignedTopics) == 0) {




21,736✔






1112



    mInfo("showing consumer:0x%" PRIx64 " no assigned topic, skip", pConsumer->consumerId);




479✔






1113



    goto END;




479✔





1114



  }










1115












1116













1117



  int32_t topicSz = taosArrayGetSize(pConsumer->assignedTopics);




21,257✔






1118



  bool    hasTopic = true;




21,257✔






1119



  if (topicSz == 0) {




21,257✔






NEW


1120



    hasTopic = false;




×








NEW


1121



    topicSz = 1;




×







1122



  }










1123













1124



  if (*numOfRows + topicSz > rowsCapacity) {




21,257✔






NEW


1125



    MND_TMQ_RETURN_CHECK(blockDataEnsureCapacity(pBlock, *numOfRows + topicSz));




×







1126



  }










1127













1128



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




42,514✔






1129



    MND_TMQ_RETURN_CHECK(buildResult(pConsumer, pShow, pBlock, *numOfRows, taosArrayGetP(pConsumer->assignedTopics, i), hasTopic));




21,257✔






1130



    (*numOfRows)++;




21,257✔





1131



  }










1132













1133



END:




21,257✔






1134



  taosRUnLockLatch(&pConsumer->lock);




21,736✔






1135



  PRINT_LOG_END




21,736✔






1136



  return code;




21,736✔





1137



}










1138












1139



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




9,665✔





1140



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




9,665✔





1141



    return TSDB_CODE_INVALID_PARA;




×







1142



  }










1143



  SMnode         *pMnode = pReq->info.node;




9,665✔





1144



  SSdb           *pSdb = pMnode->pSdb;




9,665✔





1145



  int32_t         numOfRows = 0;




9,665✔





1146



  SMqConsumerObj *pConsumer = NULL;




9,665✔





1147



  int32_t         code = 0;




9,665✔






1148



  int32_t         lino = 0;




9,665✔






1149



  PRINT_LOG_START




9,665✔





1150












1151



  while (numOfRows < rowsCapacity) {




31,401✔





1152



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




31,401✔





1153



    if (pShow->pIter == NULL) {




31,401✔





1154



      break;




9,665✔





1155



    }











1156



    MND_TMQ_RETURN_CHECK(retrieveOneConsumer(pConsumer, &numOfRows, pShow, pBlock, rowsCapacity));




21,736✔





1157



    










1158



    pBlock->info.rows = numOfRows;




21,736✔






1159



    sdbRelease(pSdb, pConsumer);




21,736✔






1160



    pConsumer = NULL;




21,736✔





1161



  }










1162












1163



  pShow->numOfRows += numOfRows;




9,665✔





1164












1165



END:




9,665✔






1166



  sdbRelease(pSdb, pConsumer);




9,665✔






1167



  sdbCancelFetch(pSdb, pShow->pIter);




9,665✔






1168



  if (code != 0) {




9,665✔






NEW


1169



    mError("show consumer failed, code:%d", code);




×








NEW


1170



    return code;




×







1171



  } else {











1172



    mDebug("show consumer processed, numOfRows:%d", numOfRows);




9,665✔






1173



    return numOfRows;




9,665✔





1174



  }










1175



}










1176












1177



static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {




×







1178



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




×







1179



  SSdb *pSdb = pMnode->pSdb;




×







1180



  sdbCancelFetchByType(pSdb, pIter, SDB_CONSUMER);




×







1181



}










1182












1183



const char *mndConsumerStatusName(int status) {




1,257,287✔





1184



  switch (status) {




1,257,287✔





1185



    case MQ_CONSUMER_STATUS_READY:




522,500✔





1186



      return "ready";




522,500✔





1187



    case MQ_CONSUMER_STATUS_REBALANCE:




734,787✔





1188



      return "rebalancing";




734,787✔





1189



    default:




×







1190



      return "unknown";




×







1191



  }










1192



}
























    

  • 

    •