From a5a10f706895fdec3168ad722c2c4ba6301c0901 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 26 Apr 2022 17:08:42 +0800 Subject: [PATCH] feat: tmq support show --- example/src/tmq.c | 1 - include/common/tcommon.h | 89 +++--------- include/common/tmsg.h | 150 ++------------------ include/common/tmsgdef.h | 2 +- include/util/tdef.h | 19 +-- source/client/src/tmq.c | 73 +++++----- source/dnode/mgmt/mgmt_mnode/src/mmHandle.c | 2 +- source/dnode/mnode/impl/inc/mndDef.h | 12 +- source/dnode/mnode/impl/src/mndConsumer.c | 22 +-- source/dnode/mnode/impl/src/mndDef.c | 2 +- source/dnode/mnode/impl/src/mndPerfSchema.c | 24 ++-- source/dnode/mnode/impl/src/mndScheduler.c | 7 +- source/dnode/mnode/impl/src/mndTopic.c | 24 ++-- source/util/src/terror.c | 4 +- 14 files changed, 125 insertions(+), 306 deletions(-) diff --git a/example/src/tmq.c b/example/src/tmq.c index 21f60ada5d..5b8f66f666 100644 --- a/example/src/tmq.c +++ b/example/src/tmq.c @@ -18,7 +18,6 @@ #include #include #include "taos.h" -#include "osSleep.h" static int running = 1; static void msg_process(TAOS_RES* msg) { diff --git a/include/common/tcommon.h b/include/common/tcommon.h index 7c308f9354..fa0ceca8a4 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -66,13 +66,13 @@ typedef struct SDataBlockInfo { int32_t rows; int32_t rowSize; union { - int64_t uid; // from which table of uid, comes from this data block + int64_t uid; // from which table of uid, comes from this data block int64_t blockId; }; - uint64_t groupId; // no need to serialize - int16_t numOfCols; - int16_t hasVarCol; - int16_t capacity; + uint64_t groupId; // no need to serialize + int16_t numOfCols; + int16_t hasVarCol; + int16_t capacity; } SDataBlockInfo; typedef struct SSDataBlock { @@ -122,59 +122,6 @@ static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) { static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) { blockDestroyInner(pBlock); } -static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp) { - int32_t tlen = 0; - int32_t sz = 0; - // tlen += taosEncodeFixedI64(buf, pRsp->consumerId); - tlen += taosEncodeFixedI64(buf, pRsp->reqOffset); - tlen += taosEncodeFixedI64(buf, pRsp->rspOffset); - tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum); - tlen += taosEncodeFixedI32(buf, pRsp->numOfTopics); - if (pRsp->numOfTopics == 0) return tlen; - tlen += taosEncodeSSchemaWrapper(buf, pRsp->schema); - if (pRsp->pBlockData) { - sz = taosArrayGetSize(pRsp->pBlockData); - } - tlen += taosEncodeFixedI32(buf, sz); - for (int32_t i = 0; i < sz; i++) { - SSDataBlock* pBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i); - tlen += tEncodeDataBlock(buf, pBlock); - } - return tlen; -} - -static FORCE_INLINE void* tDecodeSMqPollRsp(void* buf, SMqPollRsp* pRsp) { - int32_t sz; - // buf = taosDecodeFixedI64(buf, &pRsp->consumerId); - buf = taosDecodeFixedI64(buf, &pRsp->reqOffset); - buf = taosDecodeFixedI64(buf, &pRsp->rspOffset); - buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum); - buf = taosDecodeFixedI32(buf, &pRsp->numOfTopics); - if (pRsp->numOfTopics == 0) return buf; - pRsp->schema = (SSchemaWrapper*)taosMemoryCalloc(1, sizeof(SSchemaWrapper)); - if (pRsp->schema == NULL) return NULL; - buf = taosDecodeSSchemaWrapper(buf, pRsp->schema); - buf = taosDecodeFixedI32(buf, &sz); - pRsp->pBlockData = taosArrayInit(sz, sizeof(SSDataBlock)); - for (int32_t i = 0; i < sz; i++) { - SSDataBlock block = {0}; - tDecodeDataBlock(buf, &block); - taosArrayPush(pRsp->pBlockData, &block); - } - return buf; -} - -static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqPollRsp* pRsp) { - if (pRsp->schema) { - if (pRsp->schema->nCols) { - taosMemoryFreeClear(pRsp->schema->pSchema); - } - taosMemoryFree(pRsp->schema); - } - taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))blockDestroyInner); - pRsp->pBlockData = NULL; -} - //====================================================================================================================== // the following structure shared by parser and executor typedef struct SColumn { @@ -195,22 +142,22 @@ typedef struct SColumn { } SColumn; typedef struct STableBlockDistInfo { - uint16_t rowSize; - uint16_t numOfFiles; - uint32_t numOfTables; - uint64_t totalSize; - uint64_t totalRows; - int32_t maxRows; - int32_t minRows; - int32_t firstSeekTimeUs; - uint32_t numOfRowsInMemTable; - uint32_t numOfSmallBlocks; - SArray *dataBlockInfos; + uint16_t rowSize; + uint16_t numOfFiles; + uint32_t numOfTables; + uint64_t totalSize; + uint64_t totalRows; + int32_t maxRows; + int32_t minRows; + int32_t firstSeekTimeUs; + uint32_t numOfRowsInMemTable; + uint32_t numOfSmallBlocks; + SArray* dataBlockInfos; } STableBlockDistInfo; enum { FUNC_PARAM_TYPE_VALUE = 0x1, - FUNC_PARAM_TYPE_COLUMN= 0x2, + FUNC_PARAM_TYPE_COLUMN = 0x2, }; typedef struct SFunctParam { @@ -241,7 +188,7 @@ typedef struct SExprInfo { struct tExprNode* pExpr; } SExprInfo; -#define QUERY_ASC_FORWARD_STEP 1 +#define QUERY_ASC_FORWARD_STEP 1 #define QUERY_DESC_FORWARD_STEP -1 #define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 26cbd35e90..b554c4f0e5 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1507,12 +1507,12 @@ typedef struct { } SDDropTopicReq; typedef struct { - float xFilesFactor; - int32_t delay; - int32_t qmsg1Len; - int32_t qmsg2Len; - char* qmsg1; // pAst1:qmsg1:SRetention1 => trigger aggr task1 - char* qmsg2; // pAst2:qmsg2:SRetention2 => trigger aggr task2 + float xFilesFactor; + int32_t delay; + int32_t qmsg1Len; + int32_t qmsg2Len; + char* qmsg1; // pAst1:qmsg1:SRetention1 => trigger aggr task1 + char* qmsg2; // pAst2:qmsg2:SRetention2 => trigger aggr task2 } SRSmaParam; typedef struct SVCreateTbReq { @@ -1622,103 +1622,11 @@ typedef struct { char data[]; } SVShowTablesFetchRsp; -typedef struct SMqCMGetSubEpReq { +typedef struct { int64_t consumerId; int32_t epoch; char cgroup[TSDB_CGROUP_LEN]; -} SMqCMGetSubEpReq; - -static FORCE_INLINE int32_t tEncodeSMsgHead(void** buf, const SMsgHead* pMsg) { - int32_t tlen = 0; - tlen += taosEncodeFixedI32(buf, pMsg->contLen); - tlen += taosEncodeFixedI32(buf, pMsg->vgId); - return tlen; -} - -typedef struct SMqHbRsp { - int8_t status; // idle or not - int8_t vnodeChanged; - int8_t epChanged; // should use new epset - int8_t reserved; - SEpSet epSet; -} SMqHbRsp; - -static FORCE_INLINE int32_t taosEncodeSMqHbRsp(void** buf, const SMqHbRsp* pRsp) { - int32_t tlen = 0; - tlen += taosEncodeFixedI8(buf, pRsp->status); - tlen += taosEncodeFixedI8(buf, pRsp->vnodeChanged); - tlen += taosEncodeFixedI8(buf, pRsp->epChanged); - tlen += taosEncodeSEpSet(buf, &pRsp->epSet); - return tlen; -} - -static FORCE_INLINE void* taosDecodeSMqHbRsp(void* buf, SMqHbRsp* pRsp) { - buf = taosDecodeFixedI8(buf, &pRsp->status); - buf = taosDecodeFixedI8(buf, &pRsp->vnodeChanged); - buf = taosDecodeFixedI8(buf, &pRsp->epChanged); - buf = taosDecodeSEpSet(buf, &pRsp->epSet); - return buf; -} - -typedef struct SMqHbOneTopicBatchRsp { - char topicName[TSDB_TOPIC_FNAME_LEN]; - SArray* rsps; // SArray -} SMqHbOneTopicBatchRsp; - -static FORCE_INLINE int32_t taosEncodeSMqHbOneTopicBatchRsp(void** buf, const SMqHbOneTopicBatchRsp* pBatchRsp) { - int32_t tlen = 0; - tlen += taosEncodeString(buf, pBatchRsp->topicName); - int32_t sz = taosArrayGetSize(pBatchRsp->rsps); - tlen += taosEncodeFixedI32(buf, sz); - for (int32_t i = 0; i < sz; i++) { - SMqHbRsp* pRsp = (SMqHbRsp*)taosArrayGet(pBatchRsp->rsps, i); - tlen += taosEncodeSMqHbRsp(buf, pRsp); - } - return tlen; -} - -static FORCE_INLINE void* taosDecodeSMqHbOneTopicBatchRsp(void* buf, SMqHbOneTopicBatchRsp* pBatchRsp) { - int32_t sz; - buf = taosDecodeStringTo(buf, pBatchRsp->topicName); - buf = taosDecodeFixedI32(buf, &sz); - pBatchRsp->rsps = taosArrayInit(sz, sizeof(SMqHbRsp)); - for (int32_t i = 0; i < sz; i++) { - SMqHbRsp rsp; - buf = taosDecodeSMqHbRsp(buf, &rsp); - buf = taosArrayPush(pBatchRsp->rsps, &rsp); - } - return buf; -} - -typedef struct SMqHbBatchRsp { - int64_t consumerId; - SArray* batchRsps; // SArray -} SMqHbBatchRsp; - -static FORCE_INLINE int32_t taosEncodeSMqHbBatchRsp(void** buf, const SMqHbBatchRsp* pBatchRsp) { - int32_t tlen = 0; - tlen += taosEncodeFixedI64(buf, pBatchRsp->consumerId); - int32_t sz; - tlen += taosEncodeFixedI32(buf, sz); - for (int32_t i = 0; i < sz; i++) { - SMqHbOneTopicBatchRsp* pRsp = (SMqHbOneTopicBatchRsp*)taosArrayGet(pBatchRsp->batchRsps, i); - tlen += taosEncodeSMqHbOneTopicBatchRsp(buf, pRsp); - } - return tlen; -} - -static FORCE_INLINE void* taosDecodeSMqHbBatchRsp(void* buf, SMqHbBatchRsp* pBatchRsp) { - buf = taosDecodeFixedI64(buf, &pBatchRsp->consumerId); - int32_t sz; - buf = taosDecodeFixedI32(buf, &sz); - pBatchRsp->batchRsps = taosArrayInit(sz, sizeof(SMqHbOneTopicBatchRsp)); - for (int32_t i = 0; i < sz; i++) { - SMqHbOneTopicBatchRsp rsp; - buf = taosDecodeSMqHbOneTopicBatchRsp(buf, &rsp); - buf = taosArrayPush(pBatchRsp->batchRsps, &rsp); - } - return buf; -} +} SMqAskEpReq; typedef struct { int32_t key; @@ -2442,22 +2350,6 @@ typedef struct { int64_t consumerId; } SMqRspHead; -#if 0 -typedef struct { - SMsgHead head; - - int64_t consumerId; - int64_t blockingTime; - int32_t epoch; - int8_t withSchema; - char cgroup[TSDB_CGROUP_LEN]; - - int64_t currentOffset; - uint64_t reqId; - char topic[TSDB_TOPIC_FNAME_LEN]; -} SMqPollReq; -#endif - typedef struct { SMsgHead head; char subKey[TSDB_SUBSCRIBE_KEY_LEN]; @@ -2481,18 +2373,6 @@ typedef struct { SSchemaWrapper schema; } SMqSubTopicEp; -typedef struct { - SMqRspHead head; - int64_t reqOffset; - int64_t rspOffset; - int32_t skipLogNum; - // TODO: replace with topic name - int32_t numOfTopics; - // TODO: remove from msg - SSchemaWrapper* schema; - SArray* pBlockData; // SArray -} SMqPollRsp; - typedef struct { SMqRspHead head; int64_t reqOffset; @@ -2616,7 +2496,7 @@ typedef struct { SMqRspHead head; char cgroup[TSDB_CGROUP_LEN]; SArray* topics; // SArray -} SMqCMGetSubEpRsp; +} SMqAskEpRsp; static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { // taosMemoryFree(pSubTopicEp->schema.pSchema); @@ -2638,10 +2518,6 @@ static FORCE_INLINE void* tDecodeSMqSubVgEp(void* buf, SMqSubVgEp* pVgEp) { return buf; } -static FORCE_INLINE void tDeleteSMqCMGetSubEpRsp(SMqCMGetSubEpRsp* pRsp) { - taosArrayDestroyEx(pRsp->topics, (void (*)(void*))tDeleteSMqSubTopicEp); -} - static FORCE_INLINE int32_t tEncodeSMqSubTopicEp(void** buf, const SMqSubTopicEp* pTopicEp) { int32_t tlen = 0; tlen += taosEncodeString(buf, pTopicEp->topic); @@ -2674,7 +2550,7 @@ static FORCE_INLINE void* tDecodeSMqSubTopicEp(void* buf, SMqSubTopicEp* pTopicE return buf; } -static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSubEpRsp* pRsp) { +static FORCE_INLINE int32_t tEncodeSMqAskEpRsp(void** buf, const SMqAskEpRsp* pRsp) { int32_t tlen = 0; // tlen += taosEncodeString(buf, pRsp->cgroup); int32_t sz = taosArrayGetSize(pRsp->topics); @@ -2686,7 +2562,7 @@ static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSu return tlen; } -static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* pRsp) { +static FORCE_INLINE void* tDecodeSMqAskEpRsp(void* buf, SMqAskEpRsp* pRsp) { // buf = taosDecodeStringTo(buf, pRsp->cgroup); int32_t sz; buf = taosDecodeFixedI32(buf, &sz); @@ -2702,6 +2578,10 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p return buf; } +static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) { + taosArrayDestroyEx(pRsp->topics, (void (*)(void*))tDeleteSMqSubTopicEp); +} + #pragma pack(pop) #ifdef __cplusplus diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index 97ee66a2da..1976db3a12 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -145,7 +145,7 @@ enum { TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp) - TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-mq-ask-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp) + TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mnode-mq-ask-ep", SMqAskEpReq, SMqAskEpReq) TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, SMTimerReq) TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMTimerReq, SMTimerReq) TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg) diff --git a/include/util/tdef.h b/include/util/tdef.h index aa0d0bd8ff..39850f5552 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -125,12 +125,12 @@ extern const int32_t TYPE_BYTES[15]; #define TSDB_INS_TABLE_QUERIES "queries" #define TSDB_INS_TABLE_VNODES "vnodes" -#define TSDB_PERFORMANCE_SCHEMA_DB "performance_schema" -#define TSDB_PERFS_TABLE_CONNECTIONS "connections" -#define TSDB_PERFS_TABLE_QUERIES "queries" -#define TSDB_PERFS_TABLE_TOPICS "topics" -#define TSDB_PERFS_TABLE_CONSUMERS "consumers" -#define TSDB_PERFS_TABLE_SUBSCRIBES "subscribes" +#define TSDB_PERFORMANCE_SCHEMA_DB "performance_schema" +#define TSDB_PERFS_TABLE_CONNECTIONS "connections" +#define TSDB_PERFS_TABLE_QUERIES "queries" +#define TSDB_PERFS_TABLE_TOPICS "topics" +#define TSDB_PERFS_TABLE_CONSUMERS "consumers" +#define TSDB_PERFS_TABLE_SUBSCRIPTIONS "subscriptions" #define TSDB_INDEX_TYPE_SMA "SMA" #define TSDB_INDEX_TYPE_FULLTEXT "FULLTEXT" @@ -270,7 +270,7 @@ typedef enum ELogicConditionType { #define TSDB_MAX_TAGS 128 #define TSDB_MAX_TAG_CONDITIONS 1024 -#define TSDB_MAX_JSON_TAG_LEN 16384 +#define TSDB_MAX_JSON_TAG_LEN 16384 #define TSDB_AUTH_LEN 16 #define TSDB_PASSWORD_LEN 32 @@ -284,8 +284,9 @@ typedef enum ELogicConditionType { #define TSDB_IPv4ADDR_LEN 16 #define TSDB_FILENAME_LEN 128 #define TSDB_SHOW_SQL_LEN 512 -#define TSDB_SHOW_SUBQUERY_LEN 1000 #define TSDB_SLOW_QUERY_SQL_LEN 512 +#define TSDB_SHOW_SUBQUERY_LEN 1000 +#define TSDB_SHOW_LIST_LEN 1000 #define TSDB_TRANS_STAGE_LEN 12 #define TSDB_TRANS_TYPE_LEN 16 @@ -372,7 +373,7 @@ typedef enum ELogicConditionType { #define TSDB_DB_STREAM_MODE_OFF 0 #define TSDB_DB_STREAM_MODE_ON 1 #define TSDB_DEFAULT_DB_STREAM_MODE 0 -#define TSDB_DB_SINGLE_STABLE_ON 0 +#define TSDB_DB_SINGLE_STABLE_ON 0 #define TSDB_DB_SINGLE_STABLE_OFF 1 #define TSDB_DEFAULT_DB_SINGLE_STABLE 0 diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index 66e10f8c6a..e0fb46e122 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -25,7 +25,14 @@ #include "tref.h" #include "ttimer.h" -int32_t tmqAskEp(tmq_t* tmq, bool sync); +int32_t tmqAskEp(tmq_t* tmq, bool async); + +typedef struct { + int8_t inited; + tmr_h timer; +} SMqMgmt; + +static SMqMgmt tmqMgmt = {0}; typedef struct { int8_t tmqRspType; @@ -33,9 +40,9 @@ typedef struct { } SMqRspWrapper; typedef struct { - int8_t tmqRspType; - int32_t epoch; - SMqCMGetSubEpRsp msg; + int8_t tmqRspType; + int32_t epoch; + SMqAskEpRsp msg; } SMqAskEpRspWrapper; struct tmq_list_t { @@ -64,13 +71,6 @@ struct tmq_conf_t { tmq_commit_cb* commit_cb; }; -typedef struct { - int8_t inited; - tmr_h timer; -} SMqMgmt; - -static SMqMgmt tmqMgmt = {0}; - struct tmq_t { // conf char groupId[TSDB_CGROUP_LEN]; @@ -164,7 +164,7 @@ typedef struct { typedef struct { tmq_t* tmq; int32_t code; - int32_t sync; + int32_t async; tsem_t rspSem; } SMqAskEpCbParam; @@ -188,6 +188,7 @@ typedef struct { tmq_conf_t* tmq_conf_new() { tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t)); conf->autoCommit = false; + conf->autoCommitInterval = 5000; conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST; return conf; } @@ -324,7 +325,7 @@ int32_t tmqHandleAllDelayedTask(tmq_t* tmq) { if (pTaskType == NULL) break; if (*pTaskType == TMQ_DELAYED_TASK__HB) { - tmqAskEp(tmq, false); + tmqAskEp(tmq, true); taosTmrReset(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer, &tmq->hbTimer); } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) { tmq_commit(tmq, NULL, true); @@ -472,8 +473,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) { // set conf strcpy(pTmq->clientId, conf->clientId); strcpy(pTmq->groupId, conf->groupId); - /*pTmq->autoCommit = conf->autoCommit;*/ - pTmq->autoCommit = 0; + pTmq->autoCommit = conf->autoCommit; pTmq->autoCommitInterval = conf->autoCommitInterval; pTmq->commit_cb = conf->commit_cb; pTmq->resetOffsetCfg = conf->resetOffset; @@ -662,8 +662,8 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { if (code != 0) goto FAIL; // TODO: add max retry cnt - while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, true)) { - tscDebug("not ready, retry\n"); + while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) { + tscDebug("not ready, retry"); taosMsleep(500); } @@ -854,7 +854,7 @@ CREATE_MSG_FAIL: return -1; } -bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) { +bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) { /*printf("call update ep %d\n", epoch);*/ bool set = false; int32_t topicNumGet = taosArrayGetSize(pRsp->topics); @@ -936,7 +936,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) { tmq_t* tmq = pParam->tmq; pParam->code = code; if (code != 0) { - tscError("consumer %ld get topic endpoint error, not ready, wait:%d", tmq->consumerId, pParam->sync); + tscError("consumer %ld get topic endpoint error, not ready, wait:%d", tmq->consumerId, pParam->async); goto END; } @@ -950,15 +950,15 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) { goto END; } - if (pParam->sync) { - SMqCMGetSubEpRsp rsp; - tDecodeSMqCMGetSubEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp); + if (!pParam->async) { + SMqAskEpRsp rsp; + tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp); /*printf("rsp epoch %ld sz %ld\n", rsp.epoch, rsp.topics->size);*/ /*printf("tmq epoch %ld sz %ld\n", tmq->epoch, tmq->clientTopics->size);*/ if (tmqUpdateEp(tmq, head->epoch, &rsp)) { atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY); } - tDeleteSMqCMGetSubEpRsp(&rsp); + tDeleteSMqAskEpRsp(&rsp); } else { SMqAskEpRspWrapper* pWrapper = taosAllocateQitem(sizeof(SMqAskEpRspWrapper)); if (pWrapper == NULL) { @@ -969,7 +969,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) { pWrapper->tmqRspType = TMQ_MSG_TYPE__EP_RSP; pWrapper->epoch = head->epoch; memcpy(&pWrapper->msg, pMsg->pData, sizeof(SMqRspHead)); - tDecodeSMqCMGetSubEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg); + tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg); taosWriteQitem(tmq->mqueue, pWrapper); /*tsem_post(&tmq->rspSem);*/ @@ -978,13 +978,13 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) { END: /*atomic_store_8(&tmq->epStatus, 0);*/ - if (pParam->sync) { + if (!pParam->async) { tsem_post(&pParam->rspSem); } return code; } -int32_t tmqAskEp(tmq_t* tmq, bool sync) { +int32_t tmqAskEp(tmq_t* tmq, bool async) { int32_t code = 0; #if 0 int8_t epStatus = atomic_val_compare_exchange_8(&tmq->epStatus, 0, 1); @@ -995,8 +995,8 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) { } atomic_store_32(&tmq->epSkipCnt, 0); #endif - int32_t tlen = sizeof(SMqCMGetSubEpReq); - SMqCMGetSubEpReq* req = taosMemoryMalloc(tlen); + int32_t tlen = sizeof(SMqAskEpReq); + SMqAskEpReq* req = taosMemoryMalloc(tlen); if (req == NULL) { tscError("failed to malloc get subscribe ep buf"); /*atomic_store_8(&tmq->epStatus, 0);*/ @@ -1014,7 +1014,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) { return -1; } pParam->tmq = tmq; - pParam->sync = sync; + pParam->async = async; tsem_init(&pParam->rspSem, 0, 0); SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo)); @@ -1036,7 +1036,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) { sendInfo->requestObjRefId = 0; sendInfo->param = pParam; sendInfo->fp = tmqAskEpCb; - sendInfo->msgType = TDMT_MND_GET_SUB_EP; + sendInfo->msgType = TDMT_MND_MQ_ASK_EP; SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp); @@ -1045,7 +1045,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) { int64_t transporterId = 0; asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); - if (sync) { + if (!async) { tsem_wait(&pParam->rspSem); code = pParam->code; taosMemoryFree(pParam); @@ -1209,7 +1209,7 @@ int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset) /*printf("ep %d %d\n", rspMsg->head.epoch, tmq->epoch);*/ if (rspWrapper->epoch > atomic_load_32(&tmq->epoch)) { SMqAskEpRspWrapper* pEpRspWrapper = (SMqAskEpRspWrapper*)rspWrapper; - SMqCMGetSubEpRsp* rspMsg = &pEpRspWrapper->msg; + SMqAskEpRsp* rspMsg = &pEpRspWrapper->msg; tmqUpdateEp(tmq, rspWrapper->epoch, rspMsg); /*tmqClearUnhandleMsg(tmq);*/ *pReset = true; @@ -1271,15 +1271,6 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) { SMqRspObj* rspObj; int64_t startTime = taosGetTimestampMs(); - // TODO: put into delayed queue -#if 0 - int8_t status = atomic_load_8(&tmq->status); - while (0 != tmqAskEp(tmq, status != TMQ_CONSUMER_STATUS__READY)) { - tscDebug("not ready, retry\n"); - taosSsleep(1); - } -#endif - rspObj = tmqHandleAllRsp(tmq, blocking_time, false); if (rspObj) { return (TAOS_RES*)rspObj; diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c index afe57e3d8f..ed9384a869 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c @@ -211,7 +211,7 @@ void mmInitMsgHandle(SMgmtWrapper *pWrapper) { dmSetMsgHandle(pWrapper, TDMT_MND_DROP_TOPIC, mmProcessWriteMsg, DEFAULT_HANDLE); dmSetMsgHandle(pWrapper, TDMT_MND_SUBSCRIBE, mmProcessWriteMsg, DEFAULT_HANDLE); dmSetMsgHandle(pWrapper, TDMT_MND_MQ_COMMIT_OFFSET, mmProcessWriteMsg, DEFAULT_HANDLE); - dmSetMsgHandle(pWrapper, TDMT_MND_GET_SUB_EP, mmProcessReadMsg, DEFAULT_HANDLE); + dmSetMsgHandle(pWrapper, TDMT_MND_MQ_ASK_EP, mmProcessReadMsg, DEFAULT_HANDLE); dmSetMsgHandle(pWrapper, TDMT_VND_MQ_VG_CHANGE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE); dmSetMsgHandle(pWrapper, TDMT_MND_CREATE_STREAM, mmProcessWriteMsg, DEFAULT_HANDLE); dmSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY_RSP, mmProcessWriteMsg, DEFAULT_HANDLE); diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index cc56db354e..98954dc7ae 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -436,14 +436,12 @@ static FORCE_INLINE void* tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset) } typedef struct { - char name[TSDB_TOPIC_FNAME_LEN]; - char db[TSDB_DB_FNAME_LEN]; - int64_t createTime; - int64_t updateTime; - int64_t uid; - // TODO: use subDbUid + char name[TSDB_TOPIC_FNAME_LEN]; + char db[TSDB_DB_FNAME_LEN]; + int64_t createTime; + int64_t updateTime; + int64_t uid; int64_t dbUid; - int64_t subDbUid; int32_t version; int8_t subType; // db or table int8_t withTbName; diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 7f5df5d356..ff7a757007 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -59,7 +59,7 @@ int32_t mndInitConsumer(SMnode *pMnode) { .deleteFp = (SdbDeleteFp)mndConsumerActionDelete}; mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE, mndProcessSubscribeReq); - mndSetMsgHandle(pMnode, TDMT_MND_GET_SUB_EP, mndProcessAskEpReq); + mndSetMsgHandle(pMnode, TDMT_MND_MQ_ASK_EP, mndProcessAskEpReq); mndSetMsgHandle(pMnode, TDMT_MND_MQ_TIMER, mndProcessMqTimerMsg); mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_LOST, mndProcessConsumerLostMsg); return sdbSetTable(pMnode->pSdb, table); @@ -86,7 +86,7 @@ static int32_t mndProcessConsumerLostMsg(SNodeMsg *pMsg) { mndTransDrop(pTrans); return 0; FAIL: - // TODO delete consumer + tDeleteSMqConsumerObj(pConsumerNew); mndTransDrop(pTrans); return -1; } @@ -197,11 +197,11 @@ static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) { } static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) { - SMnode *pMnode = pMsg->pNode; - SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont; - SMqCMGetSubEpRsp rsp = {0}; - int64_t consumerId = be64toh(pReq->consumerId); - int32_t epoch = ntohl(pReq->epoch); + SMnode *pMnode = pMsg->pNode; + SMqAskEpReq *pReq = (SMqAskEpReq *)pMsg->rpcMsg.pCont; + SMqAskEpRsp rsp = {0}; + int64_t consumerId = be64toh(pReq->consumerId); + int32_t epoch = ntohl(pReq->epoch); SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->pNode, consumerId); if (pConsumer == NULL) { @@ -300,7 +300,7 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) { taosRUnLockLatch(&pConsumer->lock); } // encode rsp - int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqCMGetSubEpRsp(NULL, &rsp); + int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqAskEpRsp(NULL, &rsp); void *buf = rpcMallocCont(tlen); if (buf == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -311,10 +311,10 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) { ((SMqRspHead *)buf)->consumerId = pConsumer->consumerId; void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead)); - tEncodeSMqCMGetSubEpRsp(&abuf, &rsp); + tEncodeSMqAskEpRsp(&abuf, &rsp); // release consumer and free memory - tDeleteSMqCMGetSubEpRsp(&rsp); + tDeleteSMqAskEpRsp(&rsp); mndReleaseConsumer(pMnode, pConsumer); // send rsp @@ -322,7 +322,7 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) { pMsg->rspLen = tlen; return 0; FAIL: - tDeleteSMqCMGetSubEpRsp(&rsp); + tDeleteSMqAskEpRsp(&rsp); mndReleaseConsumer(pMnode, pConsumer); return -1; } diff --git a/source/dnode/mnode/impl/src/mndDef.c b/source/dnode/mnode/impl/src/mndDef.c index 800a013c4f..12de1f5bbc 100644 --- a/source/dnode/mnode/impl/src/mndDef.c +++ b/source/dnode/mnode/impl/src/mndDef.c @@ -215,7 +215,7 @@ SMqSubscribeObj *tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]) { if (pSubNew == NULL) return NULL; memcpy(pSubNew->key, key, TSDB_SUBSCRIBE_KEY_LEN); taosInitRWLatch(&pSubNew->lock); - pSubNew->vgNum = -1; + pSubNew->vgNum = 0; pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK); // TODO set free fp SMqConsumerEpInSub epInSub = { diff --git a/source/dnode/mnode/impl/src/mndPerfSchema.c b/source/dnode/mnode/impl/src/mndPerfSchema.c index a0ecbe9ae4..cf1cb34115 100644 --- a/source/dnode/mnode/impl/src/mndPerfSchema.c +++ b/source/dnode/mnode/impl/src/mndPerfSchema.c @@ -41,29 +41,31 @@ static const SPerfsTableSchema queriesSchema[] = { static const SPerfsTableSchema topicSchema[] = { {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, - /*{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},*/ + {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP}, {.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY}, - /*{.name = "row_len", .bytes = 4, .type = TSDB_DATA_TYPE_INT},*/ + // TODO config }; static const SPerfsTableSchema consumerSchema[] = { - {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, + {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, {.name = "app_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, - {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, {.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, - // ep - // up time - // topics + {.name = "topics", .bytes = TSDB_SHOW_LIST_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY}, + {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, + {.name = "end_point", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY}, + {.name = "up_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP}, }; -static const SPerfsTableSchema subscribeSchema[] = { +static const SPerfsTableSchema subscriptionSchema[] = { {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, - {.name = "offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, - {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY}, + {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, + {.name = "committed_offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, + {.name = "current_offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, + {.name = "skip_log_cnt", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, }; static const SPerfsTableMeta perfsMeta[] = { @@ -71,7 +73,7 @@ static const SPerfsTableMeta perfsMeta[] = { {TSDB_PERFS_TABLE_QUERIES, queriesSchema, tListLen(queriesSchema)}, {TSDB_PERFS_TABLE_TOPICS, topicSchema, tListLen(topicSchema)}, {TSDB_PERFS_TABLE_CONSUMERS, consumerSchema, tListLen(consumerSchema)}, - {TSDB_PERFS_TABLE_SUBSCRIBES, subscribeSchema, tListLen(subscribeSchema)}, + {TSDB_PERFS_TABLE_SUBSCRIPTIONS, subscriptionSchema, tListLen(subscriptionSchema)}, }; // connection/application/ diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 3dff65866c..a106cf348d 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -478,6 +478,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib SVgObj* pVgroup = NULL; SQueryPlan* pPlan = NULL; SSubplan* plan = NULL; + if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) { pPlan = qStringToQueryPlan(pTopic->physicalPlan); if (pPlan == NULL) { @@ -485,10 +486,6 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib return -1; } - ASSERT(pSub->vgNum == -1); - - pSub->vgNum = 0; - int32_t levelNum = LIST_LENGTH(pPlan->pSubplans); if (levelNum != 1) { qDestroyQueryPlan(pPlan); @@ -529,7 +526,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib pVgEp->vgId = pVgroup->vgId; taosArrayPush(pEpInSub->vgs, &pVgEp); - mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId); + mDebug("init subscription %s, assign vg: %d", pSub->key, pVgEp->vgId); if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) { int32_t msgLen; diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index 7c4e51298f..886ba028de 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -76,7 +76,6 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { SDB_SET_INT64(pRaw, dataPos, pTopic->updateTime, TOPIC_ENCODE_OVER); SDB_SET_INT64(pRaw, dataPos, pTopic->uid, TOPIC_ENCODE_OVER); SDB_SET_INT64(pRaw, dataPos, pTopic->dbUid, TOPIC_ENCODE_OVER); - SDB_SET_INT64(pRaw, dataPos, pTopic->subDbUid, TOPIC_ENCODE_OVER); SDB_SET_INT32(pRaw, dataPos, pTopic->version, TOPIC_ENCODE_OVER); SDB_SET_INT8(pRaw, dataPos, pTopic->subType, TOPIC_ENCODE_OVER); SDB_SET_INT8(pRaw, dataPos, pTopic->withTbName, TOPIC_ENCODE_OVER); @@ -139,7 +138,6 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) { SDB_GET_INT64(pRaw, dataPos, &pTopic->updateTime, TOPIC_DECODE_OVER); SDB_GET_INT64(pRaw, dataPos, &pTopic->uid, TOPIC_DECODE_OVER); SDB_GET_INT64(pRaw, dataPos, &pTopic->dbUid, TOPIC_DECODE_OVER); - SDB_GET_INT64(pRaw, dataPos, &pTopic->subDbUid, TOPIC_DECODE_OVER); SDB_GET_INT32(pRaw, dataPos, &pTopic->version, TOPIC_DECODE_OVER); SDB_GET_INT8(pRaw, dataPos, &pTopic->subType, TOPIC_DECODE_OVER); SDB_GET_INT8(pRaw, dataPos, &pTopic->withTbName, TOPIC_DECODE_OVER); @@ -520,29 +518,33 @@ static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB pShow->pIter = sdbFetch(pSdb, SDB_TOPIC, pShow->pIter, (void **)&pTopic); if (pShow->pIter == NULL) break; - int32_t cols = 0; + SColumnInfoData *pColInfo; + SName n; + int32_t cols = 0; char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; - - SName n; - tNameFromString(&n, pTopic->name, T_NAME_ACCT|T_NAME_DB); + tNameFromString(&n, pTopic->name, T_NAME_ACCT | T_NAME_DB); tNameGetDbName(&n, varDataVal(topicName)); varDataSetLen(topicName, strlen(varDataVal(topicName))); - - SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, numOfRows, (const char *)topicName, false); + char dbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; + tNameFromString(&n, pTopic->db, T_NAME_ACCT | T_NAME_DB); + tNameGetDbName(&n, varDataVal(dbName)); + varDataSetLen(dbName, strlen(varDataVal(dbName))); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataAppend(pColInfo, numOfRows, (const char *)&pTopic->createTime, false); + colDataAppend(pColInfo, numOfRows, (const char *)dbName, false); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataAppend(pColInfo, numOfRows, (const char *)&pTopic->createTime, false); + char sql[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0}; tstrncpy(&sql[VARSTR_HEADER_SIZE], pTopic->sql, TSDB_SHOW_SQL_LEN); varDataSetLen(sql, strlen(&sql[VARSTR_HEADER_SIZE])); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, numOfRows, (const char *)sql, false); -// taosMemoryFree(sql); - numOfRows++; sdbRelease(pSdb, pTopic); } diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 399a2255ac..53684ddfc7 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -277,8 +277,10 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_NOT_EXIST, "Transaction not exist TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_INVALID_STAGE, "Invalid stage to kill") TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_CANT_PARALLEL, "Invalid stage to kill") -// mnode-topic +// mnode-mq TAOS_DEFINE_ERROR(TSDB_CODE_MND_UNSUPPORTED_TOPIC, "Topic with aggregation is unsupported") +TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY, "Consumer waiting for rebalance") +TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_EXIST, "Consumer not exist") // mnode-sma TAOS_DEFINE_ERROR(TSDB_CODE_MND_SMA_ALREADY_EXIST, "SMA already exists") -- GitLab