xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #26729 from taosdata/opti/TD-31017 

fix:[TD-31017]process return value in mnode for tmq
This commit is contained in:
dapan1121 2024-07-24 09:19:03 +08:00 committed by GitHub
parent 02c90a2211 04f51842e4
commit 5384c2a5d6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 896 additions and 1091 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
include/common/tmsg.h
View File

@ -2828,19 +2828,24 @@ static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubsc
return tlen; return tlen;
} }
static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) { static FORCE_INLINE int32_t tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
buf = taosDecodeFixedI64(buf, &pReq->consumerId); buf = taosDecodeFixedI64(buf, &pReq->consumerId);
buf = taosDecodeStringTo(buf, pReq->cgroup); buf = taosDecodeStringTo(buf, pReq->cgroup);
buf = taosDecodeStringTo(buf, pReq->clientId); buf = taosDecodeStringTo(buf, pReq->clientId);
int32_t topicNum; int32_t topicNum = 0;
buf = taosDecodeFixedI32(buf, &topicNum); buf = taosDecodeFixedI32(buf, &topicNum);
pReq->topicNames = taosArrayInit(topicNum, sizeof(void*)); pReq->topicNames = taosArrayInit(topicNum, sizeof(void*));
if (pReq->topicNames == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int32_t i = 0; i < topicNum; i++) { for (int32_t i = 0; i < topicNum; i++) {
char* name; char* name = NULL;
buf = taosDecodeString(buf, &name); buf = taosDecodeString(buf, &name);
taosArrayPush(pReq->topicNames, &name); if (taosArrayPush(pReq->topicNames, &name) == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
} }
buf = taosDecodeFixedI8(buf, &pReq->withTbName); buf = taosDecodeFixedI8(buf, &pReq->withTbName);
@ -2849,7 +2854,7 @@ static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq
buf = taosDecodeFixedI8(buf, &pReq->resetOffsetCfg); buf = taosDecodeFixedI8(buf, &pReq->resetOffsetCfg);
buf = taosDecodeFixedI8(buf, &pReq->enableReplay); buf = taosDecodeFixedI8(buf, &pReq->enableReplay);
buf = taosDecodeFixedI8(buf, &pReq->enableBatchMeta); buf = taosDecodeFixedI8(buf, &pReq->enableBatchMeta);
return buf; return 0;
} }
typedef struct { typedef struct {

3
include/util/tdef.h
View File

@ -556,7 +556,8 @@ typedef struct {
char name[TSDB_LOG_VAR_LEN]; char name[TSDB_LOG_VAR_LEN];
} SLogVar; } SLogVar;
#define TMQ_SEPARATOR ':' #define TMQ_SEPARATOR ":"
#define TMQ_SEPARATOR_CHAR ':'
enum { enum {
SND_WORKER_TYPE__SHARED = 1, SND_WORKER_TYPE__SHARED = 1,

18
source/client/src/clientTmq.c
View File

@ -456,11 +456,7 @@ static int32_t doSendCommitMsg(tmq_t* tmq, int32_t vgId, SEpSet* epSet, STqOffse
pOffset.consumerId = tmq->consumerId; pOffset.consumerId = tmq->consumerId;
pOffset.offset.val = *offset; pOffset.offset.val = *offset;
int32_t groupLen = strlen(tmq->groupId); (void)snprintf(pOffset.offset.subKey, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", tmq->groupId, TMQ_SEPARATOR, pTopicName);
memcpy(pOffset.offset.subKey, tmq->groupId, groupLen);
pOffset.offset.subKey[groupLen] = TMQ_SEPARATOR;
strcpy(pOffset.offset.subKey + groupLen + 1, pTopicName);
int32_t len = 0; int32_t len = 0;
int32_t code = 0; int32_t code = 0;
tEncodeSize(tEncodeMqVgOffset, &pOffset, len, code); tEncodeSize(tEncodeMqVgOffset, &pOffset, len, code);
@ -1667,11 +1663,7 @@ static bool doUpdateLocalEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp)
} }
void tmqBuildConsumeReqImpl(SMqPollReq* pReq, tmq_t* tmq, int64_t timeout, SMqClientTopic* pTopic, SMqClientVg* pVg) { void tmqBuildConsumeReqImpl(SMqPollReq* pReq, tmq_t* tmq, int64_t timeout, SMqClientTopic* pTopic, SMqClientVg* pVg) {
int32_t groupLen = strlen(tmq->groupId); (void)snprintf(pReq->subKey, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", tmq->groupId, TMQ_SEPARATOR, pTopic->topicName);
memcpy(pReq->subKey, tmq->groupId, groupLen);
pReq->subKey[groupLen] = TMQ_SEPARATOR;
strcpy(pReq->subKey + groupLen + 1, pTopic->topicName);
pReq->withTbName = tmq->withTbName; pReq->withTbName = tmq->withTbName;
pReq->consumerId = tmq->consumerId; pReq->consumerId = tmq->consumerId;
pReq->timeout = timeout; pReq->timeout = timeout;
@ -2902,11 +2894,7 @@ int64_t getCommittedFromServer(tmq_t* tmq, char* tname, int32_t vgId, SEpSet* ep
SMqVgOffset pOffset = {0}; SMqVgOffset pOffset = {0};
pOffset.consumerId = tmq->consumerId; pOffset.consumerId = tmq->consumerId;
(void)snprintf(pOffset.offset.subKey, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", tmq->groupId, TMQ_SEPARATOR, tname);
int32_t groupLen = strlen(tmq->groupId);
memcpy(pOffset.offset.subKey, tmq->groupId, groupLen);
pOffset.offset.subKey[groupLen] = TMQ_SEPARATOR;
strcpy(pOffset.offset.subKey + groupLen + 1, tname);
int32_t len = 0; int32_t len = 0;
tEncodeSize(tEncodeMqVgOffset, &pOffset, len, code); tEncodeSize(tEncodeMqVgOffset, &pOffset, len, code);

14
source/common/src/tmsg.c
View File

@ -9197,23 +9197,23 @@ int32_t tDecodeSTqOffsetVal(SDecoder *pDecoder, STqOffsetVal *pOffsetVal) {
int32_t tFormatOffset(char *buf, int32_t maxLen, const STqOffsetVal *pVal) { int32_t tFormatOffset(char *buf, int32_t maxLen, const STqOffsetVal *pVal) {
if (pVal->type == TMQ_OFFSET__RESET_NONE) { if (pVal->type == TMQ_OFFSET__RESET_NONE) {
snprintf(buf, maxLen, "none"); (void)snprintf(buf, maxLen, "none");
} else if (pVal->type == TMQ_OFFSET__RESET_EARLIEST) { } else if (pVal->type == TMQ_OFFSET__RESET_EARLIEST) {
snprintf(buf, maxLen, "earliest"); (void)snprintf(buf, maxLen, "earliest");
} else if (pVal->type == TMQ_OFFSET__RESET_LATEST) { } else if (pVal->type == TMQ_OFFSET__RESET_LATEST) {
snprintf(buf, maxLen, "latest"); (void)snprintf(buf, maxLen, "latest");
} else if (pVal->type == TMQ_OFFSET__LOG) { } else if (pVal->type == TMQ_OFFSET__LOG) {
snprintf(buf, maxLen, "wal:%" PRId64, pVal->version); (void)snprintf(buf, maxLen, "wal:%" PRId64, pVal->version);
} else if (pVal->type == TMQ_OFFSET__SNAPSHOT_DATA || pVal->type == TMQ_OFFSET__SNAPSHOT_META) { } else if (pVal->type == TMQ_OFFSET__SNAPSHOT_DATA || pVal->type == TMQ_OFFSET__SNAPSHOT_META) {
if (IS_VAR_DATA_TYPE(pVal->primaryKey.type)) { if (IS_VAR_DATA_TYPE(pVal->primaryKey.type)) {
char *tmp = taosMemoryCalloc(1, pVal->primaryKey.nData + 1); char *tmp = taosMemoryCalloc(1, pVal->primaryKey.nData + 1);
if (tmp == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (tmp == NULL) return TSDB_CODE_OUT_OF_MEMORY;
memcpy(tmp, pVal->primaryKey.pData, pVal->primaryKey.nData); (void)memcpy(tmp, pVal->primaryKey.pData, pVal->primaryKey.nData);
snprintf(buf, maxLen, "tsdb:%" PRId64 "|%" PRId64 ",pk type:%d,val:%s", pVal->uid, pVal->ts, (void)snprintf(buf, maxLen, "tsdb:%" PRId64 "|%" PRId64 ",pk type:%d,val:%s", pVal->uid, pVal->ts,
pVal->primaryKey.type, tmp); pVal->primaryKey.type, tmp);
taosMemoryFree(tmp); taosMemoryFree(tmp);
} else { } else {
snprintf(buf, maxLen, "tsdb:%" PRId64 "|%" PRId64 ",pk type:%d,val:%" PRId64, pVal->uid, pVal->ts, (void)snprintf(buf, maxLen, "tsdb:%" PRId64 "|%" PRId64 ",pk type:%d,val:%" PRId64, pVal->uid, pVal->ts,
pVal->primaryKey.type, pVal->primaryKey.val); pVal->primaryKey.type, pVal->primaryKey.val);
} }
} else { } else {

22
source/dnode/mnode/impl/inc/mndConsumer.h
View File

@ -30,10 +30,10 @@ enum {
int32_t mndInitConsumer(SMnode *pMnode); int32_t mndInitConsumer(SMnode *pMnode);
void mndCleanupConsumer(SMnode *pMnode); void mndCleanupConsumer(SMnode *pMnode);
void mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo* info); int32_t mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo* info);
SMqConsumerObj *mndAcquireConsumer(SMnode *pMnode, int64_t consumerId); int32_t mndAcquireConsumer(SMnode *pMnode, int64_t consumerId, SMqConsumerObj** pConsumer);
void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer); void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer);
SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer); SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer);
SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw); SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw);
@ -43,6 +43,22 @@ int32_t mndSetConsumerDropLogs(STrans *pTrans, SMqConsumerObj *pConsumer);
const char *mndConsumerStatusName(int status); const char *mndConsumerStatusName(int status);
#define MND_TMQ_NULL_CHECK(c) \
do { \
if (c == NULL) { \
code = TSDB_CODE_OUT_OF_MEMORY; \
goto END; \
} \
} while (0)
#define MND_TMQ_RETURN_CHECK(c) \
do { \
code = c; \
if (code != 0) { \
goto END; \
} \
} while (0)
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

7
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -622,7 +622,8 @@ typedef struct {
int32_t resetOffsetCfg; int32_t resetOffsetCfg;
} SMqConsumerObj; } SMqConsumerObj;
SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType, char *topic, SCMSubscribeReq *subscribe); int32_t tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType,
char *topic, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer);
void tClearSMqConsumerObj(SMqConsumerObj* pConsumer); void tClearSMqConsumerObj(SMqConsumerObj* pConsumer);
void tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer); void tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer);
int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer); int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer);
@ -665,8 +666,8 @@ typedef struct {
char* qmsg; // SubPlanToString char* qmsg; // SubPlanToString
} SMqSubscribeObj; } SMqSubscribeObj;
SMqSubscribeObj* tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]); int32_t tNewSubscribeObj(const char *key, SMqSubscribeObj **ppSub);
SMqSubscribeObj* tCloneSubscribeObj(const SMqSubscribeObj* pSub); int32_t tCloneSubscribeObj(const SMqSubscribeObj* pSub, SMqSubscribeObj **ppSub);
void tDeleteSubscribeObj(SMqSubscribeObj* pSub); void tDeleteSubscribeObj(SMqSubscribeObj* pSub);
int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub); int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub);
void* tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub, int8_t sver); void* tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub, int8_t sver);

5
source/dnode/mnode/impl/inc/mndSubscribe.h
View File

@ -26,12 +26,9 @@ int32_t mndInitSubscribe(SMnode *pMnode);
void mndCleanupSubscribe(SMnode *pMnode); void mndCleanupSubscribe(SMnode *pMnode);
int32_t mndGetGroupNumByTopic(SMnode *pMnode, const char *topicName); int32_t mndGetGroupNumByTopic(SMnode *pMnode, const char *topicName);
SMqSubscribeObj *mndAcquireSubscribe(SMnode *pMnode, const char *CGroup, const char *topicName); int32_t mndAcquireSubscribeByKey(SMnode *pMnode, const char *key, SMqSubscribeObj** pSub);
SMqSubscribeObj *mndAcquireSubscribeByKey(SMnode *pMnode, const char *key);
void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub); void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub);
void mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName);
int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topic); int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topic);
int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub); int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub);

5
source/dnode/mnode/impl/inc/mndTopic.h
View File

@ -25,13 +25,12 @@ extern "C" {
int32_t mndInitTopic(SMnode *pMnode); int32_t mndInitTopic(SMnode *pMnode);
void mndCleanupTopic(SMnode *pMnode); void mndCleanupTopic(SMnode *pMnode);
SMqTopicObj *mndAcquireTopic(SMnode *pMnode, const char *topicName); int32_t mndAcquireTopic(SMnode *pMnode, const char *topicName, SMqTopicObj **pTopic);
void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic); void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic);
int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb); int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
bool mndTopicExistsForDb(SMnode *pMnode, SDbObj *pDb); bool mndTopicExistsForDb(SMnode *pMnode, SDbObj *pDb);
const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]); void mndTopicGetShowName(const char* fullTopic, char* topic);
int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic);
int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics); int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics);
#ifdef __cplusplus #ifdef __cplusplus

