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

add self check

This commit is contained in:
Yihao Deng 2024-06-27 01:33:25 +00:00
parent ac351c5b58
commit 0a35d7ef6b
2 changed files with 4 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -558,7 +558,6 @@ int32_t restoreCheckpointData(const char* path, const char* key, int64_t chkptId
goto _EXIT; goto _EXIT;
} }
// int32_t pathLen = strlen(path) + 48;
char* checkpointRoot = taosMemoryCalloc(1, pathLen + 48); char* checkpointRoot = taosMemoryCalloc(1, pathLen + 48);
if (checkpointRoot == NULL) { if (checkpointRoot == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -575,15 +574,15 @@ int32_t restoreCheckpointData(const char* path, const char* key, int64_t chkptId
stDebug("%s check local backend dir:%s, checkpointId:%" PRId64 " succ", key, defaultPath, chkptId); stDebug("%s check local backend dir:%s, checkpointId:%" PRId64 " succ", key, defaultPath, chkptId);
char* chkptPath = taosMemoryCalloc(1, pathLen); char* chkptPath = taosMemoryCalloc(1, pathLen + 128);
if (chkptPath == NULL) { if (chkptPath == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _EXIT; goto _EXIT;
} }
if (chkptId > 0) { if (chkptId > 0) {
snprintf(chkptPath, pathLen, "%s%s%s%s%s%" PRId64 "", prefixPath, TD_DIRSEP, "checkpoints", TD_DIRSEP, "checkpoint", snprintf(chkptPath, pathLen + 127, "%s%s%s%s%s%" PRId64 "", prefixPath, TD_DIRSEP, "checkpoints", TD_DIRSEP,
chkptId); "checkpoint", chkptId);
code = rebuildFromLocalCheckpoint(key, chkptPath, chkptId, defaultPath, processVer); code = rebuildFromLocalCheckpoint(key, chkptPath, chkptId, defaultPath, processVer);
if (code != 0) { if (code != 0) {

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

@ -667,7 +667,7 @@ int32_t streamTaskBuildCheckpoint(SStreamTask* pTask) {
if (pTask->info.taskLevel != TASK_LEVEL__SINK) { if (pTask->info.taskLevel != TASK_LEVEL__SINK) {
stDebug("s-task:%s level:%d start gen checkpoint, checkpointId:%" PRId64, id, pTask->info.taskLevel, ckId); stDebug("s-task:%s level:%d start gen checkpoint, checkpointId:%" PRId64, id, pTask->info.taskLevel, ckId);
int64_t ver = 0; int64_t ver = pTask->chkInfo.processedVer;
code = streamBackendDoCheckpoint(pTask->pBackend, ckId, ver); code = streamBackendDoCheckpoint(pTask->pBackend, ckId, ver);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
stError("s-task:%s gen checkpoint:%" PRId64 " failed, code:%s", id, ckId, tstrerror(terrno)); stError("s-task:%s gen checkpoint:%" PRId64 " failed, code:%s", id, ckId, tstrerror(terrno));