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

fix(stream): set the init start timestamp.

This commit is contained in:
Haojun Liao 2023-09-11 13:49:43 +08:00
parent 0980e021e5
commit c729fc989f
2 changed files with 6 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -983,6 +983,9 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t sversion, char* msg, int32_t ms
bool restored = pTq->pVnode->restored; bool restored = pTq->pVnode->restored;
if (p != NULL && restored) { if (p != NULL && restored) {
p->tsInfo.init = taosGetTimestampMs();
tqDebug("s-task:%s set the init ts:%"PRId64, p->id.idStr, p->tsInfo.init);
streamTaskCheckDownstream(p); streamTaskCheckDownstream(p);
} else if (!restored) { } else if (!restored) {
tqWarn("s-task:%s not launched since vnode(vgId:%d) not ready", p->id.idStr, vgId); tqWarn("s-task:%s not launched since vnode(vgId:%d) not ready", p->id.idStr, vgId);

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

@ -94,6 +94,9 @@ int32_t tqCheckAndRunStreamTask(STQ* pTq) {
continue; continue;
} }
pTask->tsInfo.init = taosGetTimestampMs();
tqDebug("s-task:%s set the init ts:%"PRId64, pTask->id.idStr, pTask->tsInfo.init);
streamSetStatusNormal(pTask); streamSetStatusNormal(pTask);
streamTaskCheckDownstream(pTask); streamTaskCheckDownstream(pTask);