diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 21d1a528da..10d5140a0b 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -27,6 +27,7 @@ #define EMPTY_BLOCK_POLL_IDLE_DURATION 10 #define DEFAULT_AUTO_COMMIT_INTERVAL 5000 #define DEFAULT_HEARTBEAT_INTERVAL 3000 +#define DEFAULT_ASKEP_INTERVAL 1000 struct SMqMgmt { int8_t inited; @@ -737,35 +738,33 @@ static void generateTimedTask(int64_t refId, int32_t type) { if (tmq == NULL) return; int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t), DEF_QITEM, 0); - if (pTaskType == NULL) return; + if (pTaskType != NULL){ + *pTaskType = type; + if (taosWriteQitem(tmq->delayedTask, pTaskType) == 0){ + tsem2_post(&tmq->rspSem); + } + } - *pTaskType = type; - taosWriteQitem(tmq->delayedTask, pTaskType); - tsem2_post(&tmq->rspSem); taosReleaseRef(tmqMgmt.rsetId, refId); } void tmqAssignAskEpTask(void* param, void* tmrId) { - int64_t refId = *(int64_t*)param; + int64_t refId = (int64_t)param; generateTimedTask(refId, TMQ_DELAYED_TASK__ASK_EP); - taosMemoryFree(param); } void tmqReplayTask(void* param, void* tmrId) { - int64_t refId = *(int64_t*)param; + int64_t refId = (int64_t)param; tmq_t* tmq = taosAcquireRef(tmqMgmt.rsetId, refId); - if (tmq == NULL) goto END; + if (tmq == NULL) return; tsem2_post(&tmq->rspSem); taosReleaseRef(tmqMgmt.rsetId, refId); -END: - taosMemoryFree(param); } void tmqAssignDelayedCommitTask(void* param, void* tmrId) { - int64_t refId = *(int64_t*)param; + int64_t refId = (int64_t)param; generateTimedTask(refId, TMQ_DELAYED_TASK__COMMIT); - taosMemoryFree(param); } int32_t tmqHbCb(void* param, SDataBuf* pMsg, int32_t code) { @@ -802,11 +801,10 @@ int32_t tmqHbCb(void* param, SDataBuf* pMsg, int32_t code) { } void tmqSendHbReq(void* param, void* tmrId) { - int64_t refId = *(int64_t*)param; + int64_t refId = (int64_t)param; tmq_t* tmq = taosAcquireRef(tmqMgmt.rsetId, refId); if (tmq == NULL) { - taosMemoryFree(param); return; } @@ -880,7 +878,9 @@ void tmqSendHbReq(void* param, void* tmrId) { OVER: tDestroySMqHbReq(&req); - taosTmrReset(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, param, tmqMgmt.timer, &tmq->hbLiveTimer); + if(tmrId != NULL){ + taosTmrReset(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, param, tmqMgmt.timer, &tmq->hbLiveTimer); + } taosReleaseRef(tmqMgmt.rsetId, refId); } @@ -908,21 +908,14 @@ int32_t tmqHandleAllDelayedTask(tmq_t* pTmq) { if (*pTaskType == TMQ_DELAYED_TASK__ASK_EP) { askEp(pTmq, NULL, false, false); - int64_t* pRefId = taosMemoryMalloc(sizeof(int64_t)); - *pRefId = pTmq->refId; - tscDebug("consumer:0x%" PRIx64 " retrieve ep from mnode in 1s", pTmq->consumerId); - taosTmrReset(tmqAssignAskEpTask, 1000, pRefId, tmqMgmt.timer, &pTmq->epTimer); + taosTmrReset(tmqAssignAskEpTask, DEFAULT_ASKEP_INTERVAL, (void*)(pTmq->refId), tmqMgmt.timer, &pTmq->epTimer); } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) { tmq_commit_cb* pCallbackFn = pTmq->commitCb ? pTmq->commitCb : defaultCommitCbFn; - asyncCommitAllOffsets(pTmq, pCallbackFn, pTmq->commitCbUserParam); - int64_t* pRefId = taosMemoryMalloc(sizeof(int64_t)); - *pRefId = pTmq->refId; - tscDebug("consumer:0x%" PRIx64 " next commit to vnode(s) in %.2fs", pTmq->consumerId, pTmq->autoCommitInterval / 1000.0); - taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, pRefId, tmqMgmt.timer, &pTmq->commitTimer); + taosTmrReset(tmqAssignDelayedCommitTask, pTmq->autoCommitInterval, (void*)(pTmq->refId), tmqMgmt.timer, &pTmq->commitTimer); } else { tscError("consumer:0x%" PRIx64 " invalid task type:%d", pTmq->consumerId, *pTaskType); } @@ -1171,9 +1164,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) { goto _failed; } - int64_t* pRefId = taosMemoryMalloc(sizeof(int64_t)); - *pRefId = pTmq->refId; - pTmq->hbLiveTimer = taosTmrStart(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, pRefId, tmqMgmt.timer); + pTmq->hbLiveTimer = taosTmrStart(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, (void*)pTmq->refId, tmqMgmt.timer); char buf[TSDB_OFFSET_LEN] = {0}; STqOffsetVal offset = {.type = pTmq->resetOffsetCfg}; @@ -1301,18 +1292,9 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { } // init ep timer - if (tmq->epTimer == NULL) { - int64_t* pRefId1 = taosMemoryMalloc(sizeof(int64_t)); - *pRefId1 = tmq->refId; - tmq->epTimer = taosTmrStart(tmqAssignAskEpTask, 1000, pRefId1, tmqMgmt.timer); - } - + tmq->epTimer = taosTmrStart(tmqAssignAskEpTask, DEFAULT_ASKEP_INTERVAL, (void*)(tmq->refId), tmqMgmt.timer); // init auto commit timer - if (tmq->autoCommit && tmq->commitTimer == NULL) { - int64_t* pRefId2 = taosMemoryMalloc(sizeof(int64_t)); - *pRefId2 = tmq->refId; - tmq->commitTimer = taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, pRefId2, tmqMgmt.timer); - } + tmq->commitTimer = taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, (void*)(tmq->refId), tmqMgmt.timer); FAIL: taosArrayDestroyP(req.topicNames, taosMemoryFree); @@ -2015,9 +1997,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout) { pVg->blockReceiveTs = taosGetTimestampMs(); pVg->blockSleepForReplay = pRsp->rsp.sleepTime; if (pVg->blockSleepForReplay > 0) { - int64_t* pRefId1 = taosMemoryMalloc(sizeof(int64_t)); - *pRefId1 = tmq->refId; - taosTmrStart(tmqReplayTask, pVg->blockSleepForReplay, pRefId1, tmqMgmt.timer); + taosTmrStart(tmqReplayTask, pVg->blockSleepForReplay, (void*)(tmq->refId), tmqMgmt.timer); } } tscDebug("consumer:0x%" PRIx64 " process poll rsp, vgId:%d, offset:%s, blocks:%d, rows:%" PRId64 @@ -2274,7 +2254,7 @@ int32_t tmq_consumer_close(tmq_t* tmq) { return code; } } - taosSsleep(2); // sleep 2s for hb to send offset and rows to server + tmqSendHbReq((void*)(tmq->refId), NULL); tmq_list_t* lst = tmq_list_new(); int32_t code = tmq_subscribe(tmq, lst); diff --git a/tests/system-test/7-tmq/tmqParamsTest.py b/tests/system-test/7-tmq/tmqParamsTest.py index b25f23ef11..a323dff19e 100644 --- a/tests/system-test/7-tmq/tmqParamsTest.py +++ b/tests/system-test/7-tmq/tmqParamsTest.py @@ -134,6 +134,8 @@ class TDTestCase: if offset_value != "earliest" and offset_value != "": if offset_value == "latest": offset_value_list = list(map(lambda x: (x[-2].replace("wal:", "").replace("earliest", "0").replace("latest", "0").replace(offset_value, "0")), subscription_info)) + if None in offset_value_list: + continue offset_value_list1 = list(map(lambda x: int(x.split("/")[0]), offset_value_list)) offset_value_list2 = list(map(lambda x: int(x.split("/")[1]), offset_value_list)) tdSql.checkEqual(offset_value_list1 == offset_value_list2, True) @@ -142,6 +144,8 @@ class TDTestCase: tdSql.checkEqual(sum(rows_value_list), expected_res) elif offset_value == "none": offset_value_list = list(map(lambda x: x[-2], subscription_info)) + if None in offset_value_list: + continue offset_value_list1 = list(map(lambda x: (x.split("/")[0]), offset_value_list)) tdSql.checkEqual(offset_value_list1, ['none']*len(subscription_info)) rows_value_list = list(map(lambda x: x[-1], subscription_info)) @@ -155,6 +159,8 @@ class TDTestCase: # tdSql.checkEqual(sum(rows_value_list), expected_res) else: offset_value_list = list(map(lambda x: x[-2], subscription_info)) + if None in offset_value_list: + continue offset_value_list1 = list(map(lambda x: (x.split("/")[0]), offset_value_list)) tdSql.checkEqual(offset_value_list1, [None]*len(subscription_info)) rows_value_list = list(map(lambda x: x[-1], subscription_info)) @@ -162,6 +168,8 @@ class TDTestCase: else: if offset_value != "none": offset_value_list = list(map(lambda x: (x[-2].replace("wal:", "").replace("earliest", "0").replace("latest", "0").replace(offset_value, "0")), subscription_info)) + if None in offset_value_list: + continue offset_value_list1 = list(map(lambda x: int(x.split("/")[0]), offset_value_list)) offset_value_list2 = list(map(lambda x: int(x.split("/")[1]), offset_value_list)) tdSql.checkEqual(offset_value_list1 <= offset_value_list2, True) @@ -170,6 +178,8 @@ class TDTestCase: tdSql.checkEqual(sum(rows_value_list), expected_res) else: offset_value_list = list(map(lambda x: x[-2], subscription_info)) + if None in offset_value_list: + continue offset_value_list1 = list(map(lambda x: (x.split("/")[0]), offset_value_list)) tdSql.checkEqual(offset_value_list1, ['none']*len(subscription_info)) rows_value_list = list(map(lambda x: x[-1], subscription_info))