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

refactor: do some internal refactor.

This commit is contained in:
Haojun Liao 2023-11-08 14:02:08 +08:00
parent 809dd064a7
commit 658fafd03c
3 changed files with 6 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/mnode/impl/src/mndStream.c
View File

@ -2571,7 +2571,7 @@ int32_t doKillActiveCheckpointTrans(SMnode *pMnode) {
} }
if (transId == 0) { if (transId == 0) {
mError("failed to find the checkpoint trans, reset not executed"); mDebug("failed to find the checkpoint trans, reset not executed");
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

6
source/dnode/vnode/src/tq/tq.c
View File

@ -1919,6 +1919,8 @@ int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg) {
pMeta->startInfo.tasksWillRestart = 0; pMeta->startInfo.tasksWillRestart = 0;
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
} else { } else {
tqDebug("vgId:%d all %d task(s) nodeEp updated and closed", vgId, numOfTasks);
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
#if 0 #if 0
tqStartStreamTaskAsync(pTq, true); tqStartStreamTaskAsync(pTq, true);
@ -1950,13 +1952,13 @@ int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg) {
} }
if (vnodeIsRoleLeader(pTq->pVnode) && !tsDisableStream) { if (vnodeIsRoleLeader(pTq->pVnode) && !tsDisableStream) {
tqInfo("vgId:%d restart all stream tasks after all tasks being updated", vgId); tqInfo("vgId:%d start all stream tasks after all being updated", vgId);
tqResetStreamTaskStatus(pTq); tqResetStreamTaskStatus(pTq);
tqStartStreamTaskAsync(pTq, false); tqStartStreamTaskAsync(pTq, false);
} else { } else {
tqInfo("vgId:%d, follower node not start stream tasks", vgId); tqInfo("vgId:%d, follower node not start stream tasks", vgId);
} }
taosArrayDestroy(req.pNodeList);
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
#endif #endif
} }

2
source/dnode/vnode/src/tq/tqStreamTask.c
View File

@ -302,7 +302,7 @@ int32_t tqResetStreamTaskStatus(STQ* pTq) {
int32_t vgId = TD_VID(pTq->pVnode); int32_t vgId = TD_VID(pTq->pVnode);
int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList); int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);
tqDebug("vgId:%d start all %d stream task(s)", vgId, numOfTasks); tqDebug("vgId:%d reset all %d stream task(s) status to be uninit", vgId, numOfTasks);
if (numOfTasks == 0) { if (numOfTasks == 0) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }