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

refactor(stream): inject the reject dispatch error in stream to test the effect of fix.

This commit is contained in:
Haojun Liao 2024-12-16 13:53:13 +08:00
parent aa187f1ee3
commit 779c1bd8ad
2 changed files with 7 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mnode/impl/src/mndStreamUtil.c
View File

@ -1377,8 +1377,8 @@ int32_t mndFindChangedNodeInfo(SMnode *pMnode, const SArray *pPrevNodeList, cons
return TSDB_CODE_INVALID_PARA; return TSDB_CODE_INVALID_PARA;
} }
pInfo->pUpdateNodeList = taosArrayInit(4, sizeof(SNodeUpdateInfo)), pInfo->pUpdateNodeList = taosArrayInit(4, sizeof(SNodeUpdateInfo));
pInfo->pDBMap = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK); pInfo->pDBMap = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK);
if (pInfo->pUpdateNodeList == NULL || pInfo->pDBMap == NULL) { if (pInfo->pUpdateNodeList == NULL || pInfo->pDBMap == NULL) {
mndDestroyVgroupChangeInfo(pInfo); mndDestroyVgroupChangeInfo(pInfo);

5
source/libs/stream/src/streamDispatch.c
View File

@ -1872,6 +1872,11 @@ int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, S
} }
} }
#if 0
// inject errors, and always refuse the upstream dispatch msg and trigger the task nodeEpset update trans.
status = TASK_INPUT_STATUS__REFUSED;
#endif
{ {
// do send response with the input status // do send response with the input status
int32_t code = buildDispatchRsp(pTask, pReq, status, &pRsp->pCont); int32_t code = buildDispatchRsp(pTask, pReq, status, &pRsp->pCont);