510
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -40,7 +40,6 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg);
static int32_t mndProcessAskEpReq(SRpcMsg *pMsg); static int32_t mndProcessAskEpReq(SRpcMsg *pMsg);
static int32_t mndProcessMqHbReq(SRpcMsg *pMsg); static int32_t mndProcessMqHbReq(SRpcMsg *pMsg);
static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg); static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg);
static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg);
int32_t mndInitConsumer(SMnode *pMnode) { int32_t mndInitConsumer(SMnode *pMnode) {
SSdbTable table = { SSdbTable table = {
@ -57,7 +56,6 @@ int32_t mndInitConsumer(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq); mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq); mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq);
// mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessMqTimerMsg); // mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessMqTimerMsg);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
mndSetMsgHandle(pMnode, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, mndProcessConsumerClearMsg); mndSetMsgHandle(pMnode, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, mndProcessConsumerClearMsg);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer); mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer);
@ -68,13 +66,10 @@ int32_t mndInitConsumer(SMnode *pMnode) {
void mndCleanupConsumer(SMnode *pMnode) {} void mndCleanupConsumer(SMnode *pMnode) {}
void mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo *info) { int32_t mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo *info) {
void *msg = rpcMallocCont(sizeof(int64_t)); int32_t code = 0;
if (msg == NULL) { void *msg = rpcMallocCont(sizeof(int64_t));
mError("consumer:0x%" PRIx64 " failed to clear consumer due to out of memory. alloc size:%d", consumerId, MND_TMQ_NULL_CHECK(msg);
(int32_t)sizeof(int64_t));
return;
}
*(int64_t*)msg = consumerId; *(int64_t*)msg = consumerId;
SRpcMsg rpcMsg = { SRpcMsg rpcMsg = {
@ -85,108 +80,51 @@ void mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SR
}; };
mInfo("mndSendConsumerMsg type:%d consumer:0x%" PRIx64, msgType, consumerId); mInfo("mndSendConsumerMsg type:%d consumer:0x%" PRIx64, msgType, consumerId);
int32_t code = tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg); MND_TMQ_RETURN_CHECK(tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg));
if (code != 0){ return code;
mError("consumer:%"PRId64" send consumer msg:%d error:%d", consumerId, msgType, code);
} END:
taosMemoryFree(msg);
return code;
} }
static int32_t validateTopics(const SArray *pTopicList, SMnode *pMnode, const char *pUser, static int32_t validateTopics(const SArray *pTopicList, SMnode *pMnode, const char *pUser,
bool enableReplay) { bool enableReplay) {
SMqTopicObj *pTopic = NULL; SMqTopicObj *pTopic = NULL;
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
int32_t numOfTopics = taosArrayGetSize(pTopicList); int32_t numOfTopics = taosArrayGetSize(pTopicList);
for (int32_t i = 0; i < numOfTopics; i++) { for (int32_t i = 0; i < numOfTopics; i++) {
char *pOneTopic = taosArrayGetP(pTopicList, i); char *pOneTopic = taosArrayGetP(pTopicList, i);
pTopic = mndAcquireTopic(pMnode, pOneTopic); MND_TMQ_RETURN_CHECK(mndAcquireTopic(pMnode, pOneTopic, &pTopic));
if (pTopic == NULL) { // terrno has been set by callee function MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pUser, MND_OPER_SUBSCRIBE, pTopic));
code = TSDB_CODE_MND_RETURN_VALUE_NULL; MND_TMQ_RETURN_CHECK(grantCheckExpire(TSDB_GRANT_SUBSCRIPTION));
if (terrno != 0) code = terrno;
goto FAILED;
}
TAOS_CHECK_GOTO(mndCheckTopicPrivilege(pMnode, pUser, MND_OPER_SUBSCRIBE, pTopic), &lino, FAILED);
TAOS_CHECK_GOTO(grantCheckExpire(TSDB_GRANT_SUBSCRIPTION), &lino, FAILED);
if (enableReplay) { if (enableReplay) {
if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) { if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) {
code = TSDB_CODE_TMQ_REPLAY_NOT_SUPPORT; code = TSDB_CODE_TMQ_REPLAY_NOT_SUPPORT;
goto FAILED; goto END;
} else if (pTopic->ntbUid == 0 && pTopic->ctbStbUid == 0) { } else if (pTopic->ntbUid == 0 && pTopic->ctbStbUid == 0) {
SDbObj *pDb = mndAcquireDb(pMnode, pTopic->db); SDbObj *pDb = mndAcquireDb(pMnode, pTopic->db);
if (pDb == NULL) { if (pDb == NULL) {
code = TSDB_CODE_MND_RETURN_VALUE_NULL; code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno; goto END;
goto FAILED;
} }
if (pDb->cfg.numOfVgroups != 1) { if (pDb->cfg.numOfVgroups != 1) {
mndReleaseDb(pMnode, pDb); mndReleaseDb(pMnode, pDb);
code = TSDB_CODE_TMQ_REPLAY_NEED_ONE_VGROUP; code = TSDB_CODE_TMQ_REPLAY_NEED_ONE_VGROUP;
goto FAILED; goto END;
} }
mndReleaseDb(pMnode, pDb); mndReleaseDb(pMnode, pDb);
} }
} }
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
} }
return 0; return 0;
FAILED:
mndReleaseTopic(pMnode, pTopic);
TAOS_RETURN(code);
}
static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg) {
int32_t code = 0;
int32_t lino = 0;
SMnode *pMnode = pMsg->info.node;
SMqConsumerRecoverMsg *pRecoverMsg = pMsg->pCont;
SMqConsumerObj *pConsumerNew = NULL;
STrans *pTrans = NULL;
SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, pRecoverMsg->consumerId);
if (pConsumer == NULL) {
mError("cannot find consumer %" PRId64 " when processing consumer recover msg", pRecoverMsg->consumerId);
code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno;
goto END;
}
mInfo("receive consumer recover msg, consumer:0x%" PRIx64 " status:%d(%s)", pRecoverMsg->consumerId,
pConsumer->status, mndConsumerStatusName(pConsumer->status));
if (pConsumer->status != MQ_CONSUMER_STATUS_LOST) {
code = TSDB_CODE_MND_CONSUMER_NOT_READY;
goto END;
}
pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup, CONSUMER_UPDATE_REC, NULL, NULL);
if (pConsumerNew == NULL){
code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno;
goto END;
}
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "recover-csm");
if (pTrans == NULL) {
code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno;
goto END;
}
TAOS_CHECK_GOTO(validateTopics(pConsumer->assignedTopics, pMnode, pMsg->info.conn.user, false), &lino, END);
TAOS_CHECK_GOTO(mndSetConsumerCommitLogs(pTrans, pConsumerNew), &lino, END);
code = mndTransPrepare(pMnode, pTrans);
END: END:
mndReleaseConsumer(pMnode, pConsumer); mndReleaseTopic(pMnode, pTopic);
tDeleteSMqConsumerObj(pConsumerNew); return code;
mndTransDrop(pTrans);
TAOS_RETURN(code);
} }
static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) { static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {
@ -196,55 +134,39 @@ static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {
SMqConsumerClearMsg *pClearMsg = pMsg->pCont; SMqConsumerClearMsg *pClearMsg = pMsg->pCont;
SMqConsumerObj *pConsumerNew = NULL; SMqConsumerObj *pConsumerNew = NULL;
STrans *pTrans = NULL; STrans *pTrans = NULL;
SMqConsumerObj *pConsumer = NULL;
SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, pClearMsg->consumerId); MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, pClearMsg->consumerId, &pConsumer));
if (pConsumer == NULL) {
mError("consumer:0x%" PRIx64 " failed to be found to clear it", pClearMsg->consumerId);
TAOS_RETURN(code);
}
mInfo("consumer:0x%" PRIx64 " needs to be cleared, status %s", pClearMsg->consumerId, mInfo("consumer:0x%" PRIx64 " needs to be cleared, status %s", pClearMsg->consumerId,
mndConsumerStatusName(pConsumer->status)); mndConsumerStatusName(pConsumer->status));
pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup, -1, NULL, NULL); MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup, -1, NULL, NULL, &pConsumerNew));
if (pConsumerNew == NULL){
code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno;
goto END;
}
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "clear-csm"); pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "clear-csm");
if (pTrans == NULL) { MND_TMQ_NULL_CHECK(pTrans);
code = TSDB_CODE_MND_RETURN_VALUE_NULL; MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));
if (terrno != 0) code = terrno;
goto END;
}
// this is the drop action, not the update action
TAOS_CHECK_GOTO(mndSetConsumerDropLogs(pTrans, pConsumerNew), &lino, END);
code = mndTransPrepare(pMnode, pTrans); code = mndTransPrepare(pMnode, pTrans);
END: END:
mndReleaseConsumer(pMnode, pConsumer); mndReleaseConsumer(pMnode, pConsumer);
tDeleteSMqConsumerObj(pConsumerNew); tDeleteSMqConsumerObj(pConsumerNew);
mndTransDrop(pTrans); mndTransDrop(pTrans);
TAOS_RETURN(code); return code;
} }
static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRsp *rsp, char *user) { static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRsp *rsp, char *user) {
int32_t code = 0;
rsp->topicPrivileges = taosArrayInit(taosArrayGetSize(pConsumer->currentTopics), sizeof(STopicPrivilege)); rsp->topicPrivileges = taosArrayInit(taosArrayGetSize(pConsumer->currentTopics), sizeof(STopicPrivilege));
if (rsp->topicPrivileges == NULL) { MND_TMQ_NULL_CHECK(rsp->topicPrivileges);
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int32_t i = 0; i < taosArrayGetSize(pConsumer->currentTopics); i++) { for (int32_t i = 0; i < taosArrayGetSize(pConsumer->currentTopics); i++) {
char *topic = taosArrayGetP(pConsumer->currentTopics, i); char *topic = taosArrayGetP(pConsumer->currentTopics, i);
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic); SMqTopicObj *pTopic = NULL;
if (pTopic == NULL) { // terrno has been set by callee function code = mndAcquireTopic(pMnode, topic, &pTopic);
if (code != TDB_CODE_SUCCESS) {
continue; continue;
} }
STopicPrivilege *data = taosArrayReserve(rsp->topicPrivileges, 1); STopicPrivilege *data = taosArrayReserve(rsp->topicPrivileges, 1);
strcpy(data->topic, topic); MND_TMQ_NULL_CHECK(data);
(void)strcpy(data->topic, topic);
if (mndCheckTopicPrivilege(pMnode, user, MND_OPER_SUBSCRIBE, pTopic) != 0 || if (mndCheckTopicPrivilege(pMnode, user, MND_OPER_SUBSCRIBE, pTopic) != 0 ||
grantCheckExpire(TSDB_GRANT_SUBSCRIPTION) < 0) { grantCheckExpire(TSDB_GRANT_SUBSCRIPTION) < 0) {
data->noPrivilege = 1; data->noPrivilege = 1;
@ -253,22 +175,30 @@ static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRs
} }
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
} }
return 0; END:
return code;
} }
static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pConsumer){ static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pConsumer){
for (int i = 0; i < taosArrayGetSize(req->topics); i++) { for (int i = 0; i < taosArrayGetSize(req->topics); i++) {
TopicOffsetRows *data = taosArrayGet(req->topics, i); TopicOffsetRows *data = taosArrayGet(req->topics, i);
if (data == NULL){
continue;
}
mInfo("heartbeat report offset rows.%s:%s", pConsumer->cgroup, data->topicName); mInfo("heartbeat report offset rows.%s:%s", pConsumer->cgroup, data->topicName);
SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, data->topicName); SMqSubscribeObj *pSub = NULL;
if (pSub == NULL) { char key[TSDB_SUBSCRIBE_KEY_LEN] = {0};
(void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", pConsumer->cgroup, TMQ_SEPARATOR, data->topicName);
int32_t code = mndAcquireSubscribeByKey(pMnode, key, &pSub);
if (code != 0) {
mError("failed to acquire subscribe by key:%s, code:%d", key, code);
continue; continue;
} }
taosWLockLatch(&pSub->lock); taosWLockLatch(&pSub->lock);
SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t)); SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));
if (pConsumerEp) { if (pConsumerEp) {
taosArrayDestroy(pConsumerEp->offsetRows); (void)taosArrayDestroy(pConsumerEp->offsetRows);
pConsumerEp->offsetRows = data->offsetRows; pConsumerEp->offsetRows = data->offsetRows;
data->offsetRows = NULL; data->offsetRows = NULL;
} }
@ -303,36 +233,24 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
SMqHbReq req = {0}; SMqHbReq req = {0};
SMqHbRsp rsp = {0}; SMqHbRsp rsp = {0};
SMqConsumerObj *pConsumer = NULL; SMqConsumerObj *pConsumer = NULL;
MND_TMQ_RETURN_CHECK(tDeserializeSMqHbReq(pMsg->pCont, pMsg->contLen, &req));
TAOS_CHECK_GOTO(tDeserializeSMqHbReq(pMsg->pCont, pMsg->contLen, &req), NULL, end);
int64_t consumerId = req.consumerId; int64_t consumerId = req.consumerId;
pConsumer = mndAcquireConsumer(pMnode, consumerId); MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));
if (pConsumer == NULL) { MND_TMQ_RETURN_CHECK(checkPrivilege(pMnode, pConsumer, &rsp, pMsg->info.conn.user));
mError("consumer:0x%" PRIx64 " not exist", consumerId);
code = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
goto end;
}
TAOS_CHECK_GOTO(checkPrivilege(pMnode, pConsumer, &rsp, pMsg->info.conn.user), NULL, end);
atomic_store_32(&pConsumer->hbStatus, 0); atomic_store_32(&pConsumer->hbStatus, 0);
int32_t status = atomic_load_32(&pConsumer->status); int32_t status = atomic_load_32(&pConsumer->status);
if (status == MQ_CONSUMER_STATUS_LOST) { if (status == MQ_CONSUMER_STATUS_LOST) {
mInfo("try to recover consumer:0x%" PRIx64, consumerId); mInfo("try to recover consumer:0x%" PRIx64, consumerId);
mndSendConsumerMsg(pMnode, pConsumer->consumerId, TDMT_MND_TMQ_CONSUMER_RECOVER, &pMsg->info); MND_TMQ_RETURN_CHECK(mndSendConsumerMsg(pMnode, pConsumer->consumerId, TDMT_MND_TMQ_CONSUMER_RECOVER, &pMsg->info));
} }
storeOffsetRows(pMnode, &req, pConsumer); storeOffsetRows(pMnode, &req, pConsumer);
code = buildMqHbRsp(pMsg, &rsp); code = buildMqHbRsp(pMsg, &rsp);
end: END:
tDestroySMqHbRsp(&rsp); tDestroySMqHbRsp(&rsp);
mndReleaseConsumer(pMnode, pConsumer); mndReleaseConsumer(pMnode, pConsumer);
tDestroySMqHbReq(&req); tDestroySMqHbReq(&req);
TAOS_RETURN(code); return code;
} }
static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t epoch, SMqAskEpRsp *rsp){ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t epoch, SMqAskEpRsp *rsp){
@ -349,19 +267,22 @@ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t e
// handle all topics subscribed by this consumer // handle all topics subscribed by this consumer
for (int32_t i = 0; i < numOfTopics; i++) { for (int32_t i = 0; i < numOfTopics; i++) {
char *topic = taosArrayGetP(pConsumer->currentTopics, i); char *topic = taosArrayGetP(pConsumer->currentTopics, i);
SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic); SMqSubscribeObj *pSub = NULL;
// txn guarantees pSub is created char key[TSDB_SUBSCRIBE_KEY_LEN] = {0};
if (pSub == NULL) { (void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", pConsumer->cgroup, TMQ_SEPARATOR, topic);
int32_t code = mndAcquireSubscribeByKey(pMnode, key, &pSub);
if (code != 0) {
continue; continue;
} }
taosRLockLatch(&pSub->lock); taosRLockLatch(&pSub->lock);
SMqSubTopicEp topicEp = {0}; SMqSubTopicEp topicEp = {0};
strcpy(topicEp.topic, topic); (void)strcpy(topicEp.topic, topic);
// 2.1 fetch topic schema // 2.1 fetch topic schema
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic); SMqTopicObj *pTopic = NULL;
if (pTopic == NULL) { code = mndAcquireTopic(pMnode, topic, &pTopic);
if (code != TDB_CODE_SUCCESS) {
taosRUnLockLatch(&pSub->lock); taosRUnLockLatch(&pSub->lock);
mndReleaseSubscribe(pMnode, pSub); mndReleaseSubscribe(pMnode, pSub);
continue; continue;
@ -371,16 +292,27 @@ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t e
topicEp.schema.nCols = pTopic->schema.nCols; topicEp.schema.nCols = pTopic->schema.nCols;
if (topicEp.schema.nCols) { if (topicEp.schema.nCols) {
topicEp.schema.pSchema = taosMemoryCalloc(topicEp.schema.nCols, sizeof(SSchema)); topicEp.schema.pSchema = taosMemoryCalloc(topicEp.schema.nCols, sizeof(SSchema));
memcpy(topicEp.schema.pSchema, pTopic->schema.pSchema, topicEp.schema.nCols * sizeof(SSchema)); if (topicEp.schema.pSchema == NULL) {
taosRUnLockLatch(&pTopic->lock);
taosRUnLockLatch(&pSub->lock);
mndReleaseSubscribe(pMnode, pSub);
mndReleaseTopic(pMnode, pTopic);
return TSDB_CODE_OUT_OF_MEMORY;
}
(void)memcpy(topicEp.schema.pSchema, pTopic->schema.pSchema, topicEp.schema.nCols * sizeof(SSchema));
} }
taosRUnLockLatch(&pTopic->lock); taosRUnLockLatch(&pTopic->lock);
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
// 2.2 iterate all vg assigned to the consumer of that topic // 2.2 iterate all vg assigned to the consumer of that topic
SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t)); SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));
int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs); if (pConsumerEp == NULL) {
taosRUnLockLatch(&pConsumer->lock);
// this customer assigned vgroups taosRUnLockLatch(&pSub->lock);
mndReleaseSubscribe(pMnode, pSub);
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs);
topicEp.vgs = taosArrayInit(vgNum, sizeof(SMqSubVgEp)); topicEp.vgs = taosArrayInit(vgNum, sizeof(SMqSubVgEp));
if (topicEp.vgs == NULL) { if (topicEp.vgs == NULL) {
taosRUnLockLatch(&pConsumer->lock); taosRUnLockLatch(&pConsumer->lock);
@ -391,9 +323,9 @@ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t e
for (int32_t j = 0; j < vgNum; j++) { for (int32_t j = 0; j < vgNum; j++) {
SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j); SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
// char offsetKey[TSDB_PARTITION_KEY_LEN]; if (pVgEp == NULL) {
// mndMakePartitionKey(offsetKey, pConsumer->cgroup, topic, pVgEp->vgId); continue;
}
if (epoch == -1) { if (epoch == -1) {
SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId); SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId);
if (pVgroup) { if (pVgroup) {
@ -401,17 +333,10 @@ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t e
mndReleaseVgroup(pMnode, pVgroup); mndReleaseVgroup(pMnode, pVgroup);
} }
} }
// 2.2.1 build vg ep SMqSubVgEp vgEp = {.epSet = pVgEp->epSet, .vgId = pVgEp->vgId, .offset = -1};
SMqSubVgEp vgEp = { (void)taosArrayPush(topicEp.vgs, &vgEp);
.epSet = pVgEp->epSet,
.vgId = pVgEp->vgId,
.offset = -1,
};
taosArrayPush(topicEp.vgs, &vgEp);
} }
taosArrayPush(rsp->topics, &topicEp); (void)taosArrayPush(rsp->topics, &topicEp);
taosRUnLockLatch(&pSub->lock); taosRUnLockLatch(&pSub->lock);
mndReleaseSubscribe(pMnode, pSub); mndReleaseSubscribe(pMnode, pSub);
} }
@ -437,12 +362,15 @@ static int32_t buildAskEpRsp(SRpcMsg *pMsg, SMqAskEpRsp *rsp, int32_t serverEpoc
pHead->walever = 0; pHead->walever = 0;
void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead)); void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
tEncodeSMqAskEpRsp(&abuf, rsp); if (tEncodeSMqAskEpRsp(&abuf, rsp) < 0) {
rpcFreeCont(buf);
return TSDB_CODE_TSC_INTERNAL_ERROR;
}
// send rsp // send rsp
pMsg->info.rsp = buf; pMsg->info.rsp = buf;
pMsg->info.rspLen = tlen; pMsg->info.rspLen = tlen;
TAOS_RETURN(code); return code;
} }
static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) { static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
@ -450,33 +378,22 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
SMqAskEpReq req = {0}; SMqAskEpReq req = {0};
SMqAskEpRsp rsp = {0}; SMqAskEpRsp rsp = {0};
int32_t code = 0; int32_t code = 0;
SMqConsumerObj *pConsumer = NULL;
TAOS_CHECK_RETURN(tDeserializeSMqAskEpReq(pMsg->pCont, pMsg->contLen, &req)); MND_TMQ_RETURN_CHECK(tDeserializeSMqAskEpReq(pMsg->pCont, pMsg->contLen, &req));
int64_t consumerId = req.consumerId; int64_t consumerId = req.consumerId;
MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));
SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
if (pConsumer == NULL) {
mError("consumer:0x%" PRIx64 " group:%s not exists in sdb", consumerId, req.cgroup);
return TSDB_CODE_MND_CONSUMER_NOT_EXIST;
}
if (strncmp(req.cgroup, pConsumer->cgroup, tListLen(pConsumer->cgroup)) != 0) { if (strncmp(req.cgroup, pConsumer->cgroup, tListLen(pConsumer->cgroup)) != 0) {
mError("consumer:0x%" PRIx64 " group:%s not consistent with data in sdb, saved cgroup:%s", consumerId, req.cgroup, mError("consumer:0x%" PRIx64 " group:%s not consistent with data in sdb, saved cgroup:%s", consumerId, req.cgroup,
pConsumer->cgroup); pConsumer->cgroup);
code = TSDB_CODE_MND_CONSUMER_NOT_EXIST; code = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
goto END; goto END;
} }
atomic_store_32(&pConsumer->hbStatus, 0); atomic_store_32(&pConsumer->hbStatus, 0);
// 1. check consumer status
int32_t status = atomic_load_32(&pConsumer->status); int32_t status = atomic_load_32(&pConsumer->status);
if (status == MQ_CONSUMER_STATUS_LOST) { if (status == MQ_CONSUMER_STATUS_LOST) {
mndSendConsumerMsg(pMnode, pConsumer->consumerId, TDMT_MND_TMQ_CONSUMER_RECOVER, &pMsg->info); MND_TMQ_RETURN_CHECK(mndSendConsumerMsg(pMnode, pConsumer->consumerId, TDMT_MND_TMQ_CONSUMER_RECOVER, &pMsg->info));
} }
if (status != MQ_CONSUMER_STATUS_READY) { if (status != MQ_CONSUMER_STATUS_READY) {
mInfo("consumer:0x%" PRIx64 " not ready, status: %s", consumerId, mndConsumerStatusName(status)); mInfo("consumer:0x%" PRIx64 " not ready, status: %s", consumerId, mndConsumerStatusName(status));
code = TSDB_CODE_MND_CONSUMER_NOT_READY; code = TSDB_CODE_MND_CONSUMER_NOT_READY;
@ -490,10 +407,7 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
if (epoch != serverEpoch) { if (epoch != serverEpoch) {
mInfo("process ask ep, consumer:0x%" PRIx64 "(epoch %d) update with server epoch %d", mInfo("process ask ep, consumer:0x%" PRIx64 "(epoch %d) update with server epoch %d",
consumerId, epoch, serverEpoch); consumerId, epoch, serverEpoch);
code = addEpSetInfo(pMnode, pConsumer, epoch, &rsp); MND_TMQ_RETURN_CHECK(addEpSetInfo(pMnode, pConsumer, epoch, &rsp));
if(code != 0){
goto END;
}
} }
code = buildAskEpRsp(pMsg, &rsp, serverEpoch, consumerId); code = buildAskEpRsp(pMsg, &rsp, serverEpoch, consumerId);
@ -507,27 +421,29 @@ END:
int32_t mndSetConsumerDropLogs(STrans *pTrans, SMqConsumerObj *pConsumer) { int32_t mndSetConsumerDropLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {
int32_t code = 0; int32_t code = 0;
SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer); SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
if (pCommitRaw == NULL) { MND_TMQ_NULL_CHECK(pCommitRaw);
code = TSDB_CODE_MND_RETURN_VALUE_NULL; code = mndTransAppendCommitlog(pTrans, pCommitRaw);
if (terrno != 0) code = terrno; if (code != 0) {
TAOS_RETURN(code); sdbFreeRaw(pCommitRaw);
goto END;
} }
TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw)); MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED)); END:
TAOS_RETURN(code); return code;
} }
int32_t mndSetConsumerCommitLogs(STrans *pTrans, SMqConsumerObj *pConsumer) { int32_t mndSetConsumerCommitLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {
int32_t code = 0; int32_t code = 0;
SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer); SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
if (pCommitRaw == NULL) { MND_TMQ_NULL_CHECK(pCommitRaw);
code = TSDB_CODE_MND_RETURN_VALUE_NULL; code = mndTransAppendCommitlog(pTrans, pCommitRaw);
if (terrno != 0) code = terrno; if (code != 0) {
TAOS_RETURN(code); sdbFreeRaw(pCommitRaw);
goto END;
} }
TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw)); MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY)); END:
TAOS_RETURN(code); return code;
} }
static void freeItem(void *param) { static void freeItem(void *param) {
@ -538,29 +454,35 @@ static void freeItem(void *param) {
} }
static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerObj *pConsumerNew){ static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerObj *pConsumerNew){
int32_t code = 0;
pConsumerNew->rebNewTopics = taosArrayInit(0, sizeof(void *)); pConsumerNew->rebNewTopics = taosArrayInit(0, sizeof(void *));
MND_TMQ_NULL_CHECK(pConsumerNew->rebNewTopics);
pConsumerNew->rebRemovedTopics = taosArrayInit(0, sizeof(void *)); pConsumerNew->rebRemovedTopics = taosArrayInit(0, sizeof(void *));
if(pConsumerNew->rebNewTopics == NULL || pConsumerNew->rebRemovedTopics == NULL){ MND_TMQ_NULL_CHECK(pConsumerNew->rebRemovedTopics);
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t newTopicNum = taosArrayGetSize(pConsumerNew->assignedTopics); int32_t newTopicNum = taosArrayGetSize(pConsumerNew->assignedTopics);
int32_t oldTopicNum = taosArrayGetSize(pExistedConsumer->currentTopics); int32_t oldTopicNum = taosArrayGetSize(pExistedConsumer->currentTopics);
int32_t i = 0, j = 0; int32_t i = 0, j = 0;
while (i < oldTopicNum || j < newTopicNum) { while (i < oldTopicNum || j < newTopicNum) {
if (i >= oldTopicNum) { if (i >= oldTopicNum) {
char *newTopicCopy = taosStrdup(taosArrayGetP(pConsumerNew->assignedTopics, j)); void* tmp = taosArrayGetP(pConsumerNew->assignedTopics, j);
taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy); MND_TMQ_NULL_CHECK(tmp);
char *newTopicCopy = taosStrdup(tmp);
MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy));
j++; j++;
continue; continue;
} else if (j >= newTopicNum) { } else if (j >= newTopicNum) {
char *oldTopicCopy = taosStrdup(taosArrayGetP(pExistedConsumer->currentTopics, i)); void* tmp = taosArrayGetP(pExistedConsumer->currentTopics, i);
taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy); MND_TMQ_NULL_CHECK(tmp);
char *oldTopicCopy = taosStrdup(tmp);
MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy));
i++; i++;
continue; continue;
} else { } else {
char *oldTopic = taosArrayGetP(pExistedConsumer->currentTopics, i); char *oldTopic = taosArrayGetP(pExistedConsumer->currentTopics, i);
MND_TMQ_NULL_CHECK(oldTopic);
char *newTopic = taosArrayGetP(pConsumerNew->assignedTopics, j); char *newTopic = taosArrayGetP(pConsumerNew->assignedTopics, j);
MND_TMQ_NULL_CHECK(newTopic);
int comp = strcmp(oldTopic, newTopic); int comp = strcmp(oldTopic, newTopic);
if (comp == 0) { if (comp == 0) {
i++; i++;
@ -568,22 +490,22 @@ static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerOb
continue; continue;
} else if (comp < 0) { } else if (comp < 0) {
char *oldTopicCopy = taosStrdup(oldTopic); char *oldTopicCopy = taosStrdup(oldTopic);
taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy); MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy));
i++; i++;
continue; continue;
} else { } else {
char *newTopicCopy = taosStrdup(newTopic); char *newTopicCopy = taosStrdup(newTopic);
taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy); MND_TMQ_NULL_CHECK(taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy));
j++; j++;
continue; continue;
} }
} }
} }
return 0; END:
return code;
} }
static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){ static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){
int32_t code = 0;
taosArraySort(pTopicList, taosArrayCompareString); taosArraySort(pTopicList, taosArrayCompareString);
taosArrayRemoveDuplicate(pTopicList, taosArrayCompareString, freeItem); taosArrayRemoveDuplicate(pTopicList, taosArrayCompareString, freeItem);
@ -594,24 +516,21 @@ static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){
return TSDB_CODE_TMQ_GROUP_OUT_OF_RANGE; return TSDB_CODE_TMQ_GROUP_OUT_OF_RANGE;
} }
} }
TAOS_RETURN(code); return 0;
} }
static SMqConsumerObj* buildSubConsumer(SMnode *pMnode, SCMSubscribeReq *subscribe){ static int32_t buildSubConsumer(SMnode *pMnode, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer){
int64_t consumerId = subscribe->consumerId; int64_t consumerId = subscribe->consumerId;
char *cgroup = subscribe->cgroup; char *cgroup = subscribe->cgroup;
SMqConsumerObj *pConsumerNew = NULL; SMqConsumerObj *pConsumerNew = NULL;
SMqConsumerObj *pExistedConsumer = NULL; SMqConsumerObj *pExistedConsumer = NULL;
pExistedConsumer = mndAcquireConsumer(pMnode, consumerId); int32_t code = mndAcquireConsumer(pMnode, consumerId, &pExistedConsumer);
if (pExistedConsumer == NULL) { if (code != 0) {
mInfo("receive subscribe request from new consumer:0x%" PRIx64 mInfo("receive subscribe request from new consumer:0x%" PRIx64
",cgroup:%s, numOfTopics:%d", consumerId, ",cgroup:%s, numOfTopics:%d", consumerId,
subscribe->cgroup, (int32_t)taosArrayGetSize(subscribe->topicNames)); subscribe->cgroup, (int32_t)taosArrayGetSize(subscribe->topicNames));
pConsumerNew = tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_INSERT_SUB, NULL, subscribe); MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_INSERT_SUB, NULL, subscribe, &pConsumerNew));
if (pConsumerNew == NULL) {
goto _over;
}
} else { } else {
int32_t status = atomic_load_32(&pExistedConsumer->status); int32_t status = atomic_load_32(&pExistedConsumer->status);
@ -621,82 +540,53 @@ static SMqConsumerObj* buildSubConsumer(SMnode *pMnode, SCMSubscribeReq *subscri
(int32_t)taosArrayGetSize(subscribe->topicNames)); (int32_t)taosArrayGetSize(subscribe->topicNames));
if (status != MQ_CONSUMER_STATUS_READY) { if (status != MQ_CONSUMER_STATUS_READY) {
terrno = TSDB_CODE_MND_CONSUMER_NOT_READY; code = TSDB_CODE_MND_CONSUMER_NOT_READY;
goto _over; goto END;
}
pConsumerNew = tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_UPDATE_SUB, NULL, subscribe);
if (pConsumerNew == NULL) {
goto _over;
}
int32_t code = getTopicAddDelete(pExistedConsumer, pConsumerNew);
if (code != 0){
terrno = code;
goto _over;
} }
MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_UPDATE_SUB, NULL, subscribe, &pConsumerNew));
MND_TMQ_RETURN_CHECK(getTopicAddDelete(pExistedConsumer, pConsumerNew));
} }
mndReleaseConsumer(pMnode, pExistedConsumer); mndReleaseConsumer(pMnode, pExistedConsumer);
return pConsumerNew; if (ppConsumer){
*ppConsumer = pConsumerNew;
}
return code;
_over: END:
mndReleaseConsumer(pMnode, pExistedConsumer); mndReleaseConsumer(pMnode, pExistedConsumer);
tDeleteSMqConsumerObj(pConsumerNew); tDeleteSMqConsumerObj(pConsumerNew);
return NULL; return code;
} }
int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) { int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
SMnode *pMnode = pMsg->info.node; SMnode *pMnode = pMsg->info.node;
char *msgStr = pMsg->pCont; char *msgStr = pMsg->pCont;
int32_t code = 0; int32_t code = 0;
SCMSubscribeReq subscribe = {0};
tDeserializeSCMSubscribeReq(msgStr, &subscribe);
SMqConsumerObj *pConsumerNew = NULL; SMqConsumerObj *pConsumerNew = NULL;
STrans *pTrans = NULL; STrans *pTrans = NULL;
SCMSubscribeReq subscribe = {0};
MND_TMQ_RETURN_CHECK(tDeserializeSCMSubscribeReq(msgStr, &subscribe));
if(taosArrayGetSize(subscribe.topicNames) == 0){ if(taosArrayGetSize(subscribe.topicNames) == 0){
SMqConsumerObj *pConsumerTmp = mndAcquireConsumer(pMnode, subscribe.consumerId); SMqConsumerObj *pConsumerTmp = NULL;
if(pConsumerTmp == NULL){ MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, subscribe.consumerId, &pConsumerTmp));
goto _over;
}
mndReleaseConsumer(pMnode, pConsumerTmp); mndReleaseConsumer(pMnode, pConsumerTmp);
} }
MND_TMQ_RETURN_CHECK(checkAndSortTopic(pMnode, subscribe.topicNames));
code = checkAndSortTopic(pMnode, subscribe.topicNames);
if(code != TSDB_CODE_SUCCESS){
goto _over;
}
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "subscribe"); pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "subscribe");
if (pTrans == NULL) { MND_TMQ_NULL_CHECK(pTrans);
code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno;
goto _over;
}
code = validateTopics(subscribe.topicNames, pMnode, pMsg->info.conn.user, subscribe.enableReplay); MND_TMQ_RETURN_CHECK(validateTopics(subscribe.topicNames, pMnode, pMsg->info.conn.user, subscribe.enableReplay));
if (code != TSDB_CODE_SUCCESS) { MND_TMQ_RETURN_CHECK(buildSubConsumer(pMnode, &subscribe, &pConsumerNew));
goto _over; MND_TMQ_RETURN_CHECK(mndSetConsumerCommitLogs(pTrans, pConsumerNew));
} MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));
pConsumerNew = buildSubConsumer(pMnode, &subscribe);
if(pConsumerNew == NULL){
code = TSDB_CODE_MND_RETURN_VALUE_NULL;
if (terrno != 0) code = terrno;
goto _over;
}
TAOS_CHECK_GOTO(mndSetConsumerCommitLogs(pTrans, pConsumerNew), NULL, _over);
TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _over);
code = TSDB_CODE_ACTION_IN_PROGRESS; code = TSDB_CODE_ACTION_IN_PROGRESS;
_over: END:
mndTransDrop(pTrans); mndTransDrop(pTrans);
tDeleteSMqConsumerObj(pConsumerNew); tDeleteSMqConsumerObj(pConsumerNew);
taosArrayDestroyP(subscribe.topicNames, (FDelete)taosMemoryFree); taosArrayDestroyP(subscribe.topicNames, (FDelete)taosMemoryFree);
TAOS_RETURN(code); return code;
} }
SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) { SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {
@ -713,7 +603,9 @@ SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {
if (buf == NULL) goto CM_ENCODE_OVER; if (buf == NULL) goto CM_ENCODE_OVER;
void *abuf = buf; void *abuf = buf;
tEncodeSMqConsumerObj(&abuf, pConsumer); if(tEncodeSMqConsumerObj(&abuf, pConsumer) < 0){
goto CM_ENCODE_OVER;
}
int32_t dataPos = 0; int32_t dataPos = 0;
SDB_SET_INT32(pRaw, dataPos, tlen, CM_ENCODE_OVER); SDB_SET_INT32(pRaw, dataPos, tlen, CM_ENCODE_OVER);
@ -838,8 +730,7 @@ static bool existInCurrentTopicList(const SMqConsumerObj *pConsumer, const char
int32_t size = taosArrayGetSize(pConsumer->currentTopics); int32_t size = taosArrayGetSize(pConsumer->currentTopics);
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
char *topic = taosArrayGetP(pConsumer->currentTopics, i); char *topic = taosArrayGetP(pConsumer->currentTopics, i);
if (topic && strcmp(topic, pTopic) == 0) {
if (strcmp(topic, pTopic) == 0) {
existing = true; existing = true;
break; break;
} }
@ -865,32 +756,45 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
} else if (pNewConsumer->updateType == CONSUMER_UPDATE_REC) { } else if (pNewConsumer->updateType == CONSUMER_UPDATE_REC) {
int32_t sz = taosArrayGetSize(pOldConsumer->assignedTopics); int32_t sz = taosArrayGetSize(pOldConsumer->assignedTopics);
for (int32_t i = 0; i < sz; i++) { for (int32_t i = 0; i < sz; i++) {
char *topic = taosStrdup(taosArrayGetP(pOldConsumer->assignedTopics, i)); void * tmp = taosArrayGetP(pOldConsumer->assignedTopics, i);
taosArrayPush(pOldConsumer->rebNewTopics, &topic); if (tmp == NULL){
return TSDB_CODE_TMQ_INVALID_MSG;
}
char *topic = taosStrdup(tmp);
if (taosArrayPush(pOldConsumer->rebNewTopics, &topic) == NULL) {
taosMemoryFree(topic);
return TSDB_CODE_TMQ_INVALID_MSG;
}
} }
pOldConsumer->status = MQ_CONSUMER_STATUS_REBALANCE; pOldConsumer->status = MQ_CONSUMER_STATUS_REBALANCE;
mInfo("consumer:0x%" PRIx64 " recover update", pOldConsumer->consumerId); mInfo("consumer:0x%" PRIx64 " recover update", pOldConsumer->consumerId);
} else if (pNewConsumer->updateType == CONSUMER_UPDATE_REB) { } else if (pNewConsumer->updateType == CONSUMER_UPDATE_REB) {
atomic_add_fetch_32(&pOldConsumer->epoch, 1); (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);
pOldConsumer->rebalanceTime = taosGetTimestampMs(); pOldConsumer->rebalanceTime = taosGetTimestampMs();
mInfo("[rebalance] consumer:0x%" PRIx64 " rebalance update, only rebalance time", pOldConsumer->consumerId); mInfo("[rebalance] consumer:0x%" PRIx64 " rebalance update, only rebalance time", pOldConsumer->consumerId);
} else if (pNewConsumer->updateType == CONSUMER_ADD_REB) { } else if (pNewConsumer->updateType == CONSUMER_ADD_REB) {
char *pNewTopic = taosStrdup(taosArrayGetP(pNewConsumer->rebNewTopics, 0)); void *tmp = taosArrayGetP(pNewConsumer->rebNewTopics, 0);
if (tmp == NULL){
return TSDB_CODE_TMQ_INVALID_MSG;
}
char *pNewTopic = taosStrdup(tmp);
removeFromTopicList(pOldConsumer->rebNewTopics, pNewTopic, pOldConsumer->consumerId, "new"); removeFromTopicList(pOldConsumer->rebNewTopics, pNewTopic, pOldConsumer->consumerId, "new");
bool existing = existInCurrentTopicList(pOldConsumer, pNewTopic); bool existing = existInCurrentTopicList(pOldConsumer, pNewTopic);
if (existing) { if (existing) {
mError("[rebalance] consumer:0x%" PRIx64 " add new topic:%s should not in currentTopics", pOldConsumer->consumerId, pNewTopic); mError("[rebalance] consumer:0x%" PRIx64 " add new topic:%s should not in currentTopics", pOldConsumer->consumerId, pNewTopic);
taosMemoryFree(pNewTopic); taosMemoryFree(pNewTopic);
} else { } else {
taosArrayPush(pOldConsumer->currentTopics, &pNewTopic); if (taosArrayPush(pOldConsumer->currentTopics, &pNewTopic) == NULL) {
taosMemoryFree(pNewTopic);
return TSDB_CODE_TMQ_INVALID_MSG;
}
taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString); taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString);
} }
int32_t status = pOldConsumer->status; int32_t status = pOldConsumer->status;
updateConsumerStatus(pOldConsumer); updateConsumerStatus(pOldConsumer);
pOldConsumer->rebalanceTime = taosGetTimestampMs(); pOldConsumer->rebalanceTime = taosGetTimestampMs();
atomic_add_fetch_32(&pOldConsumer->epoch, 1); (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);
mInfo("[rebalance] consumer:0x%" PRIx64 " rebalance update add, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64 mInfo("[rebalance] consumer:0x%" PRIx64 " rebalance update add, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64
", current topics:%d, newTopics:%d, removeTopics:%d", ", current topics:%d, newTopics:%d, removeTopics:%d",
@ -901,13 +805,16 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
} else if (pNewConsumer->updateType == CONSUMER_REMOVE_REB) { } else if (pNewConsumer->updateType == CONSUMER_REMOVE_REB) {
char *topic = taosArrayGetP(pNewConsumer->rebRemovedTopics, 0); char *topic = taosArrayGetP(pNewConsumer->rebRemovedTopics, 0);
if (topic == NULL){
return TSDB_CODE_TMQ_INVALID_MSG;
}
removeFromTopicList(pOldConsumer->rebRemovedTopics, topic, pOldConsumer->consumerId, "remove"); removeFromTopicList(pOldConsumer->rebRemovedTopics, topic, pOldConsumer->consumerId, "remove");
removeFromTopicList(pOldConsumer->currentTopics, topic, pOldConsumer->consumerId, "current"); removeFromTopicList(pOldConsumer->currentTopics, topic, pOldConsumer->consumerId, "current");
int32_t status = pOldConsumer->status; int32_t status = pOldConsumer->status;
updateConsumerStatus(pOldConsumer); updateConsumerStatus(pOldConsumer);
pOldConsumer->rebalanceTime = taosGetTimestampMs(); pOldConsumer->rebalanceTime = taosGetTimestampMs();
atomic_add_fetch_32(&pOldConsumer->epoch, 1); (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);
mInfo("[rebalance]consumer:0x%" PRIx64 " rebalance update remove, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64 mInfo("[rebalance]consumer:0x%" PRIx64 " rebalance update remove, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64
", current topics:%d, newTopics:%d, removeTopics:%d", ", current topics:%d, newTopics:%d, removeTopics:%d",
@ -921,13 +828,13 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
return 0; return 0;
} }
SMqConsumerObj *mndAcquireConsumer(SMnode *pMnode, int64_t consumerId) { int32_t mndAcquireConsumer(SMnode *pMnode, int64_t consumerId, SMqConsumerObj** pConsumer) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
SMqConsumerObj *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId); *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);
if (pConsumer == NULL) { if (*pConsumer == NULL) {
terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST; return TSDB_CODE_MND_CONSUMER_NOT_EXIST;
} }
return pConsumer; return 0;
} }
void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) { void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {
@ -940,6 +847,7 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0; int32_t numOfRows = 0;
SMqConsumerObj *pConsumer = NULL; SMqConsumerObj *pConsumer = NULL;
int32_t code = 0;
while (numOfRows < rowsCapacity) { while (numOfRows < rowsCapacity) {
pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer); pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer);
@ -964,34 +872,37 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
} }
if (numOfRows + topicSz > rowsCapacity) { if (numOfRows + topicSz > rowsCapacity) {
blockDataEnsureCapacity(pBlock, numOfRows + topicSz); MND_TMQ_RETURN_CHECK(blockDataEnsureCapacity(pBlock, numOfRows + topicSz));
} }
for (int32_t i = 0; i < topicSz; i++) { for (int32_t i = 0; i < topicSz; i++) {
SColumnInfoData *pColInfo; SColumnInfoData *pColInfo = NULL;
int32_t cols = 0; int32_t cols = 0;
// consumer id // consumer id
char consumerIdHex[32] = {0}; char consumerIdHex[32] = {0};
sprintf(varDataVal(consumerIdHex), "0x%" PRIx64, pConsumer->consumerId); (void)sprintf(varDataVal(consumerIdHex), "0x%" PRIx64, pConsumer->consumerId);
varDataSetLen(consumerIdHex, strlen(varDataVal(consumerIdHex))); varDataSetLen(consumerIdHex, strlen(varDataVal(consumerIdHex)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)consumerIdHex, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)consumerIdHex, false));
// consumer group // consumer group
char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0}; char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(cgroup, pConsumer->cgroup); STR_TO_VARSTR(cgroup, pConsumer->cgroup);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)cgroup, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)cgroup, false));
// client id // client id
char clientId[256 + VARSTR_HEADER_SIZE] = {0}; char clientId[256 + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(clientId, pConsumer->clientId); STR_TO_VARSTR(clientId, pConsumer->clientId);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)clientId, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)clientId, false));
// status // status
char status[20 + VARSTR_HEADER_SIZE] = {0}; char status[20 + VARSTR_HEADER_SIZE] = {0};
@ -999,46 +910,48 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
STR_TO_VARSTR(status, pStatusName); STR_TO_VARSTR(status, pStatusName);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)status, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)status, false));
// one subscribed topic // one subscribed topic
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
MND_TMQ_NULL_CHECK(pColInfo);
if (hasTopic) { if (hasTopic) {
char topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0}; char topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
const char *topicName = mndTopicGetShowName(taosArrayGetP(pConsumer->assignedTopics, i)); mndTopicGetShowName(taosArrayGetP(pConsumer->assignedTopics, i), topic + VARSTR_HEADER_SIZE);
STR_TO_VARSTR(topic, topicName); *(VarDataLenT *)(topic) = strlen(topic + VARSTR_HEADER_SIZE);
colDataSetVal(pColInfo, numOfRows, (const char *)topic, false); MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)topic, false));
} else { } else {
colDataSetVal(pColInfo, numOfRows, NULL, true); MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, NULL, true));
} }
// end point
/*pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);*/
/*colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->ep, true);*/
// up time // up time
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->createTime, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->createTime, false));
// subscribe time // subscribe time
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->subscribeTime, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->subscribeTime, false));
// rebalance time // rebalance time
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->rebalanceTime, pConsumer->rebalanceTime == 0); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->rebalanceTime, pConsumer->rebalanceTime == 0));
char buf[TSDB_OFFSET_LEN] = {0}; char buf[TSDB_OFFSET_LEN] = {0};
STqOffsetVal pVal = {.type = pConsumer->resetOffsetCfg}; STqOffsetVal pVal = {.type = pConsumer->resetOffsetCfg};
tFormatOffset(buf, TSDB_OFFSET_LEN, &pVal); MND_TMQ_RETURN_CHECK(tFormatOffset(buf, TSDB_OFFSET_LEN, &pVal));
char parasStr[64 + TSDB_OFFSET_LEN + VARSTR_HEADER_SIZE] = {0}; char parasStr[64 + TSDB_OFFSET_LEN + VARSTR_HEADER_SIZE] = {0};
sprintf(varDataVal(parasStr), "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName, (void)sprintf(varDataVal(parasStr), "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName,
pConsumer->autoCommit, pConsumer->autoCommitInterval, buf); pConsumer->autoCommit, pConsumer->autoCommitInterval, buf);
varDataSetLen(parasStr, strlen(varDataVal(parasStr))); varDataSetLen(parasStr, strlen(varDataVal(parasStr)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)parasStr, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)parasStr, false));
numOfRows++; numOfRows++;
} }
@ -1051,6 +964,9 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
pShow->numOfRows += numOfRows; pShow->numOfRows += numOfRows;
return numOfRows; return numOfRows;
END:
return code;
} }
static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) { static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {

79
source/dnode/mnode/impl/src/mndDef.c
View File

@ -264,16 +264,17 @@ void *tDecodeSMqVgEp(const void *buf, SMqVgEp *pVgEp, int8_t sver) {
static void *topicNameDup(void *p) { return taosStrdup((char *)p); } static void *topicNameDup(void *p) { return taosStrdup((char *)p); }
SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType, char *topic, SCMSubscribeReq *subscribe) { int32_t tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t updateType,
terrno = 0; char *topic, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer) {
int32_t code = 0;
SMqConsumerObj *pConsumer = taosMemoryCalloc(1, sizeof(SMqConsumerObj)); SMqConsumerObj *pConsumer = taosMemoryCalloc(1, sizeof(SMqConsumerObj));
if (pConsumer == NULL) { if (pConsumer == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
return NULL; goto END;
} }
pConsumer->consumerId = consumerId; pConsumer->consumerId = consumerId;
memcpy(pConsumer->cgroup, cgroup, TSDB_CGROUP_LEN); (void)memcpy(pConsumer->cgroup, cgroup, TSDB_CGROUP_LEN);
pConsumer->epoch = 0; pConsumer->epoch = 0;
pConsumer->status = MQ_CONSUMER_STATUS_REBALANCE; pConsumer->status = MQ_CONSUMER_STATUS_REBALANCE;
@ -286,20 +287,26 @@ SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t upda
if (updateType == CONSUMER_ADD_REB){ if (updateType == CONSUMER_ADD_REB){
pConsumer->rebNewTopics = taosArrayInit(0, sizeof(void *)); pConsumer->rebNewTopics = taosArrayInit(0, sizeof(void *));
if(pConsumer->rebNewTopics == NULL){ if(pConsumer->rebNewTopics == NULL){
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto END; goto END;
} }
char* topicTmp = taosStrdup(topic); char* topicTmp = taosStrdup(topic);
taosArrayPush(pConsumer->rebNewTopics, &topicTmp); if (taosArrayPush(pConsumer->rebNewTopics, &topicTmp) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
}else if (updateType == CONSUMER_REMOVE_REB) { }else if (updateType == CONSUMER_REMOVE_REB) {
pConsumer->rebRemovedTopics = taosArrayInit(0, sizeof(void *)); pConsumer->rebRemovedTopics = taosArrayInit(0, sizeof(void *));
if(pConsumer->rebRemovedTopics == NULL){ if(pConsumer->rebRemovedTopics == NULL){
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto END; goto END;
} }
char* topicTmp = taosStrdup(topic); char* topicTmp = taosStrdup(topic);
taosArrayPush(pConsumer->rebRemovedTopics, &topicTmp); if (taosArrayPush(pConsumer->rebRemovedTopics, &topicTmp) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
}else if (updateType == CONSUMER_INSERT_SUB){ }else if (updateType == CONSUMER_INSERT_SUB){
tstrncpy(pConsumer->clientId, subscribe->clientId, tListLen(pConsumer->clientId)); tstrncpy(pConsumer->clientId, subscribe->clientId, tListLen(pConsumer->clientId));
pConsumer->withTbName = subscribe->withTbName; pConsumer->withTbName = subscribe->withTbName;
@ -307,10 +314,9 @@ SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t upda
pConsumer->autoCommitInterval = subscribe->autoCommitInterval; pConsumer->autoCommitInterval = subscribe->autoCommitInterval;
pConsumer->resetOffsetCfg = subscribe->resetOffsetCfg; pConsumer->resetOffsetCfg = subscribe->resetOffsetCfg;
pConsumer->rebNewTopics = taosArrayDup(subscribe->topicNames, topicNameDup); pConsumer->rebNewTopics = taosArrayDup(subscribe->topicNames, topicNameDup);
if (pConsumer->rebNewTopics == NULL){ if (pConsumer->rebNewTopics == NULL){
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto END; goto END;
} }
pConsumer->assignedTopics = subscribe->topicNames; pConsumer->assignedTopics = subscribe->topicNames;
@ -320,11 +326,12 @@ SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char *cgroup, int8_t upda
subscribe->topicNames = NULL; subscribe->topicNames = NULL;
} }
return pConsumer; *ppConsumer = pConsumer;
return 0;
END: END:
tDeleteSMqConsumerObj(pConsumer); tDeleteSMqConsumerObj(pConsumer);
return NULL; return code;
} }
void tClearSMqConsumerObj(SMqConsumerObj *pConsumer) { void tClearSMqConsumerObj(SMqConsumerObj *pConsumer) {
@ -540,33 +547,36 @@ void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp, int8_t s
return (void *)buf; return (void *)buf;
} }
SMqSubscribeObj *tNewSubscribeObj(const char *key) { int32_t tNewSubscribeObj(const char *key, SMqSubscribeObj **ppSub) {
terrno = 0; int32_t code = 0;
SMqSubscribeObj *pSubObj = taosMemoryCalloc(1, sizeof(SMqSubscribeObj)); SMqSubscribeObj *pSubObj = taosMemoryCalloc(1, sizeof(SMqSubscribeObj));
if (pSubObj == NULL) { MND_TMQ_NULL_CHECK(pSubObj);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
memcpy(pSubObj->key, key, TSDB_SUBSCRIBE_KEY_LEN); (void)memcpy(pSubObj->key, key, TSDB_SUBSCRIBE_KEY_LEN);
taosInitRWLatch(&pSubObj->lock); taosInitRWLatch(&pSubObj->lock);
pSubObj->vgNum = 0; pSubObj->vgNum = 0;
pSubObj->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK); pSubObj->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
MND_TMQ_NULL_CHECK(pSubObj->consumerHash);
// TODO set hash free fp
/*taosHashSetFreeFp(pSubObj->consumerHash, tDeleteSMqConsumerEp);*/
pSubObj->unassignedVgs = taosArrayInit(0, POINTER_BYTES); pSubObj->unassignedVgs = taosArrayInit(0, POINTER_BYTES);
return pSubObj; MND_TMQ_NULL_CHECK(pSubObj->unassignedVgs);
if (ppSub){
*ppSub = pSubObj;
}
return code;
END:
taosMemoryFree(pSubObj);
return code;
} }
SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) { int32_t tCloneSubscribeObj(const SMqSubscribeObj *pSub, SMqSubscribeObj **ppSub) {
terrno = 0; int32_t code = 0;
SMqSubscribeObj *pSubNew = taosMemoryMalloc(sizeof(SMqSubscribeObj)); SMqSubscribeObj *pSubNew = taosMemoryMalloc(sizeof(SMqSubscribeObj));
if (pSubNew == NULL) { if (pSubNew == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
return NULL; goto END;
} }
memcpy(pSubNew->key, pSub->key, TSDB_SUBSCRIBE_KEY_LEN); (void)memcpy(pSubNew->key, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
taosInitRWLatch(&pSubNew->lock); taosInitRWLatch(&pSubNew->lock);
pSubNew->dbUid = pSub->dbUid; pSubNew->dbUid = pSub->dbUid;
@ -576,8 +586,7 @@ SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) {
pSubNew->vgNum = pSub->vgNum; pSubNew->vgNum = pSub->vgNum;
pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK); pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
// TODO set hash free fp
/*taosHashSetFreeFp(pSubNew->consumerHash, tDeleteSMqConsumerEp);*/
void *pIter = NULL; void *pIter = NULL;
SMqConsumerEp *pConsumerEp = NULL; SMqConsumerEp *pConsumerEp = NULL;
while (1) { while (1) {
@ -592,9 +601,13 @@ SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) {
} }
pSubNew->unassignedVgs = taosArrayDup(pSub->unassignedVgs, (__array_item_dup_fn_t)tCloneSMqVgEp); pSubNew->unassignedVgs = taosArrayDup(pSub->unassignedVgs, (__array_item_dup_fn_t)tCloneSMqVgEp);
pSubNew->offsetRows = taosArrayDup(pSub->offsetRows, NULL); pSubNew->offsetRows = taosArrayDup(pSub->offsetRows, NULL);
memcpy(pSubNew->dbName, pSub->dbName, TSDB_DB_FNAME_LEN); (void)memcpy(pSubNew->dbName, pSub->dbName, TSDB_DB_FNAME_LEN);
pSubNew->qmsg = taosStrdup(pSub->qmsg); pSubNew->qmsg = taosStrdup(pSub->qmsg);
return pSubNew; if (ppSub) {
*ppSub = pSubNew;
}
END:
return code;
} }
void tDeleteSubscribeObj(SMqSubscribeObj *pSub) { void tDeleteSubscribeObj(SMqSubscribeObj *pSub) {

42
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -771,21 +771,22 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) { if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {
pPlan = qStringToQueryPlan(pTopic->physicalPlan); pPlan = qStringToQueryPlan(pTopic->physicalPlan);
if (pPlan == NULL) { if (pPlan == NULL) {
code = TSDB_CODE_QRY_INVALID_INPUT; return TSDB_CODE_QRY_INVALID_INPUT;
TAOS_RETURN(code);
} }
} else if (pTopic->subType == TOPIC_SUB_TYPE__TABLE && pTopic->ast != NULL) { } else if (pTopic->subType == TOPIC_SUB_TYPE__TABLE && pTopic->ast != NULL) {
SNode* pAst = NULL; SNode* pAst = NULL;
if ((code = nodesStringToNode(pTopic->ast, &pAst)) != 0) { code = nodesStringToNode(pTopic->ast, &pAst);
if (code != 0) {
mError("topic:%s, failed to create since %s", pTopic->name, terrstr()); mError("topic:%s, failed to create since %s", pTopic->name, terrstr());
TAOS_RETURN(code); return code;
} }
SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true}; SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};
if ((code = qCreateQueryPlan(&cxt, &pPlan, NULL)) != 0) { code = qCreateQueryPlan(&cxt, &pPlan, NULL);
if (code != 0) {
mError("failed to create topic:%s since %s", pTopic->name, terrstr()); mError("failed to create topic:%s since %s", pTopic->name, terrstr());
nodesDestroyNode(pAst); nodesDestroyNode(pAst);
TAOS_RETURN(code); return code;
} }
nodesDestroyNode(pAst); nodesDestroyNode(pAst);
} }
@ -793,18 +794,19 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
if (pPlan) { if (pPlan) {
int32_t levelNum = LIST_LENGTH(pPlan->pSubplans); int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
if (levelNum != 1) { if (levelNum != 1) {
qDestroyQueryPlan(pPlan);
code = TSDB_CODE_MND_INVALID_TOPIC_QUERY; code = TSDB_CODE_MND_INVALID_TOPIC_QUERY;
TAOS_RETURN(code); goto END;
} }
SNodeListNode* pNodeListNode = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0); SNodeListNode* pNodeListNode = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
if (pNodeListNode == NULL){
code = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
int32_t opNum = LIST_LENGTH(pNodeListNode->pNodeList); int32_t opNum = LIST_LENGTH(pNodeListNode->pNodeList);
if (opNum != 1) { if (opNum != 1) {
qDestroyQueryPlan(pPlan);
code = TSDB_CODE_MND_INVALID_TOPIC_QUERY; code = TSDB_CODE_MND_INVALID_TOPIC_QUERY;
TAOS_RETURN(code); goto END;
} }
pSubplan = (SSubplan*)nodesListGetNode(pNodeListNode->pNodeList, 0); pSubplan = (SSubplan*)nodesListGetNode(pNodeListNode->pNodeList, 0);
@ -825,12 +827,18 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
pSub->vgNum++; pSub->vgNum++;
SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp)); SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
if (pVgEp == NULL){
code = TSDB_CODE_OUT_OF_MEMORY;
goto END;
}
pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup); pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);
pVgEp->vgId = pVgroup->vgId; pVgEp->vgId = pVgroup->vgId;
taosArrayPush(pSub->unassignedVgs, &pVgEp); if (taosArrayPush(pSub->unassignedVgs, &pVgEp) == NULL){
code = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pVgEp);
goto END;
}
mInfo("init subscription %s for topic:%s assign vgId:%d", pSub->key, pTopic->name, pVgEp->vgId); mInfo("init subscription %s for topic:%s assign vgId:%d", pSub->key, pTopic->name, pVgEp->vgId);
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
} }
@ -838,14 +846,14 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
int32_t msgLen; int32_t msgLen;
if (qSubPlanToString(pSubplan, &pSub->qmsg, &msgLen) < 0) { if (qSubPlanToString(pSubplan, &pSub->qmsg, &msgLen) < 0) {
qDestroyQueryPlan(pPlan);
code = TSDB_CODE_QRY_INVALID_INPUT; code = TSDB_CODE_QRY_INVALID_INPUT;
TAOS_RETURN(code); goto END;
} }
} else { } else {
pSub->qmsg = taosStrdup(""); pSub->qmsg = taosStrdup("");
} }
END:
qDestroyQueryPlan(pPlan); qDestroyQueryPlan(pPlan);
TAOS_RETURN(code); return code;
} }

