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

enh(tmq): do not show closed consumer

This commit is contained in:
Liu Jicong 2022-05-11 02:58:17 +08:00
parent 3b56457f68
commit 15dd272154
5 changed files with 29 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tmsg.h
View File

@ -1493,7 +1493,7 @@ typedef struct {
} SMVSubscribeRsp; } SMVSubscribeRsp;
typedef struct { typedef struct {
char name[TSDB_TABLE_FNAME_LEN]; char name[TSDB_TOPIC_FNAME_LEN];
int8_t igNotExists; int8_t igNotExists;
} SMDropTopicReq; } SMDropTopicReq;

4
source/client/src/tmq.c
View File

@ -1307,6 +1307,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
} }
tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) { tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
if (tmq->status == TMQ_CONSUMER_STATUS__READY) {
tmq_list_t* lst = tmq_list_new(); tmq_list_t* lst = tmq_list_new();
tmq_resp_err_t rsp = tmq_subscribe(tmq, lst); tmq_resp_err_t rsp = tmq_subscribe(tmq, lst);
tmq_list_destroy(lst); tmq_list_destroy(lst);
@ -1316,6 +1317,9 @@ tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
} else { } else {
return TMQ_RESP_ERR__FAIL; return TMQ_RESP_ERR__FAIL;
} }
}
// TODO: free resources
return TMQ_RESP_ERR__SUCCESS;
} }
const char* tmq_err2str(tmq_resp_err_t err) { const char* tmq_err2str(tmq_resp_err_t err) {

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

@ -29,6 +29,7 @@ enum {
MQ_CONSUMER_STATUS__LOST, MQ_CONSUMER_STATUS__LOST,
MQ_CONSUMER_STATUS__LOST_IN_REB, MQ_CONSUMER_STATUS__LOST_IN_REB,
MQ_CONSUMER_STATUS__LOST_REBD, MQ_CONSUMER_STATUS__LOST_REBD,
MQ_CONSUMER_STATUS__REMOVED,
}; };
int32_t mndInitConsumer(SMnode *pMnode); int32_t mndInitConsumer(SMnode *pMnode);

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

@ -486,6 +486,14 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) {
} }
} }
if (pConsumerOld && taosArrayGetSize(pConsumerNew->rebNewTopics) == 0 &&
taosArrayGetSize(pConsumerNew->rebRemovedTopics) == 0) {
/*if (taosArrayGetSize(pConsumerNew->assignedTopics) == 0) {*/
/*pConsumerNew->updateType = */
/*}*/
goto SUBSCRIBE_OVER;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg);
if (pTrans == NULL) goto SUBSCRIBE_OVER; if (pTrans == NULL) goto SUBSCRIBE_OVER;
if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto SUBSCRIBE_OVER; if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto SUBSCRIBE_OVER;
@ -684,9 +692,6 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
if (pOldConsumer->status == MQ_CONSUMER_STATUS__MODIFY || if (pOldConsumer->status == MQ_CONSUMER_STATUS__MODIFY ||
pOldConsumer->status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) { pOldConsumer->status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
pOldConsumer->status = MQ_CONSUMER_STATUS__READY; pOldConsumer->status = MQ_CONSUMER_STATUS__READY;
// TODO: remove
/*if (taosArrayGetSize(pOldConsumer->assignedTopics) == 0) {*/
/*}*/
} else if (pOldConsumer->status == MQ_CONSUMER_STATUS__LOST_IN_REB || } else if (pOldConsumer->status == MQ_CONSUMER_STATUS__LOST_IN_REB ||
pOldConsumer->status == MQ_CONSUMER_STATUS__LOST) { pOldConsumer->status == MQ_CONSUMER_STATUS__LOST) {
pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_REBD; pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_REBD;

13
tests/test/c/tmqSim.c
View File

@ -331,12 +331,6 @@ void loop_consume(SThreadInfo* pInfo) {
} }
} }
err = tmq_consumer_close(pInfo->tmq);
if (err) {
printf("tmq_consumer_close() fail, reason: %s\n", tmq_err2str(err));
exit(-1);
}
pInfo->consumeMsgCnt = totalMsgs; pInfo->consumeMsgCnt = totalMsgs;
pInfo->consumeRowCnt = totalRows; pInfo->consumeRowCnt = totalRows;
@ -372,6 +366,13 @@ void* consumeThreadFunc(void* param) {
return NULL; return NULL;
} }
err = tmq_consumer_close(pInfo->tmq);
if (err) {
printf("tmq_consumer_close() fail, reason: %s\n", tmq_err2str(err));
exit(-1);
}
pInfo->tmq = NULL;
// save consume result into consumeresult table // save consume result into consumeresult table
saveConsumeResult(pInfo); saveConsumeResult(pInfo);