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

fix stream backend convert

This commit is contained in:
yihaoDeng 2023-11-17 11:56:22 +08:00
parent c7c1bdef6e
commit 1dbdb32dd7
1 changed files with 27 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -319,6 +319,11 @@ int32_t rebuildFromRemoteChkp(char* key, char* chkpPath, int64_t chkpId, char* d
return code; return code;
} }
int32_t len = strlen(defaultPath) + 32;
char* tmp = taosMemoryCalloc(1, len);
sprintf(tmp, "%s%s%s", defaultPath, TD_DIRSEP, "_tmp");
if (taosIsDir(tmp)) taosRemoveDir(tmp);
SArray* list = taosArrayInit(2, sizeof(void*)); SArray* list = taosArrayInit(2, sizeof(void*));
code = remoteChkp_readMetaData(chkpPath, list); code = remoteChkp_readMetaData(chkpPath, list);
if (code == 0) { if (code == 0) {
@ -328,19 +333,31 @@ int32_t rebuildFromRemoteChkp(char* key, char* chkpPath, int64_t chkpId, char* d
if (code == 0) { if (code == 0) {
if (taosIsDir(defaultPath)) { if (taosIsDir(defaultPath)) {
taosRemoveDir(defaultPath); taosRenameFile(defaultPath, tmp);
} }
taosMkDir(defaultPath); taosMkDir(defaultPath);
code = copyFiles(chkpPath, defaultPath); code = copyFiles(chkpPath, defaultPath);
} }
if (code != 0) {
if (taosIsDir(defaultPath)) taosRemoveDir(defaultPath);
if (taosIsDir(tmp)) taosRenameFile(tmp, defaultPath);
}
return code; return code;
} }
int32_t rebuildFromLocalChkp(char* key, char* chkpPath, int64_t chkpId, char* defaultPath) { int32_t rebuildFromLocalChkp(char* key, char* chkpPath, int64_t chkpId, char* defaultPath) {
int32_t code = 0; int32_t code = -1;
int32_t len = strlen(defaultPath) + 32;
char* tmp = taosMemoryCalloc(1, len);
sprintf(tmp, "%s%s%s", defaultPath, TD_DIRSEP, "_tmp");
if (taosIsDir(chkpPath) && isValidCheckpoint(chkpPath)) { if (taosIsDir(chkpPath) && isValidCheckpoint(chkpPath)) {
if (taosIsDir(tmp)) {
taosRemoveDir(tmp);
}
if (taosIsDir(defaultPath)) { if (taosIsDir(defaultPath)) {
taosRemoveDir(defaultPath); taosRenameFile(defaultPath, tmp);
} }
taosMkDir(defaultPath); taosMkDir(defaultPath);
code = copyFiles(chkpPath, defaultPath); code = copyFiles(chkpPath, defaultPath);
@ -349,10 +366,14 @@ int32_t rebuildFromLocalChkp(char* key, char* chkpPath, int64_t chkpId, char* de
} else { } else {
stInfo("start to restart stream backend at checkpoint path: %s", chkpPath); stInfo("start to restart stream backend at checkpoint path: %s", chkpPath);
} }
return code;
} }
return -1; if (code != 0) {
if (taosIsDir(defaultPath)) taosRemoveDir(defaultPath);
if (taosIsDir(tmp)) taosRenameFile(tmp, defaultPath);
}
taosMemoryFree(tmp);
return code;
} }
int32_t rebuildFromlocalDefault(char* key, char* chkpPath, int64_t chkpId, char* defaultPath) { int32_t rebuildFromlocalDefault(char* key, char* chkpPath, int64_t chkpId, char* defaultPath) {