diff --git a/src/dnode/src/dnodeVWrite.c b/src/dnode/src/dnodeVWrite.c index b20e6c9749bedbef99777b1ff2d8e6c3372b3706..d439dd985ec85ea1120eaa2c11900e727fa22e94 100644 --- a/src/dnode/src/dnodeVWrite.c +++ b/src/dnode/src/dnodeVWrite.c @@ -247,6 +247,11 @@ static void *dnodeProcessWriteQueue(void *param) { if (type == TAOS_QTYPE_RPC) { pWrite = (SWriteMsg *)item; dnodeSendRpcVnodeWriteRsp(pVnode, item, pWrite->rpcMsg.code); + } else if (type == TAOS_QTYPE_FWD) { + pHead = (SWalHead *)item; + vnodeConfirmForward(pVnode, pHead->version, 0); + taosFreeQitem(item); + vnodeRelease(pVnode); } else { taosFreeQitem(item); vnodeRelease(pVnode); diff --git a/src/inc/vnode.h b/src/inc/vnode.h index a034bc5706b3c7b3f4826ad0a35382503386669c..f1cd7486a4746fb7b8a236d140f25fc73e79c68d 100644 --- a/src/inc/vnode.h +++ b/src/inc/vnode.h @@ -60,6 +60,7 @@ void* vnodeGetWal(void *pVnode); int32_t vnodeProcessWrite(void *pVnode, int qtype, void *pHead, void *item); int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes); void vnodeBuildStatusMsg(void *param); +void vnodeConfirmForward(void *param, uint64_t version, int32_t code); void vnodeSetAccess(SDMVgroupAccess *pAccess, int32_t numOfVnodes); void vnodeCleanupResources(); diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index 0050de33994c48a89abcb107e350d1dec7e2527c..5bd206beec21b128c6a653c1d17122d7bf35448a 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -779,7 +779,8 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) { vError("vgId: %d, failed to read vnode cfg, quorum not found", pVnode->vgId); goto PARSE_OVER; } - pVnode->syncCfg.quorum = (int8_t)quorum->valueint; + //pVnode->syncCfg.quorum = (int8_t)quorum->valueint; + pVnode->syncCfg.quorum = 2; cJSON *nodeInfos = cJSON_GetObjectItem(root, "nodeInfos"); if (!nodeInfos || nodeInfos->type != cJSON_Array) { diff --git a/src/vnode/src/vnodeWrite.c b/src/vnode/src/vnodeWrite.c index 171557acb61b78b3ac8ac7dff7531a21a654daf6..e44436c8b76b1e804742edaa4d4763c49411d788 100644 --- a/src/vnode/src/vnodeWrite.c +++ b/src/vnode/src/vnodeWrite.c @@ -89,6 +89,11 @@ int32_t vnodeProcessWrite(void *param1, int qtype, void *param2, void *item) { return syncCode; } +void vnodeConfirmForward(void *param, uint64_t version, int32_t code) { + SVnodeObj *pVnode = (SVnodeObj *)param; + syncConfirmForward(pVnode->sync, version, code); +} + static int32_t vnodeProcessSubmitMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pRet) { int32_t code = TSDB_CODE_SUCCESS;