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

fix(stream): clear the active checkpoint info after report to mnode.

This commit is contained in:
Haojun Liao 2024-06-28 23:40:05 +08:00
parent 5585a141d5
commit e74b8473f1
3 changed files with 12 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/stream/src/streamCheckpoint.c
View File

@ -708,7 +708,6 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
} else { // clear the checkpoint info if failed } else { // clear the checkpoint info if failed
taosThreadMutexLock(&pTask->lock); taosThreadMutexLock(&pTask->lock);
streamTaskSetFailedCheckpointId(pTask); // set failed checkpoint id before clear the checkpoint info streamTaskSetFailedCheckpointId(pTask); // set failed checkpoint id before clear the checkpoint info
streamTaskClearCheckInfo(pTask, true);
taosThreadMutexUnlock(&pTask->lock); taosThreadMutexUnlock(&pTask->lock);
code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_CHECKPOINT_DONE); code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_CHECKPOINT_DONE);

19
source/libs/stream/src/streamMeta.c
View File

@ -1053,17 +1053,20 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
entry.sinkDataSize = SIZE_IN_MiB((*pTask)->execInfo.sink.dataSize); entry.sinkDataSize = SIZE_IN_MiB((*pTask)->execInfo.sink.dataSize);
} }
if ((*pTask)->chkInfo.pActiveInfo->activeId != 0) { SActiveCheckpointInfo* p = (*pTask)->chkInfo.pActiveInfo;
entry.checkpointInfo.failed = if (p->activeId != 0) {
((*pTask)->chkInfo.pActiveInfo->failedId >= (*pTask)->chkInfo.pActiveInfo->activeId) ? 1 : 0; entry.checkpointInfo.failed = (p->failedId >= p->activeId) ? 1 : 0;
entry.checkpointInfo.activeId = (*pTask)->chkInfo.pActiveInfo->activeId; entry.checkpointInfo.activeId = p->activeId;
entry.checkpointInfo.activeTransId = (*pTask)->chkInfo.pActiveInfo->transId; entry.checkpointInfo.activeTransId = p->transId;
if (entry.checkpointInfo.failed) { if (entry.checkpointInfo.failed) {
stInfo("s-task:%s set kill checkpoint trans in hb, transId:%d", (*pTask)->id.idStr, stInfo("s-task:%s set kill checkpoint trans in hb, transId:%d, clear the active checkpointInfo",
(*pTask)->chkInfo.pActiveInfo->transId); (*pTask)->id.idStr, p->transId);
taosThreadMutexLock(&(*pTask)->lock);
streamTaskClearCheckInfo((*pTask), true);
taosThreadMutexUnlock(&(*pTask)->lock);
} }
(*pTask)->chkInfo.pActiveInfo->failedId = 0;
} }
if ((*pTask)->exec.pWalReader != NULL) { if ((*pTask)->exec.pWalReader != NULL) {

1
source/libs/stream/src/streamTask.c
View File

@ -1042,6 +1042,7 @@ void streamTaskClearActiveInfo(SActiveCheckpointInfo* pInfo) {
pInfo->transId = 0; pInfo->transId = 0;
pInfo->allUpstreamTriggerRecv = 0; pInfo->allUpstreamTriggerRecv = 0;
pInfo->dispatchTrigger = false; pInfo->dispatchTrigger = false;
pInfo->failedId = 0;
taosArrayClear(pInfo->pDispatchTriggerList); taosArrayClear(pInfo->pDispatchTriggerList);
taosArrayClear(pInfo->pCheckpointReadyRecvList); taosArrayClear(pInfo->pCheckpointReadyRecvList);