772
source/dnode/mnode/impl/src/mndSubscribe.c
View File

File diff suppressed because it is too large Load Diff

468
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -44,8 +44,6 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq);
static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows); static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter); static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter);
static int32_t mndSetDropTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic);
int32_t mndInitTopic(SMnode *pMnode) { int32_t mndInitTopic(SMnode *pMnode) {
SSdbTable table = { SSdbTable table = {
.sdbType = SDB_TOPIC, .sdbType = SDB_TOPIC,
@ -70,9 +68,16 @@ int32_t mndInitTopic(SMnode *pMnode) {
void mndCleanupTopic(SMnode *pMnode) {} void mndCleanupTopic(SMnode *pMnode) {}
const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]) { void mndTopicGetShowName(const char* fullTopic, char* topic) {
// if (fullTopic == NULL) {
return strchr(topic, '.') + 1; return;
}
char* tmp = strchr(fullTopic, '.');
if (tmp == NULL) {
tstrncpy(topic, fullTopic, TSDB_TOPIC_FNAME_LEN);
}else {
tstrncpy(topic, tmp+1, TSDB_TOPIC_FNAME_LEN);
}
} }
SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
@ -129,7 +134,9 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
goto TOPIC_ENCODE_OVER; goto TOPIC_ENCODE_OVER;
} }
void *aswBuf = swBuf; void *aswBuf = swBuf;
taosEncodeSSchemaWrapper(&aswBuf, &pTopic->schema); if(taosEncodeSSchemaWrapper(&aswBuf, &pTopic->schema) < 0){
goto TOPIC_ENCODE_OVER;
}
SDB_SET_BINARY(pRaw, dataPos, swBuf, schemaLen, TOPIC_ENCODE_OVER); SDB_SET_BINARY(pRaw, dataPos, swBuf, schemaLen, TOPIC_ENCODE_OVER);
} }
@ -259,7 +266,7 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
} }
int16_t colId; int16_t colId;
SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER); SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER);
taosArrayPush(pTopic->ntbColIds, &colId); (void)taosArrayPush(pTopic->ntbColIds, &colId);
} }
SDB_GET_INT64(pRaw, dataPos, &pTopic->ctbStbUid, TOPIC_DECODE_OVER); SDB_GET_INT64(pRaw, dataPos, &pTopic->ctbStbUid, TOPIC_DECODE_OVER);
@ -289,25 +296,25 @@ static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic) {
taosMemoryFreeClear(pTopic->ast); taosMemoryFreeClear(pTopic->ast);
taosMemoryFreeClear(pTopic->physicalPlan); taosMemoryFreeClear(pTopic->physicalPlan);
if (pTopic->schema.nCols) taosMemoryFreeClear(pTopic->schema.pSchema); if (pTopic->schema.nCols) taosMemoryFreeClear(pTopic->schema.pSchema);
taosArrayDestroy(pTopic->ntbColIds); (void)taosArrayDestroy(pTopic->ntbColIds);
return 0; return 0;
} }
static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic) { static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic) {
mTrace("topic:%s perform update action", pOldTopic->name); mTrace("topic:%s perform update action", pOldTopic->name);
atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime); (void)atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime);
atomic_exchange_32(&pOldTopic->version, pNewTopic->version); (void)atomic_exchange_32(&pOldTopic->version, pNewTopic->version);
return 0; return 0;
} }
SMqTopicObj *mndAcquireTopic(SMnode *pMnode, const char *topicName) { int32_t mndAcquireTopic(SMnode *pMnode, const char *topicName, SMqTopicObj **pTopic) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
SMqTopicObj *pTopic = sdbAcquire(pSdb, SDB_TOPIC, topicName); *pTopic = sdbAcquire(pSdb, SDB_TOPIC, topicName);
if (pTopic == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) { if (*pTopic == NULL) {
terrno = TSDB_CODE_MND_TOPIC_NOT_EXIST; return TSDB_CODE_MND_TOPIC_NOT_EXIST;
} }
return pTopic; return TDB_CODE_SUCCESS;
} }
void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) { void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {
@ -316,25 +323,23 @@ void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {
} }
static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) { static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
terrno = TSDB_CODE_MND_INVALID_TOPIC; if (pCreate->sql == NULL) return TSDB_CODE_MND_INVALID_TOPIC;
if (pCreate->sql == NULL) return -1;
if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) { if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
if (pCreate->ast == NULL || pCreate->ast[0] == 0) return -1; if (pCreate->ast == NULL || pCreate->ast[0] == 0) return TSDB_CODE_MND_INVALID_TOPIC;
} else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) { } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {
if (pCreate->subStbName[0] == 0) return -1; if (pCreate->subStbName[0] == 0) return TSDB_CODE_MND_INVALID_TOPIC;
} else if (pCreate->subType == TOPIC_SUB_TYPE__DB) { } else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {
if (pCreate->subDbName[0] == 0) return -1; if (pCreate->subDbName[0] == 0) return TSDB_CODE_MND_INVALID_TOPIC;
} }
terrno = TSDB_CODE_SUCCESS;
return 0; return 0;
} }
static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) { static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) {
SNodeList *pNodeList = NULL; SNodeList *pNodeList = NULL;
nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList); int32_t code = 0;
MND_TMQ_RETURN_CHECK(nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList));
int64_t suid = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->suid; int64_t suid = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->suid;
int8_t tableType = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->tableType; int8_t tableType = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->tableType;
if (tableType == TSDB_CHILD_TABLE) { if (tableType == TSDB_CHILD_TABLE) {
@ -345,17 +350,19 @@ static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) {
SColumnNode *pCol = (SColumnNode *)pNode; SColumnNode *pCol = (SColumnNode *)pNode;
if (pCol->tableType == TSDB_NORMAL_TABLE) { if (pCol->tableType == TSDB_NORMAL_TABLE) {
pTopic->ntbUid = pCol->tableId; pTopic->ntbUid = pCol->tableId;
taosArrayPush(pTopic->ntbColIds, &pCol->colId); MND_TMQ_NULL_CHECK(taosArrayPush(pTopic->ntbColIds, &pCol->colId));
} }
} }
} }
nodesDestroyList(pNodeList); nodesDestroyList(pNodeList);
return 0;
END:
return code;
} }
static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj *topicObj){ static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj *topicObj){
STqCheckInfo info; STqCheckInfo info = {0};
memcpy(info.topic, topicObj->name, TSDB_TOPIC_FNAME_LEN); (void)memcpy(info.topic, topicObj->name, TSDB_TOPIC_FNAME_LEN);
info.ntbUid = topicObj->ntbUid; info.ntbUid = topicObj->ntbUid;
info.colIdList = topicObj->ntbColIds; info.colIdList = topicObj->ntbColIds;
// broadcast forbid alter info // broadcast forbid alter info
@ -363,7 +370,7 @@ static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL; SVgObj *pVgroup = NULL;
int32_t code = 0; int32_t code = 0;
void *buf = NULL; void *buf = NULL;
while (1) { while (1) {
// iterate vg // iterate vg
@ -378,16 +385,17 @@ static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj
int32_t len; int32_t len;
tEncodeSize(tEncodeSTqCheckInfo, &info, len, code); tEncodeSize(tEncodeSTqCheckInfo, &info, len, code);
if (code != 0) { if (code != 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto END; goto END;
} }
buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len); buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
MND_TMQ_NULL_CHECK(buf);
void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead)); void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
SEncoder encoder; SEncoder encoder = {0};
tEncoderInit(&encoder, abuf, len); tEncoderInit(&encoder, abuf, len);
code = tEncodeSTqCheckInfo(&encoder, &info); code = tEncodeSTqCheckInfo(&encoder, &info);
if (code < 0) { if (code < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY; tEncoderClear(&encoder);
goto END; goto END;
} }
tEncoderClear(&encoder); tEncoderClear(&encoder);
@ -398,10 +406,7 @@ static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj
action.pCont = buf; action.pCont = buf;
action.contLen = sizeof(SMsgHead) + len; action.contLen = sizeof(SMsgHead) + len;
action.msgType = TDMT_VND_TMQ_ADD_CHECKINFO; action.msgType = TDMT_VND_TMQ_ADD_CHECKINFO;
code = mndTransAppendRedoAction(pTrans, &action); MND_TMQ_RETURN_CHECK(mndTransAppendRedoAction(pTrans, &action));
if (code != 0) {
goto END;
}
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
buf = NULL; buf = NULL;
} }
@ -417,33 +422,22 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
const char *userName) { const char *userName) {
mInfo("start to create topic:%s", pCreate->name); mInfo("start to create topic:%s", pCreate->name);
STrans *pTrans = NULL; STrans *pTrans = NULL;
int32_t code = -1; int32_t code = 0;
SNode *pAst = NULL; SNode *pAst = NULL;
SQueryPlan *pPlan = NULL; SQueryPlan *pPlan = NULL;
SMqTopicObj topicObj = {0}; SMqTopicObj topicObj = {0};
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq, "create-topic"); pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq, "create-topic");
if (pTrans == NULL) { MND_TMQ_NULL_CHECK(pTrans);
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
code = -1;
goto _OUT;
}
mndTransSetDbName(pTrans, pDb->name, NULL); mndTransSetDbName(pTrans, pDb->name, NULL);
code = mndTransCheckConflict(pMnode, pTrans); MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));
if (code != 0) {
goto _OUT;
}
mInfo("trans:%d to create topic:%s", pTrans->id, pCreate->name); mInfo("trans:%d to create topic:%s", pTrans->id, pCreate->name);
tstrncpy(topicObj.name, pCreate->name, TSDB_TOPIC_FNAME_LEN); tstrncpy(topicObj.name, pCreate->name, TSDB_TOPIC_FNAME_LEN);
tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN); tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN);
tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN); tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);
code = mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_TOPIC, &topicObj); MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_TOPIC, &topicObj));
if (code != 0) {
goto _OUT;
}
topicObj.createTime = taosGetTimestampMs(); topicObj.createTime = taosGetTimestampMs();
topicObj.updateTime = topicObj.createTime; topicObj.updateTime = topicObj.createTime;
@ -458,56 +452,25 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) { if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
topicObj.ast = taosStrdup(pCreate->ast); topicObj.ast = taosStrdup(pCreate->ast);
topicObj.astLen = strlen(pCreate->ast) + 1; topicObj.astLen = strlen(pCreate->ast) + 1;
qDebugL("topic:%s ast %s", topicObj.name, topicObj.ast); qDebugL("topic:%s ast %s", topicObj.name, topicObj.ast);
MND_TMQ_RETURN_CHECK(nodesStringToNode(pCreate->ast, &pAst));
code = nodesStringToNode(pCreate->ast, &pAst);
if (code != 0) {
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
goto _OUT;
}
SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true}; SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};
code = qCreateQueryPlan(&cxt, &pPlan, NULL); MND_TMQ_RETURN_CHECK(qCreateQueryPlan(&cxt, &pPlan, NULL));
if (code != 0) {
mError("failed to create topic:%s since %s", pCreate->name, terrstr());
goto _OUT;
}
topicObj.ntbColIds = taosArrayInit(0, sizeof(int16_t)); topicObj.ntbColIds = taosArrayInit(0, sizeof(int16_t));
if (topicObj.ntbColIds == NULL) { MND_TMQ_NULL_CHECK(topicObj.ntbColIds);
terrno = TSDB_CODE_OUT_OF_MEMORY; MND_TMQ_RETURN_CHECK(extractTopicTbInfo(pAst, &topicObj));
code = terrno;
goto _OUT;
}
extractTopicTbInfo(pAst, &topicObj);
if (topicObj.ntbUid == 0) { if (topicObj.ntbUid == 0) {
taosArrayDestroy(topicObj.ntbColIds); (void)taosArrayDestroy(topicObj.ntbColIds);
topicObj.ntbColIds = NULL; topicObj.ntbColIds = NULL;
} }
code = qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema); MND_TMQ_RETURN_CHECK(qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema));
if (code != 0) { MND_TMQ_RETURN_CHECK(nodesNodeToString((SNode *)pPlan, false, &topicObj.physicalPlan, NULL));
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
goto _OUT;
}
code = nodesNodeToString((SNode *)pPlan, false, &topicObj.physicalPlan, NULL);
if (code != 0) {
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
goto _OUT;
}
} else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) { } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {
SStbObj *pStb = mndAcquireStb(pMnode, pCreate->subStbName); SStbObj *pStb = mndAcquireStb(pMnode, pCreate->subStbName);
if (pStb == NULL) { MND_TMQ_NULL_CHECK(pStb);
terrno = TSDB_CODE_MND_STB_NOT_EXIST; tstrncpy(topicObj.stbName, pCreate->subStbName, TSDB_TABLE_FNAME_LEN);
code = terrno;
goto _OUT;
}
strcpy(topicObj.stbName, pCreate->subStbName);
topicObj.stbUid = pStb->uid; topicObj.stbUid = pStb->uid;
mndReleaseStb(pMnode, pStb); mndReleaseStb(pMnode, pStb);
if(pCreate->ast != NULL){ if(pCreate->ast != NULL){
@ -518,34 +481,25 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
} }
SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj); SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) { MND_TMQ_NULL_CHECK(pCommitRaw);
mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr()); code = mndTransAppendCommitlog(pTrans, pCommitRaw);
code = -1; if(code != 0) {
goto _OUT; sdbFreeRaw(pCommitRaw);
goto END;
} }
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY); MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
if (topicObj.ntbUid != 0) { if (topicObj.ntbUid != 0) {
code = sendCheckInfoToVnode(pTrans, pMnode, &topicObj); MND_TMQ_RETURN_CHECK(sendCheckInfoToVnode(pTrans, pMnode, &topicObj));
if (code != 0){
goto _OUT;
}
} }
MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));
if (mndTransPrepare(pMnode, pTrans) != 0) {
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
code = -1;
goto _OUT;
}
code = TSDB_CODE_ACTION_IN_PROGRESS; code = TSDB_CODE_ACTION_IN_PROGRESS;
_OUT: END:
taosMemoryFreeClear(topicObj.physicalPlan); taosMemoryFreeClear(topicObj.physicalPlan);
taosMemoryFreeClear(topicObj.sql); taosMemoryFreeClear(topicObj.sql);
taosMemoryFreeClear(topicObj.ast); taosMemoryFreeClear(topicObj.ast);
taosArrayDestroy(topicObj.ntbColIds); (void)taosArrayDestroy(topicObj.ntbColIds);
if (topicObj.schema.nCols) { if (topicObj.schema.nCols) {
taosMemoryFreeClear(topicObj.schema.pSchema); taosMemoryFreeClear(topicObj.schema.pSchema);
} }
@ -557,75 +511,64 @@ _OUT:
static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) { static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
int32_t code = -1; int32_t code = TDB_CODE_SUCCESS;
SMqTopicObj *pTopic = NULL; SMqTopicObj *pTopic = NULL;
SDbObj *pDb = NULL; SDbObj *pDb = NULL;
SCMCreateTopicReq createTopicReq = {0}; SCMCreateTopicReq createTopicReq = {0};
if (tDeserializeSCMCreateTopicReq(pReq->pCont, pReq->contLen, &createTopicReq) != 0) { if (tDeserializeSCMCreateTopicReq(pReq->pCont, pReq->contLen, &createTopicReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG; code = TSDB_CODE_INVALID_MSG;
goto _OVER; goto END;
} }
mInfo("topic:%s start to create, sql:%s", createTopicReq.name, createTopicReq.sql); mInfo("topic:%s start to create, sql:%s", createTopicReq.name, createTopicReq.sql);
if (mndCheckCreateTopicReq(&createTopicReq) != 0) { MND_TMQ_RETURN_CHECK(mndCheckCreateTopicReq(&createTopicReq));
mError("topic:%s failed to create since %s", createTopicReq.name, terrstr());
goto _OVER;
}
pTopic = mndAcquireTopic(pMnode, createTopicReq.name); code = mndAcquireTopic(pMnode, createTopicReq.name, &pTopic);
if (pTopic != NULL) { if (code == TDB_CODE_SUCCESS) {
if (createTopicReq.igExists) { if (createTopicReq.igExists) {
mInfo("topic:%s already exist, ignore exist is set", createTopicReq.name); mInfo("topic:%s already exist, ignore exist is set", createTopicReq.name);
code = 0; goto END;
goto _OVER;
} else { } else {
terrno = TSDB_CODE_MND_TOPIC_ALREADY_EXIST; code = TSDB_CODE_MND_TOPIC_ALREADY_EXIST;
goto _OVER; goto END;
} }
} else if (terrno != TSDB_CODE_MND_TOPIC_NOT_EXIST) { } else if (code != TSDB_CODE_MND_TOPIC_NOT_EXIST) {
goto _OVER; goto END;
} }
pDb = mndAcquireDb(pMnode, createTopicReq.subDbName); pDb = mndAcquireDb(pMnode, createTopicReq.subDbName);
if (pDb == NULL) { MND_TMQ_NULL_CHECK(pDb);
terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
goto _OVER;
}
if (pDb->cfg.walRetentionPeriod == 0) { if (pDb->cfg.walRetentionPeriod == 0) {
terrno = TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO; code = TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO;
mError("db:%s, not allowed to create topic when WAL_RETENTION_PERIOD is zero", pDb->name); mError("db:%s, not allowed to create topic when WAL_RETENTION_PERIOD is zero", pDb->name);
goto _OVER; goto END;
} }
if (sdbGetSize(pMnode->pSdb, SDB_TOPIC) >= tmqMaxTopicNum){ if (sdbGetSize(pMnode->pSdb, SDB_TOPIC) >= tmqMaxTopicNum){
terrno = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE; code = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;
mError("topic num out of range"); mError("topic num out of range");
goto _OVER; goto END;
}
if ((terrno = grantCheck(TSDB_GRANT_SUBSCRIPTION)) < 0) {
goto _OVER;
} }
MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));
code = mndCreateTopic(pMnode, pReq, &createTopicReq, pDb, pReq->info.conn.user); code = mndCreateTopic(pMnode, pReq, &createTopicReq, pDb, pReq->info.conn.user);
if (code == 0) { if (code == 0) {
code = TSDB_CODE_ACTION_IN_PROGRESS; code = TSDB_CODE_ACTION_IN_PROGRESS;
} }
SName dbname = {0}; {
tNameFromString(&dbname, createTopicReq.subDbName, T_NAME_ACCT | T_NAME_DB); SName dbname = {0};
(void)tNameFromString(&dbname, createTopicReq.subDbName, T_NAME_ACCT | T_NAME_DB); // ignore error
SName topicName = {0};
(void)tNameFromString(&topicName, createTopicReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE); // ignore error
auditRecord(pReq, pMnode->clusterId, "createTopic", dbname.dbname, topicName.dbname,
createTopicReq.sql, strlen(createTopicReq.sql));
}
SName topicName = {0}; END:
tNameFromString(&topicName, createTopicReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
//reuse this function for topic
auditRecord(pReq, pMnode->clusterId, "createTopic", dbname.dbname, topicName.dbname,
createTopicReq.sql, strlen(createTopicReq.sql));
_OVER:
if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) { if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
mError("failed to create topic:%s since %s", createTopicReq.name, terrstr()); mError("failed to create topic:%s since %s", createTopicReq.name, terrstr());
} }
@ -638,22 +581,20 @@ _OVER:
} }
static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SRpcMsg *pReq, SMqTopicObj *pTopic) { static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SRpcMsg *pReq, SMqTopicObj *pTopic) {
int32_t code = -1; int32_t code = 0;
if (mndUserRemoveTopic(pMnode, pTrans, pTopic->name) != 0) { SSdbRaw *pCommitRaw = NULL;
goto _OVER; MND_TMQ_RETURN_CHECK(mndUserRemoveTopic(pMnode, pTrans, pTopic->name));
pCommitRaw = mndTopicActionEncode(pTopic);
MND_TMQ_NULL_CHECK(pCommitRaw);
code = mndTransAppendCommitlog(pTrans, pCommitRaw);
if(code != 0) {
sdbFreeRaw(pCommitRaw);
goto END;
} }
MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));
SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic); END:
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
(void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
if (mndTransPrepare(pMnode, pTrans) != 0) {
goto _OVER;
}
code = 0;
_OVER:
return code; return code;
} }
@ -661,7 +602,7 @@ static bool checkTopic(SArray *topics, char *topicName){
int32_t sz = taosArrayGetSize(topics); int32_t sz = taosArrayGetSize(topics);
for (int32_t i = 0; i < sz; i++) { for (int32_t i = 0; i < sz; i++) {
char *name = taosArrayGetP(topics, i); char *name = taosArrayGetP(topics, i);
if (strcmp(name, topicName) == 0) { if (name && strcmp(name, topicName) == 0) {
return true; return true;
} }
} }
@ -682,29 +623,26 @@ static int32_t mndDropConsumerByTopic(SMnode *pMnode, STrans *pTrans, char *topi
bool found = checkTopic(pConsumer->assignedTopics, topicName); bool found = checkTopic(pConsumer->assignedTopics, topicName);
if (found){ if (found){
if (pConsumer->status == MQ_CONSUMER_STATUS_LOST) { if (pConsumer->status == MQ_CONSUMER_STATUS_LOST) {
code = mndSetConsumerDropLogs(pTrans, pConsumer); MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumer));
if (code != 0) {
goto end;
}
sdbRelease(pSdb, pConsumer); sdbRelease(pSdb, pConsumer);
continue; continue;
} }
mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s", mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",
topicName, pConsumer->consumerId, pConsumer->cgroup); topicName, pConsumer->consumerId, pConsumer->cgroup);
code = TSDB_CODE_MND_TOPIC_SUBSCRIBED; code = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
goto end; goto END;
} }
if (checkTopic(pConsumer->rebNewTopics, topicName) || checkTopic(pConsumer->rebRemovedTopics, topicName)) { if (checkTopic(pConsumer->rebNewTopics, topicName) || checkTopic(pConsumer->rebRemovedTopics, topicName)) {
code = TSDB_CODE_MND_TOPIC_SUBSCRIBED; code = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb new)", mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb new)",
topicName, pConsumer->consumerId, pConsumer->cgroup); topicName, pConsumer->consumerId, pConsumer->cgroup);
goto end; goto END;
} }
sdbRelease(pSdb, pConsumer); sdbRelease(pSdb, pConsumer);
} }
end: END:
sdbRelease(pSdb, pConsumer); sdbRelease(pSdb, pConsumer);
sdbCancelFetch(pSdb, pIter); sdbCancelFetch(pSdb, pIter);
return code; return code;
@ -726,13 +664,10 @@ static int32_t mndDropCheckInfoByTopic(SMnode *pMnode, STrans *pTrans, SMqTopicO
} }
buf = taosMemoryCalloc(1, sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN); buf = taosMemoryCalloc(1, sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN);
if (buf == NULL){ MND_TMQ_NULL_CHECK(buf);
code = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead)); void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId); ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);
memcpy(abuf, pTopic->name, TSDB_TOPIC_FNAME_LEN); (void)memcpy(abuf, pTopic->name, TSDB_TOPIC_FNAME_LEN);
STransAction action = {0}; STransAction action = {0};
action.epSet = mndGetVgroupEpset(pMnode, pVgroup); action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
@ -742,12 +677,12 @@ static int32_t mndDropCheckInfoByTopic(SMnode *pMnode, STrans *pTrans, SMqTopicO
code = mndTransAppendRedoAction(pTrans, &action); code = mndTransAppendRedoAction(pTrans, &action);
if (code != 0) { if (code != 0) {
taosMemoryFree(buf); taosMemoryFree(buf);
goto end; goto END;
} }
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
} }
end: END:
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
sdbCancelFetch(pSdb, pIter); sdbCancelFetch(pSdb, pIter);
return code; return code;
@ -761,70 +696,40 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
STrans *pTrans = NULL; STrans *pTrans = NULL;
if (tDeserializeSMDropTopicReq(pReq->pCont, pReq->contLen, &dropReq) != 0) { if (tDeserializeSMDropTopicReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG; return TSDB_CODE_INVALID_MSG;
return -1;
} }
pTopic = mndAcquireTopic(pMnode, dropReq.name); code = mndAcquireTopic(pMnode, dropReq.name, &pTopic);
if (pTopic == NULL) { if (code != 0) {
if (dropReq.igNotExists) { if (dropReq.igNotExists) {
mInfo("topic:%s, not exist, ignore not exist is set", dropReq.name); mInfo("topic:%s, not exist, ignore not exist is set", dropReq.name);
tFreeSMDropTopicReq(&dropReq); tFreeSMDropTopicReq(&dropReq);
return 0; return 0;
} else { } else {
terrno = TSDB_CODE_MND_TOPIC_NOT_EXIST;
mError("topic:%s, failed to drop since %s", dropReq.name, terrstr()); mError("topic:%s, failed to drop since %s", dropReq.name, terrstr());
tFreeSMDropTopicReq(&dropReq); tFreeSMDropTopicReq(&dropReq);
return -1; return code;
} }
} }
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "drop-topic"); pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "drop-topic");
if (pTrans == NULL) { MND_TMQ_NULL_CHECK(pTrans);
mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
code = -1;
goto end;
}
mndTransSetDbName(pTrans, pTopic->db, NULL); mndTransSetDbName(pTrans, pTopic->db, NULL);
code = mndTransCheckConflict(pMnode, pTrans); MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));
if (code != 0) {
goto end;
}
mInfo("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name); mInfo("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name);
MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_TOPIC, pTopic));
code = mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_TOPIC, pTopic); MND_TMQ_RETURN_CHECK(mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_READ_DB, pTopic->db));
if (code != 0) { MND_TMQ_RETURN_CHECK(mndDropConsumerByTopic(pMnode, pTrans, dropReq.name));
goto end; MND_TMQ_RETURN_CHECK(mndDropSubByTopic(pMnode, pTrans, dropReq.name));
}
code = mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_READ_DB, pTopic->db);
if (code != 0) {
goto end;
}
code = mndDropConsumerByTopic(pMnode, pTrans, dropReq.name);
if (code != 0) {
goto end;
}
code = mndDropSubByTopic(pMnode, pTrans, dropReq.name);
if (code < 0) {
mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
goto end;
}
if (pTopic->ntbUid != 0) { if (pTopic->ntbUid != 0) {
code = mndDropCheckInfoByTopic(pMnode, pTrans, pTopic); MND_TMQ_RETURN_CHECK(mndDropCheckInfoByTopic(pMnode, pTrans, pTopic));
if (code != 0) {
goto end;
}
} }
code = mndDropTopic(pMnode, pTrans, pReq, pTopic); code = mndDropTopic(pMnode, pTrans, pReq, pTopic);
end: END:
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
mndTransDrop(pTrans); mndTransDrop(pTrans);
if (code != 0) { if (code != 0) {
@ -834,8 +739,7 @@ end:
} }
SName name = {0}; SName name = {0};
tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE); (void)tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE); // ignore error
auditRecord(pReq, pMnode->clusterId, "dropTopic", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen); auditRecord(pReq, pMnode->clusterId, "dropTopic", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen);
tFreeSMDropTopicReq(&dropReq); tFreeSMDropTopicReq(&dropReq);
@ -849,8 +753,7 @@ int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
SDbObj *pDb = mndAcquireDb(pMnode, dbName); SDbObj *pDb = mndAcquireDb(pMnode, dbName);
if (pDb == NULL) { if (pDb == NULL) {
terrno = TSDB_CODE_MND_DB_NOT_SELECTED; return TSDB_CODE_MND_DB_NOT_SELECTED;
return -1;
} }
int32_t numOfTopics = 0; int32_t numOfTopics = 0;
@ -875,18 +778,24 @@ int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {
} }
static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){ static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){
char* string = NULL; char* string = NULL;
int32_t code = 0;
cJSON* columns = cJSON_CreateArray(); cJSON* columns = cJSON_CreateArray();
if (columns == NULL) { MND_TMQ_NULL_CHECK(columns);
return;
}
for (int i = 0; i < nCols; i++) { for (int i = 0; i < nCols; i++) {
cJSON* column = cJSON_CreateObject(); cJSON* column = cJSON_CreateObject();
MND_TMQ_NULL_CHECK(column);
SSchema* s = schema + i; SSchema* s = schema + i;
cJSON* cname = cJSON_CreateString(s->name); cJSON* cname = cJSON_CreateString(s->name);
cJSON_AddItemToObject(column, "name", cname); MND_TMQ_NULL_CHECK(cname);
if (!cJSON_AddItemToObject(column, "name", cname)) {
return;
}
cJSON* ctype = cJSON_CreateString(tDataTypes[s->type].name); cJSON* ctype = cJSON_CreateString(tDataTypes[s->type].name);
cJSON_AddItemToObject(column, "type", ctype); MND_TMQ_NULL_CHECK(ctype);
if (!cJSON_AddItemToObject(column, "type", ctype)) {
return;
}
int32_t length = 0; int32_t length = 0;
if (s->type == TSDB_DATA_TYPE_BINARY || s->type == TSDB_DATA_TYPE_VARBINARY || if (s->type == TSDB_DATA_TYPE_BINARY || s->type == TSDB_DATA_TYPE_VARBINARY ||
s->type == TSDB_DATA_TYPE_GEOMETRY) { s->type == TSDB_DATA_TYPE_GEOMETRY) {
@ -897,8 +806,13 @@ static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){
length = s->bytes; length = s->bytes;
} }
cJSON* cbytes = cJSON_CreateNumber(length); cJSON* cbytes = cJSON_CreateNumber(length);
cJSON_AddItemToObject(column, "length", cbytes); MND_TMQ_NULL_CHECK(cbytes);
cJSON_AddItemToArray(columns, column); if (!cJSON_AddItemToObject(column, "length", cbytes)){
return;
}
if (!cJSON_AddItemToArray(columns, column)){
return;
}
} }
string = cJSON_PrintUnformatted(columns); string = cJSON_PrintUnformatted(columns);
cJSON_Delete(columns); cJSON_Delete(columns);
@ -910,6 +824,9 @@ static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){
mError("mndRetrieveTopic build schema error json:%p, json len:%zu", string, len); mError("mndRetrieveTopic build schema error json:%p, json len:%zu", string, len);
} }
taosMemoryFree(string); taosMemoryFree(string);
END:
return;
} }
static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) { static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
@ -917,13 +834,16 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0; int32_t numOfRows = 0;
SMqTopicObj *pTopic = NULL; SMqTopicObj *pTopic = NULL;
int32_t code = 0;
char *sql = NULL;
char *schemaJson = NULL;
while (numOfRows < rowsCapacity) { while (numOfRows < rowsCapacity) {
pShow->pIter = sdbFetch(pSdb, SDB_TOPIC, pShow->pIter, (void **)&pTopic); pShow->pIter = sdbFetch(pSdb, SDB_TOPIC, pShow->pIter, (void **)&pTopic);
if (pShow->pIter == NULL) break; if (pShow->pIter == NULL) break;
SColumnInfoData *pColInfo; SColumnInfoData *pColInfo= NULL;
SName n; SName n = {0};
int32_t cols = 0; int32_t cols = 0;
char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE + 5] = {0}; char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE + 5] = {0};
@ -931,28 +851,34 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
STR_TO_VARSTR(topicName, pName); STR_TO_VARSTR(topicName, pName);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)topicName, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)topicName, false));
char dbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; char dbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&n, pTopic->db, T_NAME_ACCT | T_NAME_DB); MND_TMQ_RETURN_CHECK(tNameFromString(&n, pTopic->db, T_NAME_ACCT | T_NAME_DB));
tNameGetDbName(&n, varDataVal(dbName)); MND_TMQ_RETURN_CHECK(tNameGetDbName(&n, varDataVal(dbName)));
varDataSetLen(dbName, strlen(varDataVal(dbName))); varDataSetLen(dbName, strlen(varDataVal(dbName)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)dbName, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)dbName, false));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pTopic->createTime, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pTopic->createTime, false));
char *sql = taosMemoryMalloc(strlen(pTopic->sql) + VARSTR_HEADER_SIZE); sql = taosMemoryMalloc(strlen(pTopic->sql) + VARSTR_HEADER_SIZE);
MND_TMQ_NULL_CHECK(sql);
STR_TO_VARSTR(sql, pTopic->sql); STR_TO_VARSTR(sql, pTopic->sql);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)sql, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)sql, false));
taosMemoryFree(sql); taosMemoryFreeClear(sql);
char *schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE); schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);
MND_TMQ_NULL_CHECK(schemaJson);
if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){ if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){
schemaToJson(pTopic->schema.pSchema, pTopic->schema.nCols, schemaJson); schemaToJson(pTopic->schema.pSchema, pTopic->schema.nCols, schemaJson);
}else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){ }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){
@ -969,8 +895,9 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
} }
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)schemaJson, false); MND_TMQ_NULL_CHECK(pColInfo);
taosMemoryFree(schemaJson); MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)schemaJson, false));
taosMemoryFreeClear(schemaJson);
char mete[4 + VARSTR_HEADER_SIZE] = {0}; char mete[4 + VARSTR_HEADER_SIZE] = {0};
if(pTopic->withMeta){ if(pTopic->withMeta){
@ -980,7 +907,8 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
} }
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)mete, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)mete, false));
char type[8 + VARSTR_HEADER_SIZE] = {0}; char type[8 + VARSTR_HEADER_SIZE] = {0};
if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){ if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){
@ -992,7 +920,8 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
} }
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)type, false); MND_TMQ_NULL_CHECK(pColInfo);
MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)type, false));
numOfRows++; numOfRows++;
sdbRelease(pSdb, pTopic); sdbRelease(pSdb, pTopic);
@ -1000,24 +929,11 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
pShow->numOfRows += numOfRows; pShow->numOfRows += numOfRows;
return numOfRows; return numOfRows;
}
int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) { END:
SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic); taosMemoryFreeClear(schemaJson);
if (pCommitRaw == NULL) return -1; taosMemoryFreeClear(sql);
if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1; return code;
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
return 0;
}
static int32_t mndSetDropTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) {
SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
if (pCommitRaw == NULL) return -1;
if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
return 0;
} }
static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter) { static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter) {
@ -1047,33 +963,3 @@ bool mndTopicExistsForDb(SMnode *pMnode, SDbObj *pDb) {
return false; return false;
} }
#if 0
int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
int32_t code = 0;
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
SMqTopicObj *pTopic = NULL;
while (1) {
pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
if (pIter == NULL) break;
if (pTopic->dbUid != pDb->uid) {
sdbRelease(pSdb, pTopic);
continue;
}
if (mndSetDropTopicCommitLogs(pMnode, pTrans, pTopic) < 0) {
sdbRelease(pSdb, pTopic);
sdbCancelFetch(pSdb, pIter);
code = -1;
break;
}
sdbRelease(pSdb, pTopic);
}
return code;
}
#endif

