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

fix(stream): add check for null ptr and leader .

This commit is contained in:
Haojun Liao 2023-12-18 11:52:59 +08:00
parent f2bba670d8
commit b4a1907fb7
2 changed files with 22 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source/dnode/vnode/src/tqCommon/tqCommon.c
View File

@ -368,6 +368,7 @@ int32_t tqStreamTaskProcessScanHistoryFinishReq(SStreamMeta* pMeta, SRpcMsg* pMs
} }
int32_t tqStreamTaskProcessScanHistoryFinishRsp(SStreamMeta* pMeta, SRpcMsg* pMsg) { int32_t tqStreamTaskProcessScanHistoryFinishRsp(SStreamMeta* pMeta, SRpcMsg* pMsg) {
int32_t code = TSDB_CODE_SUCCESS;
char* msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); char* msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
int32_t msgLen = pMsg->contLen - sizeof(SMsgHead); int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
@ -379,6 +380,12 @@ int32_t tqStreamTaskProcessScanHistoryFinishRsp(SStreamMeta* pMeta, SRpcMsg* pMs
tDecodeCompleteHistoryDataMsg(&decoder, &req); tDecodeCompleteHistoryDataMsg(&decoder, &req);
tDecoderClear(&decoder); tDecoderClear(&decoder);
if (pMeta->role == NODE_ROLE_FOLLOWER) {
tqError("s-task:0x%x (vgId:%d) not handle the scan-history finish rsp, since it becomes follower",
req.upstreamTaskId, pMeta->vgId);
return TASK_DOWNSTREAM_NOT_LEADER;
}
SStreamTask* pTask = streamMetaAcquireTask(pMeta, req.streamId, req.upstreamTaskId); SStreamTask* pTask = streamMetaAcquireTask(pMeta, req.streamId, req.upstreamTaskId);
if (pTask == NULL) { if (pTask == NULL) {
tqError("vgId:%d process scan history finish rsp, failed to find task:0x%x, it may be destroyed", tqError("vgId:%d process scan history finish rsp, failed to find task:0x%x, it may be destroyed",
@ -395,11 +402,11 @@ int32_t tqStreamTaskProcessScanHistoryFinishRsp(SStreamMeta* pMeta, SRpcMsg* pMs
"s-task:%s scan-history finish rsp received from downstream task:0x%x, all downstream tasks rsp scan-history " "s-task:%s scan-history finish rsp received from downstream task:0x%x, all downstream tasks rsp scan-history "
"completed msg", "completed msg",
pTask->id.idStr, req.downstreamId); pTask->id.idStr, req.downstreamId);
streamProcessScanHistoryFinishRsp(pTask); code = streamProcessScanHistoryFinishRsp(pTask);
} }
streamMetaReleaseTask(pMeta, pTask); streamMetaReleaseTask(pMeta, pTask);
return 0; return code;
} }
int32_t tqStreamTaskProcessCheckReq(SStreamMeta* pMeta, SRpcMsg* pMsg) { int32_t tqStreamTaskProcessCheckReq(SStreamMeta* pMeta, SRpcMsg* pMsg) {

16
source/libs/stream/src/streamStart.c
View File

@ -476,11 +476,16 @@ int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRs
// automatically set the related fill-history task to be failed. // automatically set the related fill-history task to be failed.
if (HAS_RELATED_FILLHISTORY_TASK(pTask)) { if (HAS_RELATED_FILLHISTORY_TASK(pTask)) {
STaskId* pId = &pTask->hTaskInfo.id; STaskId* pId = &pTask->hTaskInfo.id;
int64_t current = taosGetTimestampMs();
SStreamTask* pHTask = streamMetaAcquireTask(pTask->pMeta, pId->streamId, pId->taskId); SStreamTask* pHTask = streamMetaAcquireTask(pTask->pMeta, pId->streamId, pId->taskId);
streamMetaUpdateTaskDownstreamStatus(pHTask->pMeta, pId->streamId, pId->taskId, pHTask->execInfo.init, if (pHTask != NULL) {
taosGetTimestampMs(), false); streamMetaUpdateTaskDownstreamStatus(pTask->pMeta, pId->streamId, pId->taskId, pHTask->execInfo.init, current,
streamMetaReleaseTask(pTask->pMeta, pHTask); false);
streamMetaReleaseTask(pTask->pMeta, pHTask);
} else {
streamMetaUpdateTaskDownstreamStatus(pTask->pMeta, pId->streamId, pId->taskId, 0, current, false);
}
} }
} else { // TASK_DOWNSTREAM_NOT_READY, let's retry in 100ms } else { // TASK_DOWNSTREAM_NOT_READY, let's retry in 100ms
STaskRecheckInfo* pInfo = createRecheckInfo(pTask, pRsp); STaskRecheckInfo* pInfo = createRecheckInfo(pTask, pRsp);
@ -657,6 +662,11 @@ int32_t streamProcessScanHistoryFinishReq(SStreamTask* pTask, SStreamScanHistory
int32_t streamProcessScanHistoryFinishRsp(SStreamTask* pTask) { int32_t streamProcessScanHistoryFinishRsp(SStreamTask* pTask) {
ETaskStatus status = streamTaskGetStatus(pTask, NULL); ETaskStatus status = streamTaskGetStatus(pTask, NULL);
// task restart now, not handle the scan-history finish rsp
if (status == TASK_STATUS__UNINIT) {
return TSDB_CODE_INVALID_MSG;
}
ASSERT(status == TASK_STATUS__SCAN_HISTORY || status == TASK_STATUS__STREAM_SCAN_HISTORY); ASSERT(status == TASK_STATUS__SCAN_HISTORY || status == TASK_STATUS__STREAM_SCAN_HISTORY);
SStreamMeta* pMeta = pTask->pMeta; SStreamMeta* pMeta = pTask->pMeta;