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

fix:put poll to push manager if wal not exist when offset is latest

This commit is contained in:
wangmm0220 2023-05-26 19:17:04 +08:00
parent 50487255b3
commit a3e214b9e8
4 changed files with 14 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/libs/wal/wal.h
View File

@ -205,7 +205,8 @@ int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead);
int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead); int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead);
int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead); int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead);
SWalRef *walRefFirstVer(SWal *, SWalRef *); void walRefFirstVer(SWal *, SWalRef *);
void walRefLastVer(SWal *, SWalRef *);
SWalRef *walRefCommittedVer(SWal *); SWalRef *walRefCommittedVer(SWal *);
SWalRef *walOpenRef(SWal *); SWalRef *walOpenRef(SWal *);

12
source/dnode/vnode/src/tq/tqUtil.c
View File

@ -115,18 +115,12 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
tqOffsetResetToData(pOffsetVal, 0, 0); tqOffsetResetToData(pOffsetVal, 0, 0);
} }
} else { } else {
pHandle->pRef = walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef); walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef);
if (pHandle->pRef == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
// offset set to previous version when init
tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer - 1); tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer - 1);
} }
} else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) { } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
// offset set to previous version when init walRefLastVer(pTq->pVnode->pWal, pHandle->pRef);
tqOffsetResetToLog(pOffsetVal, walGetLastVer(pTq->pVnode->pWal)); tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer);
} else if (reqOffset.type == TMQ_OFFSET__RESET_NONE) { } else if (reqOffset.type == TMQ_OFFSET__RESET_NONE) {
tqError("tmq poll: subkey:%s, no offset committed for consumer:0x%" PRIx64 " in vg %d, subkey %s, reset none failed", tqError("tmq poll: subkey:%s, no offset committed for consumer:0x%" PRIx64 " in vg %d, subkey %s, reset none failed",
pHandle->subKey, consumerId, vgId, pRequest->subKey); pHandle->subKey, consumerId, vgId, pRequest->subKey);

3
source/libs/executor/src/executor.c
View File

@ -1080,9 +1080,6 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
const char* id = GET_TASKID(pTaskInfo); const char* id = GET_TASKID(pTaskInfo);
// if pOffset equal to current offset, means continue consume // if pOffset equal to current offset, means continue consume
char buf[80] = {0};
tFormatOffset(buf, 80, &pTaskInfo->streamInfo.currentOffset);
qDebug("currentOffset:%s", buf);
if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) { if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) {
return 0; return 0;
} }

17
source/libs/wal/src/walRef.c
View File

@ -63,21 +63,22 @@ int32_t walSetRefVer(SWalRef *pRef, int64_t ver) {
return 0; return 0;
} }
SWalRef *walRefFirstVer(SWal *pWal, SWalRef *pRef) { void walRefFirstVer(SWal *pWal, SWalRef *pRef) {
if (pRef == NULL) {
pRef = walOpenRef(pWal);
if (pRef == NULL) {
return NULL;
}
}
taosThreadMutexLock(&pWal->mutex); taosThreadMutexLock(&pWal->mutex);
int64_t ver = walGetFirstVer(pWal); int64_t ver = walGetFirstVer(pWal);
pRef->refVer = ver; pRef->refVer = ver;
taosThreadMutexUnlock(&pWal->mutex); taosThreadMutexUnlock(&pWal->mutex);
wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, ver); wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, ver);
}
return pRef; void walRefLastVer(SWal *pWal, SWalRef *pRef) {
taosThreadMutexLock(&pWal->mutex);
int64_t ver = walGetLastVer(pWal);
pRef->refVer = ver;
taosThreadMutexUnlock(&pWal->mutex);
wDebug("vgId:%d, wal ref version %" PRId64 " for last", pWal->cfg.vgId, ver);
} }
SWalRef *walRefCommittedVer(SWal *pWal) { SWalRef *walRefCommittedVer(SWal *pWal) {