diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 834615bdaa..6caddf4703 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -1646,6 +1646,9 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "checkpointInterval"); tsStreamCheckpointInterval = pItem->i32; + TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "concurrentCheckpoint"); + tsMaxConcurrentCheckpoint = pItem->i32; + TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "streamSinkDataRate"); tsSinkDataRate = pItem->fval; diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index e3f9adf033..8abe739f25 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -1294,9 +1294,10 @@ static int32_t mndProcessStreamCheckpoint(SRpcMsg *pReq) { void* p = taosArrayPush(pList, &in); if (p) { int32_t currentSize = taosArrayGetSize(pList); - mDebug("stream:%s (uid:0x%" PRIx64 ") checkpoint interval beyond threshold: %ds(%" PRId64 - "s) beyond concurrently launch threshold:%d", - pStream->name, pStream->uid, tsStreamCheckpointInterval, duration / 1000, currentSize); + mDebug("stream:%s (uid:0x%" PRIx64 ") total %d stream(s) beyond chpt interval threshold: %ds(%" PRId64 + "s), concurrently launch threshold:%d", + pStream->name, pStream->uid, currentSize, tsStreamCheckpointInterval, duration / 1000, + tsMaxConcurrentCheckpoint); } else { mError("failed to record the checkpoint interval info, stream:0x%" PRIx64, pStream->uid); } @@ -1348,7 +1349,7 @@ static int32_t mndProcessStreamCheckpoint(SRpcMsg *pReq) { code = mndProcessStreamCheckpointTrans(pMnode, p, checkpointId, 1, true); sdbRelease(pSdb, p); - if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) { + if (code == 0 || code == TSDB_CODE_ACTION_IN_PROGRESS) { started += 1; if (started >= capacity) { @@ -1356,6 +1357,8 @@ static int32_t mndProcessStreamCheckpoint(SRpcMsg *pReq) { (started + numOfCheckpointTrans)); break; } + } else { + mError("failed to start checkpoint trans, code:%s", tstrerror(code)); } } } diff --git a/tests/script/tsim/stream/concurrentcheckpt.sim b/tests/script/tsim/stream/concurrentcheckpt.sim new file mode 100644 index 0000000000..4162617deb --- /dev/null +++ b/tests/script/tsim/stream/concurrentcheckpt.sim @@ -0,0 +1,79 @@ +system sh/stop_dnodes.sh + +system sh/deploy.sh -n dnode1 -i 1 +system sh/cfg.sh -n dnode1 -c supportVnodes -v 1 + +print ========== step1 +system sh/exec.sh -n dnode1 -s start +sql connect + +sql create database abc1 vgroups 1; +sql use abc1; +sql create table st1(ts timestamp, k int) tags(a int); +sql create table t1 using st1 tags(1); +sql create table t2 using st1 tags(2); +sql insert into t1 values(now, 1); + +sql create stream str1 trigger at_once into str_dst1 as select count(*) from st1 interval(30s); +sql create stream str2 trigger at_once into str_dst2 as select count(*) from st1 interval(30s); +sql create stream str3 trigger at_once into str_dst3 as select count(*) from st1 interval(30s); + +print ============== create 3 streams, check the concurrently checkpoint +sleep 180000 + +sql select task_id, checkpoint_id from information_schema.ins_stream_tasks order by checkpoint_id; + +print $data01 $data11 $data21 +if $data01 == $data11 then + print not allowed 2 checkpoint start completed + return -1 +endi + +if $data11 == $data21 then + print not allowed 2 checkpoints start concurrently + return -1 +endi + +system sh/exec.sh -n dnode1 -s stop -x SIGINT + +print ========== concurrent checkpoint is set 2 + +system sh/stop_dnodes.sh +system sh/deploy.sh -n dnode1 -i 1 +system sh/cfg.sh -n dnode1 -c concurrentCheckpoint -v 2 + +system sh/exec.sh -n dnode1 -s start + +print ========== step2 +system sh/exec.sh -n dnode1 -s start +sql connect + +sql create database abc1 vgroups 1; +sql use abc1; +sql create table st1(ts timestamp, k int) tags(a int); +sql create table t1 using st1 tags(1); +sql create table t2 using st1 tags(2); +sql insert into t1 values(now, 1); + +sql create stream str1 trigger at_once into str_dst1 as select count(*) from st1 interval(30s); +sql create stream str2 trigger at_once into str_dst2 as select count(*) from st1 interval(30s); +sql create stream str3 trigger at_once into str_dst3 as select count(*) from st1 interval(30s); + +print ============== create 3 streams, check the concurrently checkpoint +sleep 180000 + +sql select count(*) a, checkpoint_id from information_schema.ins_stream_tasks group by checkpoint_id order by a; +print $data00 $data01 +print $data10 $data11 + +if $data00 != 1 then + print expect 1, actual $data00 + return -1 +endi + +if $data10 != 2 then + print expect 2, actual $data10 + return -1 +endi + +system sh/exec.sh -n dnode1 -s stop -x SIGINT