18
source/dnode/mnode/impl/src/mndUser.c
View File

@ -1357,7 +1357,7 @@ SHashObj *mndDupUseDbHash(SHashObj *pOld) {
} }
int32_t mndUserDupObj(SUserObj *pUser, SUserObj *pNew) { int32_t mndUserDupObj(SUserObj *pUser, SUserObj *pNew) {
memcpy(pNew, pUser, sizeof(SUserObj)); (void)memcpy(pNew, pUser, sizeof(SUserObj));
pNew->authVersion++; pNew->authVersion++;
pNew->updateTime = taosGetTimestampMs(); pNew->updateTime = taosGetTimestampMs();
@ -1377,7 +1377,7 @@ int32_t mndUserDupObj(SUserObj *pUser, SUserObj *pNew) {
taosRUnLockLatch(&pUser->lock); taosRUnLockLatch(&pUser->lock);
if (pNew->readDbs == NULL || pNew->writeDbs == NULL || pNew->topics == NULL) { if (pNew->readDbs == NULL || pNew->writeDbs == NULL || pNew->topics == NULL) {
return -1; return TSDB_CODE_INVALID_PARA;
} }
return 0; return 0;
} }
@ -1983,10 +1983,11 @@ static int32_t mndProcessAlterUserPrivilegesReq(SAlterUserReq *pAlterReq, SMnode
if (ALTER_USER_ADD_SUBSCRIBE_TOPIC_PRIV(pAlterReq->alterType, pAlterReq->privileges)) { if (ALTER_USER_ADD_SUBSCRIBE_TOPIC_PRIV(pAlterReq->alterType, pAlterReq->privileges)) {
int32_t len = strlen(pAlterReq->objname) + 1; int32_t len = strlen(pAlterReq->objname) + 1;
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, pAlterReq->objname); SMqTopicObj *pTopic = NULL;
if (pTopic == NULL) { int32_t code = mndAcquireTopic(pMnode, pAlterReq->objname, &pTopic);
if (code != 0) {
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
return -1; return code;
} }
taosHashPut(pNewUser->topics, pTopic->name, len, pTopic->name, TSDB_TOPIC_FNAME_LEN); taosHashPut(pNewUser->topics, pTopic->name, len, pTopic->name, TSDB_TOPIC_FNAME_LEN);
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
@ -1994,10 +1995,11 @@ static int32_t mndProcessAlterUserPrivilegesReq(SAlterUserReq *pAlterReq, SMnode
if (ALTER_USER_DEL_SUBSCRIBE_TOPIC_PRIV(pAlterReq->alterType, pAlterReq->privileges)) { if (ALTER_USER_DEL_SUBSCRIBE_TOPIC_PRIV(pAlterReq->alterType, pAlterReq->privileges)) {
int32_t len = strlen(pAlterReq->objname) + 1; int32_t len = strlen(pAlterReq->objname) + 1;
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, pAlterReq->objname); SMqTopicObj *pTopic = NULL;
if (pTopic == NULL) { int32_t code = mndAcquireTopic(pMnode, pAlterReq->objname, &pTopic);
if (code != 0) {
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);
return -1; return code;
} }
taosHashRemove(pNewUser->topics, pAlterReq->objname, len); taosHashRemove(pNewUser->topics, pAlterReq->objname, len);
mndReleaseTopic(pMnode, pTopic); mndReleaseTopic(pMnode, pTopic);

9
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -2725,15 +2725,6 @@ int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj *pVgro
SDbObj dbObj = {0}; SDbObj dbObj = {0};
SArray *pArray = mndBuildDnodesArray(pMnode, 0); SArray *pArray = mndBuildDnodesArray(pMnode, 0);
// int32_t numOfTopics = 0;
// if (mndGetNumOfTopics(pMnode, pDb->name, &numOfTopics) != 0) {
// goto _OVER;
// }
// if (numOfTopics > 0) {
// terrno = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;
// goto _OVER;
// }
int32_t numOfStreams = 0; int32_t numOfStreams = 0;
if (mndGetNumOfStreams(pMnode, pDb->name, &numOfStreams) != 0) { if (mndGetNumOfStreams(pMnode, pDb->name, &numOfStreams) != 0) {
goto _OVER; goto _OVER;