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

Merge pull request #27021 from taosdata/fix/TD-31115 

fix:[TD-31115] add consumer check when drop topic or group in tmq
This commit is contained in:
dapan1121 2024-08-07 09:42:12 +08:00 committed by GitHub
parent d3afac44d1 555df05e48
commit b6a7049e75
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 62 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -30,6 +30,7 @@ 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);
void mndTopicGetShowName(const char* fullTopic, char* topic); void mndTopicGetShowName(const char* fullTopic, char* topic);
bool checkTopic(SArray *topics, char *topicName);
int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics); int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics);

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

@ -1018,37 +1018,37 @@ END:
return code; return code;
} }
//static int32_t mndDropConsumerByGroup(SMnode *pMnode, STrans *pTrans, char *cgroup, char *topic) { static int32_t mndCheckConsumerByGroup(SMnode *pMnode, STrans *pTrans, char *cgroup, char *topic) {
// void *pIter = NULL; void *pIter = NULL;
// SMqConsumerObj *pConsumer = NULL; SMqConsumerObj *pConsumer = NULL;
// int code = 0; int code = 0;
// while (1) { while (1) {
// pIter = sdbFetch(pMnode->pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer); pIter = sdbFetch(pMnode->pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
// if (pIter == NULL) { if (pIter == NULL) {
// break; break;
// } }
//
// // drop consumer in lost status, other consumers not in lost status already deleted by rebalance if (strcmp(cgroup, pConsumer->cgroup) != 0) {
// if (pConsumer->status != MQ_CONSUMER_STATUS_LOST || strcmp(cgroup, pConsumer->cgroup) != 0) { sdbRelease(pMnode->pSdb, pConsumer);
// sdbRelease(pMnode->pSdb, pConsumer); continue;
// continue; }
// }
// int32_t sz = taosArrayGetSize(pConsumer->assignedTopics); bool found = checkTopic(pConsumer->assignedTopics, topic);
// for (int32_t i = 0; i < sz; i++) { if (found){
// char *name = taosArrayGetP(pConsumer->assignedTopics, i); mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",
// if (name && strcmp(topic, name) == 0) { topic, pConsumer->consumerId, pConsumer->cgroup);
// MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumer)); code = TSDB_CODE_MND_CGROUP_USED;
// } goto END;
// } }
//
// sdbRelease(pMnode->pSdb, pConsumer); sdbRelease(pMnode->pSdb, pConsumer);
// } }
//
//END: END:
// sdbRelease(pMnode->pSdb, pConsumer); sdbRelease(pMnode->pSdb, pConsumer);
// sdbCancelFetch(pMnode->pSdb, pIter); sdbCancelFetch(pMnode->pSdb, pIter);
// return code; return code;
//} }
static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) { static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) {
SMnode *pMnode = pMsg->info.node; SMnode *pMnode = pMsg->info.node;
@ -1085,6 +1085,7 @@ static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) {
mndTransSetDbName(pTrans, pSub->dbName, NULL); mndTransSetDbName(pTrans, pSub->dbName, NULL);
MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans)); MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));
MND_TMQ_RETURN_CHECK(sendDeleteSubToVnode(pMnode, pSub, pTrans)); MND_TMQ_RETURN_CHECK(sendDeleteSubToVnode(pMnode, pSub, pTrans));
MND_TMQ_RETURN_CHECK(mndCheckConsumerByGroup(pMnode, pTrans, dropReq.cgroup, dropReq.topic));
MND_TMQ_RETURN_CHECK(mndSetDropSubCommitLogs(pMnode, pTrans, pSub)); MND_TMQ_RETURN_CHECK(mndSetDropSubCommitLogs(pMnode, pTrans, pSub));
MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans)); MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

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

@ -602,7 +602,7 @@ END:
return code; return code;
} }
static bool checkTopic(SArray *topics, char *topicName){ 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);
@ -613,44 +613,33 @@ static bool checkTopic(SArray *topics, char *topicName){
return false; return false;
} }
//static int32_t mndDropConsumerByTopic(SMnode *pMnode, STrans *pTrans, char *topicName){ static int32_t mndCheckConsumerByTopic(SMnode *pMnode, STrans *pTrans, char *topicName){
// int32_t code = 0; int32_t code = 0;
// SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
// void *pIter = NULL; void *pIter = NULL;
// SMqConsumerObj *pConsumer = NULL; SMqConsumerObj *pConsumer = NULL;
// while (1) { while (1) {
// pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer); pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
// if (pIter == NULL) { if (pIter == NULL) {
// break; break;
// } }
//
// bool found = checkTopic(pConsumer->assignedTopics, topicName); bool found = checkTopic(pConsumer->assignedTopics, topicName);
// if (found){ if (found){
// if (pConsumer->status == MQ_CONSUMER_STATUS_LOST) { mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",
// MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumer)); topicName, pConsumer->consumerId, pConsumer->cgroup);
// sdbRelease(pSdb, pConsumer); code = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
// continue; goto END;
// } }
// mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",
// topicName, pConsumer->consumerId, pConsumer->cgroup); sdbRelease(pSdb, pConsumer);
// code = TSDB_CODE_MND_TOPIC_SUBSCRIBED; }
// goto END;
// } END:
// sdbRelease(pSdb, pConsumer);
// if (checkTopic(pConsumer->rebNewTopics, topicName) || checkTopic(pConsumer->rebRemovedTopics, topicName)) { sdbCancelFetch(pSdb, pIter);
// code = TSDB_CODE_MND_TOPIC_SUBSCRIBED; return code;
// mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb new)", }
// topicName, pConsumer->consumerId, pConsumer->cgroup);
// goto END;
// }
// sdbRelease(pSdb, pConsumer);
// }
//
//END:
// sdbRelease(pSdb, pConsumer);
// sdbCancelFetch(pSdb, pIter);
// return code;
//}
static int32_t mndDropCheckInfoByTopic(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic){ static int32_t mndDropCheckInfoByTopic(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic){
// broadcast to all vnode // broadcast to all vnode
@ -725,7 +714,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_TOPIC, pTopic)); MND_TMQ_RETURN_CHECK(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)); MND_TMQ_RETURN_CHECK(mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_READ_DB, pTopic->db));
// MND_TMQ_RETURN_CHECK(mndDropConsumerByTopic(pMnode, pTrans, dropReq.name)); MND_TMQ_RETURN_CHECK(mndCheckConsumerByTopic(pMnode, pTrans, dropReq.name));
MND_TMQ_RETURN_CHECK(mndDropSubByTopic(pMnode, pTrans, dropReq.name)); MND_TMQ_RETURN_CHECK(mndDropSubByTopic(pMnode, pTrans, dropReq.name));
if (pTopic->ntbUid != 0) { if (pTopic->ntbUid != 0) {