Merge pull request #14031 from taosdata/feature/3.0_mhli

refactor(sync): propose fail when changing

include/common/tmsgdef.h

@@ -236,6 +236,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_SYNC_COMMON_RESPONSE, "sync-common-response", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_APPLY_MSG, "sync-apply-msg", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_CONFIG_CHANGE, "sync-config-change", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_SYNC_CONFIG_CHANGE_FINISH, "sync-config-change-finish", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_SNAPSHOT_SEND, "sync-snapshot-send", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_SNAPSHOT_RSP, "sync-snapshot-rsp", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SYNC_LEADER_TRANSFER, "sync-leader-transfer", NULL, NULL)

include/libs/sync/sync.h

@@ -24,6 +24,8 @@ extern "C" {
 #include "tdef.h"
 #include "tmsgcb.h"
 
+extern bool gRaftDetailLog;
+
 #define SYNC_INDEX_BEGIN 0
 #define SYNC_INDEX_INVALID -1
 
@@ -61,28 +63,35 @@ typedef struct SSyncCfg {
 } SSyncCfg;
 
 typedef struct SFsmCbMeta {
+  int32_t    code;
   SyncIndex  index;
+  SyncTerm   term;
+  uint64_t   seqNum;
   SyncIndex  lastConfigIndex;
-  bool       isWeak;
-  int32_t    code;
   ESyncState state;
-  uint64_t   seqNum;
-  SyncTerm   term;
   SyncTerm   currentTerm;
+  bool       isWeak;
   uint64_t   flag;
 } SFsmCbMeta;
 
 typedef struct SReConfigCbMeta {
-  int32_t   code;
-  SyncIndex index;
-  SyncTerm  term;
-  SyncIndex lastConfigIndex;
-  SyncTerm  currentTerm;
+  int32_t    code;
+  SyncIndex  index;
+  SyncTerm   term;
+  uint64_t   seqNum;
+  SyncIndex  lastConfigIndex;
+  ESyncState state;
+  SyncTerm   currentTerm;
+  bool       isWeak;
+  uint64_t   flag;
+
+  // config info
   SSyncCfg  oldCfg;
   SSyncCfg  newCfg;
-  bool      isDrop;
-  uint64_t  flag;
-  uint64_t  seqNum;
+  SyncIndex newCfgIndex;
+  SyncTerm  newCfgTerm;
+  uint64_t  newCfgSeqNum;
+
 } SReConfigCbMeta;
 
 typedef struct SSnapshot {
@@ -107,8 +116,7 @@ typedef struct SSyncFSM {
   void (*FpReConfigCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta);
   void (*FpLeaderTransferCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
 
-
-  int32_t (*FpGetSnapshot)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot, void *pReaderParam, void** ppReader);
+  int32_t (*FpGetSnapshot)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot, void* pReaderParam, void** ppReader);
   int32_t (*FpGetSnapshotInfo)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot);
 
   int32_t (*FpSnapshotStartRead)(struct SSyncFSM* pFsm, void** ppReader);
@@ -189,14 +197,13 @@ ESyncState  syncGetMyRole(int64_t rid);
 bool        syncIsReady(int64_t rid);
 const char* syncGetMyRoleStr(int64_t rid);
 SyncTerm    syncGetMyTerm(int64_t rid);
+SyncGroupId syncGetVgId(int64_t rid);
 void        syncGetEpSet(int64_t rid, SEpSet* pEpSet);
-int32_t     syncGetVgId(int64_t rid);
 int32_t     syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak);
 bool        syncEnvIsStart();
 const char* syncStr(ESyncState state);
 bool        syncIsRestoreFinish(int64_t rid);
 
-
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg);
 
 // build SRpcMsg, need to call syncPropose with SRpcMsg

include/libs/sync/syncTools.h

@@ -489,6 +489,40 @@ void syncLeaderTransferPrint2(char* s, const SyncLeaderTransfer* pMsg);
 void syncLeaderTransferLog(const SyncLeaderTransfer* pMsg);
 void syncLeaderTransferLog2(char* s, const SyncLeaderTransfer* pMsg);
 
+
+// ---------------------------------------------
+typedef struct SyncReconfigFinish {
+  uint32_t bytes;
+  int32_t  vgId;
+  uint32_t msgType;
+  SSyncCfg oldCfg;
+  SSyncCfg newCfg;
+  SyncIndex newCfgIndex;
+  SyncTerm newCfgTerm;
+  uint64_t newCfgSeqNum;
+
+} SyncReconfigFinish;
+
+SyncReconfigFinish* syncReconfigFinishBuild(int32_t vgId);
+void                syncReconfigFinishDestroy(SyncReconfigFinish* pMsg);
+void                syncReconfigFinishSerialize(const SyncReconfigFinish* pMsg, char* buf, uint32_t bufLen);
+void                syncReconfigFinishDeserialize(const char* buf, uint32_t len, SyncReconfigFinish* pMsg);
+char*               syncReconfigFinishSerialize2(const SyncReconfigFinish* pMsg, uint32_t* len);
+SyncReconfigFinish* syncReconfigFinishDeserialize2(const char* buf, uint32_t len);
+void                syncReconfigFinish2RpcMsg(const SyncReconfigFinish* pMsg, SRpcMsg* pRpcMsg);
+void                syncReconfigFinishFromRpcMsg(const SRpcMsg* pRpcMsg, SyncReconfigFinish* pMsg);
+SyncReconfigFinish* syncReconfigFinishFromRpcMsg2(const SRpcMsg* pRpcMsg);
+cJSON*              syncReconfigFinish2Json(const SyncReconfigFinish* pMsg);
+char*               syncReconfigFinish2Str(const SyncReconfigFinish* pMsg);
+
+// for debug ----------------------
+void syncReconfigFinishPrint(const SyncReconfigFinish* pMsg);
+void syncReconfigFinishPrint2(char* s, const SyncReconfigFinish* pMsg);
+void syncReconfigFinishLog(const SyncReconfigFinish* pMsg);
+void syncReconfigFinishLog2(char* s, const SyncReconfigFinish* pMsg);
+
+
+
 // on message ----------------------
 int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
 int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);

include/util/taoserror.h

@@ -416,6 +416,9 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SYN_NOT_LEADER                TAOS_DEF_ERROR_CODE(0, 0x090C)
 #define TSDB_CODE_SYN_ONE_REPLICA               TAOS_DEF_ERROR_CODE(0, 0x090D)
 #define TSDB_CODE_SYN_NOT_IN_NEW_CONFIG         TAOS_DEF_ERROR_CODE(0, 0x090E)
+#define TSDB_CODE_SYN_NEW_CONFIG_ERROR          TAOS_DEF_ERROR_CODE(0, 0x090F)
+#define TSDB_CODE_SYN_RECONFIG_NOT_READY        TAOS_DEF_ERROR_CODE(0, 0x0910)
+#define TSDB_CODE_SYN_PROPOSE_NOT_READY         TAOS_DEF_ERROR_CODE(0, 0x0911)
 #define TSDB_CODE_SYN_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x09FF)
 
 // tq

source/dnode/mnode/impl/src/mndMain.c

@@ -58,21 +58,21 @@ static void *mndBuildTimerMsg(int32_t *pContLen) {
 
 static void mndPullupTrans(SMnode *pMnode) {
   int32_t contLen = 0;
-  void *  pReq = mndBuildTimerMsg(&contLen);
+  void   *pReq = mndBuildTimerMsg(&contLen);
   SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS_TIMER, .pCont = pReq, .contLen = contLen};
   tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
 }
 
 static void mndCalMqRebalance(SMnode *pMnode) {
   int32_t contLen = 0;
-  void *  pReq = mndBuildTimerMsg(&contLen);
+  void   *pReq = mndBuildTimerMsg(&contLen);
   SRpcMsg rpcMsg = {.msgType = TDMT_MND_MQ_TIMER, .pCont = pReq, .contLen = contLen};
   tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
 }
 
 static void mndPullupTelem(SMnode *pMnode) {
   int32_t contLen = 0;
-  void *  pReq = mndBuildTimerMsg(&contLen);
+  void   *pReq = mndBuildTimerMsg(&contLen);
   SRpcMsg rpcMsg = {.msgType = TDMT_MND_TELEM_TIMER, .pCont = pReq, .contLen = contLen};
   tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
 }
@@ -86,8 +86,8 @@ static void mndPushTtlTime(SMnode *pMnode) {
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
     if (pIter == NULL) break;
 
-    int32_t contLen = sizeof(SMsgHead) + sizeof(int32_t);
-    SMsgHead   *pHead = rpcMallocCont(contLen);
+    int32_t   contLen = sizeof(SMsgHead) + sizeof(int32_t);
+    SMsgHead *pHead = rpcMallocCont(contLen);
     if (pHead == NULL) {
       mError("ttl time malloc err. contLen:%d", contLen);
       sdbRelease(pSdb, pVgroup);
@@ -97,13 +97,13 @@ static void mndPushTtlTime(SMnode *pMnode) {
     pHead->vgId = htonl(pVgroup->vgId);
 
     int32_t t = taosGetTimestampSec();
-    *(int32_t*)(POINTER_SHIFT(pHead, sizeof(SMsgHead))) = htonl(t);
+    *(int32_t *)(POINTER_SHIFT(pHead, sizeof(SMsgHead))) = htonl(t);
 
     SRpcMsg rpcMsg = {.msgType = TDMT_VND_DROP_TTL_TABLE, .pCont = pHead, .contLen = contLen};
 
-    SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
+    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
     int32_t code = tmsgSendReq(&epSet, &rpcMsg);
-    if(code != 0){
+    if (code != 0) {
       mError("ttl time seed err. code:%d", code);
     }
     mError("ttl time seed succ. time:%d", t);
@@ -117,7 +117,7 @@ static void *mndThreadFp(void *param) {
   setThreadName("mnode-timer");
 
   while (1) {
-    if (lastTime % (864000) == 0) {   // sleep 1 day for ttl
+    if (lastTime % (864000) == 0) {  // sleep 1 day for ttl
       mndPushTtlTime(pMnode);
     }
 
@@ -416,7 +416,7 @@ void mndStop(SMnode *pMnode) {
 }
 
 int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
-  SMnode *   pMnode = pMsg->info.node;
+  SMnode    *pMnode = pMsg->info.node;
   SSyncMgmt *pMgmt = &pMnode->syncMgmt;
   int32_t    code = 0;
 
@@ -433,15 +433,19 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
     return -1;
   }
 
-  char  logBuf[512] = {0};
-  char *syncNodeStr = sync2SimpleStr(pMgmt->sync);
-  snprintf(logBuf, sizeof(logBuf), "==mndProcessSyncMsg== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
-  static int64_t mndTick = 0;
-  if (++mndTick % 10 == 1) {
-    mTrace("sync trace msg:%s, %s", TMSG_INFO(pMsg->msgType), syncNodeStr);
-  }
-  syncRpcMsgLog2(logBuf, pMsg);
-  taosMemoryFree(syncNodeStr);
+  do {
+    char          *syncNodeStr = sync2SimpleStr(pMgmt->sync);
+    static int64_t mndTick = 0;
+    if (++mndTick % 10 == 1) {
+      mTrace("vgId:%d, sync heartbeat msg:%s, %s", syncGetVgId(pMgmt->sync), TMSG_INFO(pMsg->msgType), syncNodeStr);
+    }
+    if (gRaftDetailLog) {
+      char logBuf[512] = {0};
+      snprintf(logBuf, sizeof(logBuf), "==mndProcessSyncMsg== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
+      syncRpcMsgLog2(logBuf, pMsg);
+    }
+    taosMemoryFree(syncNodeStr);
+  } while (0);
 
   // ToDo: ugly! use function pointer
   if (syncNodeSnapshotEnable(pSyncNode)) {
@@ -580,7 +584,7 @@ static int32_t mndCheckMsgContent(SRpcMsg *pMsg) {
 }
 
 int32_t mndProcessRpcMsg(SRpcMsg *pMsg) {
-  SMnode * pMnode = pMsg->info.node;
+  SMnode  *pMnode = pMsg->info.node;
   MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(pMsg->msgType)];
   if (fp == NULL) {
     mError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
@@ -633,7 +637,7 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
                           SMonGrantInfo *pGrantInfo) {
   if (mndAcquireRpcRef(pMnode) != 0) return -1;
 
-  SSdb *  pSdb = pMnode->pSdb;
+  SSdb   *pSdb = pMnode->pSdb;
   int64_t ms = taosGetTimestampMs();
 
   pClusterInfo->dnodes = taosArrayInit(sdbGetSize(pSdb, SDB_DNODE), sizeof(SMonDnodeDesc));
@@ -709,7 +713,7 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
     pGrantInfo->timeseries_used += pVgroup->numOfTimeSeries;
     tstrncpy(desc.status, "unsynced", sizeof(desc.status));
     for (int32_t i = 0; i < pVgroup->replica; ++i) {
-      SVnodeGid *    pVgid = &pVgroup->vnodeGid[i];
+      SVnodeGid     *pVgid = &pVgroup->vnodeGid[i];
       SMonVnodeDesc *pVnDesc = &desc.vnodes[i];
       pVnDesc->dnode_id = pVgid->dnodeId;
       tstrncpy(pVnDesc->vnode_role, syncStr(pVgid->role), sizeof(pVnDesc->vnode_role));

source/dnode/mnode/impl/src/mndTrans.c

@@ -22,8 +22,8 @@
 #include "mndSync.h"
 #include "mndUser.h"
 
-#define TRANS_VER_NUMBER   1
-#define TRANS_ARRAY_SIZE   8
+#define TRANS_VER_NUMBER 1
+#define TRANS_ARRAY_SIZE 8
 #define TRANS_RESERVE_SIZE 64
 
 static SSdbRaw *mndTransActionEncode(STrans *pTrans);
@@ -896,7 +896,7 @@ static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray)
     pAction->rawWritten = 0;
     pAction->msgSent = 0;
     pAction->msgReceived = 0;
-    if (pAction->errCode == TSDB_CODE_RPC_REDIRECT || pAction->errCode == TSDB_CODE_SYN_NOT_IN_NEW_CONFIG ||
+    if (pAction->errCode == TSDB_CODE_RPC_REDIRECT || pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR ||
         pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR || pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
       pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
       mDebug("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -119,7 +119,7 @@ static int32_t vnodeProcessAlterReplicaReq(SVnode *pVnode, SRpcMsg *pMsg) {
 }
 
 void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
-  SVnode * pVnode = pInfo->ahandle;
+  SVnode  *pVnode = pInfo->ahandle;
   int32_t  vgId = pVnode->config.vgId;
   int32_t  code = 0;
   SRpcMsg *pMsg = NULL;
@@ -174,7 +174,7 @@ void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
 }
 
 void vnodeApplyMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
-  SVnode * pVnode = pInfo->ahandle;
+  SVnode  *pVnode = pInfo->ahandle;
   int32_t  vgId = pVnode->config.vgId;
   int32_t  code = 0;
   SRpcMsg *pMsg = NULL;
@@ -211,21 +211,23 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
     SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
     assert(pSyncNode != NULL);
 
-    ESyncState state = syncGetMyRole(pVnode->sync);
-    SyncTerm   currentTerm = syncGetMyTerm(pVnode->sync);
-
     SMsgHead *pHead = pMsg->pCont;
+    STraceId *trace = &pMsg->info.traceId;
 
-    char  logBuf[512] = {0};
-    char *syncNodeStr = sync2SimpleStr(pVnode->sync);
-    snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
-    static int64_t vndTick = 0;
-    STraceId *     trace = &pMsg->info.traceId;
-    if (++vndTick % 10 == 1) {
-      vGTrace("sync trace msg:%s, %s", TMSG_INFO(pMsg->msgType), syncNodeStr);
-    }
-    syncRpcMsgLog2(logBuf, pMsg);
-    taosMemoryFree(syncNodeStr);
+    do {
+      char          *syncNodeStr = sync2SimpleStr(pVnode->sync);
+      static int64_t vndTick = 0;
+      if (++vndTick % 10 == 1) {
+        vGTrace("vgId:%d, sync heartbeat msg:%s, %s", syncGetVgId(pVnode->sync), TMSG_INFO(pMsg->msgType), syncNodeStr);
+      }
+      if (gRaftDetailLog) {
+        char logBuf[512] = {0};
+        snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType,
+                 syncNodeStr);
+        syncRpcMsgLog2(logBuf, pMsg);
+      }
+      taosMemoryFree(syncNodeStr);
+    } while (0);
 
     SRpcMsg *pRpcMsg = pMsg;
 
@@ -348,7 +350,7 @@ static void vnodeSyncReconfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReCon
 }
 
 static void vnodeSyncCommitMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
-  SVnode *  pVnode = pFsm->data;
+  SVnode   *pVnode = pFsm->data;
   SSnapshot snapshot = {0};
   SyncIndex beginIndex = SYNC_INDEX_INVALID;
   char      logBuf[256] = {0};

source/libs/sync/inc/syncInt.h

@@ -159,7 +159,8 @@ typedef struct SSyncNode {
   SSyncSnapshotSender*   senders[TSDB_MAX_REPLICA];
   SSyncSnapshotReceiver* pNewNodeReceiver;
 
-  // SSnapshotMeta sMeta;
+  // is config changing
+  bool changing;
 
 } SSyncNode;
 
@@ -198,7 +199,7 @@ char*   syncNode2Str(const SSyncNode* pSyncNode);
 void    syncNodeEventLog(const SSyncNode* pSyncNode, char* str);
 char*   syncNode2SimpleStr(const SSyncNode* pSyncNode);
 bool    syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config);
-void    syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex, bool* isDrop);
+void    syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex);
 
 SSyncNode* syncNodeAcquire(int64_t rid);
 void       syncNodeRelease(SSyncNode* pNode);
@@ -238,12 +239,15 @@ int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg);
 bool                 syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId);
 SSyncSnapshotSender* syncNodeGetSnapshotSender(SSyncNode* ths, SRaftId* pDestId);
 
-int32_t     syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta);
-int32_t     syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta);
+int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta);
+int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta);
 
 void syncStartNormal(int64_t rid);
 void syncStartStandBy(int64_t rid);
 
+bool syncNodeCanChange(SSyncNode* pSyncNode);
+bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg);
+
 // for debug --------------
 void syncNodePrint(SSyncNode* pObj);
 void syncNodePrint2(char* s, SSyncNode* pObj);

source/libs/sync/inc/syncRaftCfg.h

@@ -49,14 +49,14 @@ int32_t   raftCfgClose(SRaftCfg *pRaftCfg);
 int32_t   raftCfgPersist(SRaftCfg *pRaftCfg);
 int32_t   raftCfgAddConfigIndex(SRaftCfg *pRaftCfg, SyncIndex configIndex);
 
-cJSON  *syncCfg2Json(SSyncCfg *pSyncCfg);
-char   *syncCfg2Str(SSyncCfg *pSyncCfg);
-char   *syncCfg2SimpleStr(SSyncCfg *pSyncCfg);
+cJSON * syncCfg2Json(SSyncCfg *pSyncCfg);
+char *  syncCfg2Str(SSyncCfg *pSyncCfg);
+char *  syncCfg2SimpleStr(SSyncCfg *pSyncCfg);
 int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg);
 int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg);
 
-cJSON  *raftCfg2Json(SRaftCfg *pRaftCfg);
-char   *raftCfg2Str(SRaftCfg *pRaftCfg);
+cJSON * raftCfg2Json(SRaftCfg *pRaftCfg);
+char *  raftCfg2Str(SRaftCfg *pRaftCfg);
 int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg);
 int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg);
 

source/libs/sync/inc/syncSnapshot.h

@@ -66,10 +66,11 @@ char                *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char
 typedef struct SSyncSnapshotReceiver {
   bool start;
 
-  int32_t  ack;
-  void    *pWriter;
-  SyncTerm term;
-  SyncTerm privateTerm;
+  int32_t   ack;
+  void     *pWriter;
+  SyncTerm  term;
+  SyncTerm  privateTerm;
+  SSnapshot snapshot;
 
   SSyncNode *pSyncNode;
   SRaftId    fromId;
@@ -78,12 +79,12 @@ typedef struct SSyncSnapshotReceiver {
 
 SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId);
 void                   snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
-void                   snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SRaftId fromId);
-bool                   snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
-void                   snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver, bool apply);
-cJSON                 *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
-char                  *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
-char                  *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event);
+void   snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg);
+bool   snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
+void   snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver, bool apply);
+cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
+char  *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
+char  *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event);
 
 int32_t syncNodeOnSnapshotSendCb(SSyncNode *ths, SyncSnapshotSend *pMsg);
 int32_t syncNodeOnSnapshotRspCb(SSyncNode *ths, SyncSnapshotRsp *pMsg);

source/libs/sync/src/syncMessage.c

@@ -2227,4 +2227,133 @@ void syncLeaderTransferLog2(char* s, const SyncLeaderTransfer* pMsg) {
     sTrace("syncLeaderTransferLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
     taosMemoryFree(serialized);
   }
+}
+
+// ---------------------------------------------
+SyncReconfigFinish* syncReconfigFinishBuild(int32_t vgId) {
+  uint32_t            bytes = sizeof(SyncReconfigFinish);
+  SyncReconfigFinish* pMsg = taosMemoryMalloc(bytes);
+  memset(pMsg, 0, bytes);
+  pMsg->bytes = bytes;
+  pMsg->vgId = vgId;
+  pMsg->msgType = TDMT_SYNC_CONFIG_CHANGE_FINISH;
+  return pMsg;
+}
+
+void syncReconfigFinishDestroy(SyncReconfigFinish* pMsg) {
+  if (pMsg != NULL) {
+    taosMemoryFree(pMsg);
+  }
+}
+
+void syncReconfigFinishSerialize(const SyncReconfigFinish* pMsg, char* buf, uint32_t bufLen) {
+  assert(pMsg->bytes <= bufLen);
+  memcpy(buf, pMsg, pMsg->bytes);
+}
+
+void syncReconfigFinishDeserialize(const char* buf, uint32_t len, SyncReconfigFinish* pMsg) {
+  memcpy(pMsg, buf, len);
+  assert(len == pMsg->bytes);
+}
+
+char* syncReconfigFinishSerialize2(const SyncReconfigFinish* pMsg, uint32_t* len) {
+  char* buf = taosMemoryMalloc(pMsg->bytes);
+  assert(buf != NULL);
+  syncReconfigFinishSerialize(pMsg, buf, pMsg->bytes);
+  if (len != NULL) {
+    *len = pMsg->bytes;
+  }
+  return buf;
+}
+
+SyncReconfigFinish* syncReconfigFinishDeserialize2(const char* buf, uint32_t len) {
+  uint32_t            bytes = *((uint32_t*)buf);
+  SyncReconfigFinish* pMsg = taosMemoryMalloc(bytes);
+  assert(pMsg != NULL);
+  syncReconfigFinishDeserialize(buf, len, pMsg);
+  assert(len == pMsg->bytes);
+  return pMsg;
+}
+
+void syncReconfigFinish2RpcMsg(const SyncReconfigFinish* pMsg, SRpcMsg* pRpcMsg) {
+  memset(pRpcMsg, 0, sizeof(*pRpcMsg));
+  pRpcMsg->msgType = pMsg->msgType;
+  pRpcMsg->contLen = pMsg->bytes;
+  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
+  syncReconfigFinishSerialize(pMsg, pRpcMsg->pCont, pRpcMsg->contLen);
+}
+
+void syncReconfigFinishFromRpcMsg(const SRpcMsg* pRpcMsg, SyncReconfigFinish* pMsg) {
+  syncReconfigFinishDeserialize(pRpcMsg->pCont, pRpcMsg->contLen, pMsg);
+}
+
+SyncReconfigFinish* syncReconfigFinishFromRpcMsg2(const SRpcMsg* pRpcMsg) {
+  SyncReconfigFinish* pMsg = syncReconfigFinishDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
+  assert(pMsg != NULL);
+  return pMsg;
+}
+
+cJSON* syncReconfigFinish2Json(const SyncReconfigFinish* pMsg) {
+  char   u64buf[128];
+  cJSON* pRoot = cJSON_CreateObject();
+
+  if (pMsg != NULL) {
+    cJSON_AddNumberToObject(pRoot, "bytes", pMsg->bytes);
+    cJSON_AddNumberToObject(pRoot, "vgId", pMsg->vgId);
+    cJSON_AddNumberToObject(pRoot, "msgType", pMsg->msgType);
+
+    cJSON* pOldCfg = syncCfg2Json((SSyncCfg*)(&(pMsg->oldCfg)));
+    cJSON* pNewCfg = syncCfg2Json((SSyncCfg*)(&(pMsg->newCfg)));
+    cJSON_AddItemToObject(pRoot, "oldCfg", pOldCfg);
+    cJSON_AddItemToObject(pRoot, "newCfg", pNewCfg);
+
+    snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->newCfgIndex);
+    cJSON_AddStringToObject(pRoot, "newCfgIndex", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->newCfgTerm);
+    cJSON_AddStringToObject(pRoot, "newCfgTerm", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->newCfgSeqNum);
+    cJSON_AddStringToObject(pRoot, "newCfgSeqNum", u64buf);
+  }
+
+  cJSON* pJson = cJSON_CreateObject();
+  cJSON_AddItemToObject(pJson, "SyncReconfigFinish", pRoot);
+  return pJson;
+}
+
+char* syncReconfigFinish2Str(const SyncReconfigFinish* pMsg) {
+  cJSON* pJson = syncReconfigFinish2Json(pMsg);
+  char*  serialized = cJSON_Print(pJson);
+  cJSON_Delete(pJson);
+  return serialized;
+}
+
+// for debug ----------------------
+void syncReconfigFinishPrint(const SyncReconfigFinish* pMsg) {
+  char* serialized = syncReconfigFinish2Str(pMsg);
+  printf("syncReconfigFinishPrint | len:%lu | %s \n", strlen(serialized), serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncReconfigFinishPrint2(char* s, const SyncReconfigFinish* pMsg) {
+  char* serialized = syncReconfigFinish2Str(pMsg);
+  printf("syncReconfigFinishPrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
+  fflush(NULL);
+  taosMemoryFree(serialized);
+}
+
+void syncReconfigFinishLog(const SyncReconfigFinish* pMsg) {
+  char* serialized = syncReconfigFinish2Str(pMsg);
+  sTrace("syncReconfigFinishLog | len:%lu | %s", strlen(serialized), serialized);
+  taosMemoryFree(serialized);
+}
+
+void syncReconfigFinishLog2(char* s, const SyncReconfigFinish* pMsg) {
+  if (gRaftDetailLog) {
+    char* serialized = syncReconfigFinish2Str(pMsg);
+    sTrace("syncReconfigFinishLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
+    taosMemoryFree(serialized);
+  }
 }
\ No newline at end of file

source/libs/sync/src/syncRaftCfg.c

@@ -96,14 +96,14 @@ cJSON *syncCfg2Json(SSyncCfg *pSyncCfg) {
 
 char *syncCfg2Str(SSyncCfg *pSyncCfg) {
   cJSON *pJson = syncCfg2Json(pSyncCfg);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
 
 char *syncCfg2SimpleStr(SSyncCfg *pSyncCfg) {
   int32_t len = 512;
-  char   *s = taosMemoryMalloc(len);
+  char *  s = taosMemoryMalloc(len);
   memset(s, 0, len);
 
   snprintf(s, len, "{replica-num:%d, my-index:%d, ", pSyncCfg->replicaNum, pSyncCfg->myIndex);
@@ -196,7 +196,7 @@ cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) {
 
 char *raftCfg2Str(SRaftCfg *pRaftCfg) {
   cJSON *pJson = raftCfg2Json(pRaftCfg);
-  char  *serialized = cJSON_Print(pJson);
+  char * serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -262,7 +262,7 @@ int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg) {
     (pRaftCfg->configIndexArr)[i] = atoll(pIndex->valuestring);
   }
 
-  cJSON  *pJsonSyncCfg = cJSON_GetObjectItem(pJson, "SSyncCfg");
+  cJSON * pJsonSyncCfg = cJSON_GetObjectItem(pJson, "SSyncCfg");
   int32_t code = syncCfgFromJson(pJsonSyncCfg, &(pRaftCfg->cfg));
   ASSERT(code == 0);
 

source/libs/sync/src/syncSnapshot.c

@@ -21,7 +21,8 @@
 #include "syncUtil.h"
 #include "wal.h"
 
-static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SRaftId fromId);
+static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm,
+                                    SyncSnapshotSend *pBeginMsg);
 
 //----------------------------------
 SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex) {
@@ -341,6 +342,10 @@ SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId from
     pReceiver->fromId = fromId;
     pReceiver->term = pSyncNode->pRaftStore->currentTerm;
     pReceiver->privateTerm = 0;
+    pReceiver->snapshot.data = NULL;
+    pReceiver->snapshot.lastApplyIndex = -1;
+    pReceiver->snapshot.lastApplyTerm = 0;
+    pReceiver->snapshot.lastConfigIndex = -1;
 
   } else {
     sInfo("snapshotReceiverCreate cannot create receiver");
@@ -358,11 +363,16 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
 bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceiver->start; }
 
 // begin receive snapshot msg (current term, seq begin)
-static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SRaftId fromId) {
+static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm,
+                                    SyncSnapshotSend *pBeginMsg) {
   pReceiver->term = pReceiver->pSyncNode->pRaftStore->currentTerm;
   pReceiver->privateTerm = privateTerm;
   pReceiver->ack = SYNC_SNAPSHOT_SEQ_BEGIN;
-  pReceiver->fromId = fromId;
+  pReceiver->fromId = pBeginMsg->srcId;
+
+  pReceiver->snapshot.lastApplyIndex = pBeginMsg->lastIndex;
+  pReceiver->snapshot.lastApplyTerm = pBeginMsg->lastTerm;
+  pReceiver->snapshot.lastConfigIndex = pBeginMsg->lastConfigIndex;
 
   ASSERT(pReceiver->pWriter == NULL);
   int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStartWrite(pReceiver->pSyncNode->pFsm, &(pReceiver->pWriter));
@@ -371,10 +381,10 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm p
 
 // if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver
 // if already start, force close, start again
-void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SRaftId fromId) {
+void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg) {
   if (!snapshotReceiverIsStart(pReceiver)) {
     // start
-    snapshotReceiverDoStart(pReceiver, privateTerm, fromId);
+    snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
     pReceiver->start = true;
 
   } else {
@@ -388,7 +398,7 @@ void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTer
     pReceiver->pWriter = NULL;
 
     // start again
-    snapshotReceiverDoStart(pReceiver, privateTerm, fromId);
+    snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
     pReceiver->start = true;
   }
 
@@ -449,6 +459,15 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
     cJSON_AddNumberToObject(pFromId, "vgId", pReceiver->fromId.vgId);
     cJSON_AddItemToObject(pRoot, "fromId", pFromId);
 
+    snprintf(u64buf, sizeof(u64buf), "%lu", pReceiver->snapshot.lastApplyIndex);
+    cJSON_AddStringToObject(pRoot, "snapshot.lastApplyIndex", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pReceiver->snapshot.lastApplyTerm);
+    cJSON_AddStringToObject(pRoot, "snapshot.lastApplyTerm", u64buf);
+
+    snprintf(u64buf, sizeof(u64buf), "%lu", pReceiver->snapshot.lastConfigIndex);
+    cJSON_AddStringToObject(pRoot, "snapshot.lastConfigIndex", u64buf);
+
     snprintf(u64buf, sizeof(u64buf), "%lu", pReceiver->term);
     cJSON_AddStringToObject(pRoot, "term", u64buf);
 
@@ -477,8 +496,9 @@ char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event)
   uint16_t port;
   syncUtilU642Addr(fromId.addr, host, sizeof(host), &port);
 
-  snprintf(s, len, "%s %p start:%d ack:%d term:%lu pterm:%lu %s:%d ", event, pReceiver, pReceiver->start,
-           pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port);
+  snprintf(s, len, "%s %p start:%d ack:%d term:%lu pterm:%lu from:%s:%d laindex:%ld laterm:%lu lcindex:%ld", event,
+           pReceiver, pReceiver->start, pReceiver->ack, pReceiver->term, pReceiver->privateTerm, host, port,
+           pReceiver->snapshot.lastApplyIndex, pReceiver->snapshot.lastApplyTerm, pReceiver->snapshot.lastConfigIndex);
 
   return s;
 }
@@ -495,7 +515,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
     if (pMsg->term == pSyncNode->pRaftStore->currentTerm) {
       if (pMsg->seq == SYNC_SNAPSHOT_SEQ_BEGIN) {
         // begin
-        snapshotReceiverStart(pReceiver, pMsg->privateTerm, pMsg->srcId);
+        snapshotReceiverStart(pReceiver, pMsg->privateTerm, pMsg);
         pReceiver->ack = pMsg->seq;
         needRsp = true;
 
@@ -519,42 +539,9 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
           // update new config myIndex
           SSyncCfg newSyncCfg = pMsg->lastConfig;
           syncNodeUpdateNewConfigIndex(pSyncNode, &newSyncCfg);
-          bool IamInNew = syncNodeInConfig(pSyncNode, &newSyncCfg);
-
-          bool isDrop = false;
-          if (IamInNew) {
-            char eventLog[128];
-            snprintf(eventLog, sizeof(eventLog),
-                     "update config by snapshot, lastIndex:%ld, lastTerm:%lu, lastConfigIndex:%ld", pMsg->lastIndex,
-                     pMsg->lastTerm, pMsg->lastConfigIndex);
-            syncNodeEventLog(pSyncNode, eventLog);
-
-            syncNodeUpdateConfig(pSyncNode, &newSyncCfg, pMsg->lastConfigIndex, &isDrop);
-
-          } else {
-            char eventLog[128];
-            snprintf(eventLog, sizeof(eventLog),
-                     "do not update config by snapshot, not in new, lastIndex:%ld, lastTerm:%lu, lastConfigIndex:%ld",
-                     pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex);
-            syncNodeEventLog(pSyncNode, eventLog);
-          }
-
-          // change isStandBy to normal
-          if (!isDrop) {
-            char  tmpbuf[512];
-            char *oldStr = syncCfg2SimpleStr(&oldSyncCfg);
-            char *newStr = syncCfg2SimpleStr(&newSyncCfg);
-            snprintf(tmpbuf, sizeof(tmpbuf), "config change3 from %d to %d, index:%ld, %s  -->  %s",
-                     oldSyncCfg.replicaNum, newSyncCfg.replicaNum, pMsg->lastConfigIndex, oldStr, newStr);
-            taosMemoryFree(oldStr);
-            taosMemoryFree(newStr);
-
-            if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
-              syncNodeBecomeLeader(pSyncNode, tmpbuf);
-            } else {
-              syncNodeBecomeFollower(pSyncNode, tmpbuf);
-            }
-          }
+
+          // do config change
+          syncNodeDoConfigChange(pSyncNode, &newSyncCfg, pMsg->lastConfigIndex);
         }
 
         SSnapshot snapshot;

source/libs/sync/src/syncUtil.c

@@ -261,23 +261,29 @@ bool syncUtilIsData(tmsg_t msgType) {
 #endif
 
 bool syncUtilUserPreCommit(tmsg_t msgType) {
-  if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_LEADER_TRANSFER) {
+  if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_CONFIG_CHANGE_FINISH &&
+      msgType != TDMT_SYNC_LEADER_TRANSFER) {
     return true;
   }
+
   return false;
 }
 
 bool syncUtilUserCommit(tmsg_t msgType) {
-  if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_LEADER_TRANSFER) {
+  if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_CONFIG_CHANGE_FINISH &&
+      msgType != TDMT_SYNC_LEADER_TRANSFER) {
     return true;
   }
+
   return false;
 }
 
 bool syncUtilUserRollback(tmsg_t msgType) {
-  if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_LEADER_TRANSFER) {
+  if (msgType != TDMT_SYNC_NOOP && msgType != TDMT_SYNC_CONFIG_CHANGE && msgType != TDMT_SYNC_CONFIG_CHANGE_FINISH &&
+      msgType != TDMT_SYNC_LEADER_TRANSFER) {
     return true;
   }
+
   return false;
 }
 

source/libs/sync/test/CMakeLists.txt

@@ -49,6 +49,7 @@ add_executable(syncRaftLogTest "")
 add_executable(syncRaftLogTest2 "")
 add_executable(syncRaftLogTest3 "")
 add_executable(syncLeaderTransferTest "")
+add_executable(syncReconfigFinishTest "")
 
 
 target_sources(syncTest
@@ -255,6 +256,10 @@ target_sources(syncLeaderTransferTest
     PRIVATE
     "syncLeaderTransferTest.cpp"
 )
+target_sources(syncReconfigFinishTest
+    PRIVATE
+    "syncReconfigFinishTest.cpp"
+)
 
 
 target_include_directories(syncTest
@@ -512,6 +517,11 @@ target_include_directories(syncLeaderTransferTest
     "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
+target_include_directories(syncReconfigFinishTest
+    PUBLIC
+    "${TD_SOURCE_DIR}/include/libs/sync"
+    "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
+)
 
 
 target_link_libraries(syncTest
@@ -718,6 +728,10 @@ target_link_libraries(syncLeaderTransferTest
     sync
     gtest_main
 )
+target_link_libraries(syncReconfigFinishTest
+    sync
+    gtest_main
+)
 
 
 enable_testing()

source/libs/sync/test/syncConfigChangeSnapshotTest.cpp

@@ -147,8 +147,8 @@ int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_
 void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
 
 void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta) {
-  sTrace("==callback== ==ReConfigCb== flag:0x%lX, isDrop:%d, index:%ld, code:%d, currentTerm:%lu, term:%lu",
-         cbMeta.flag, cbMeta.isDrop, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
+  sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%ld, code:%d, currentTerm:%lu, term:%lu", cbMeta.flag,
+         cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
 }
 
 SSyncFSM* createFsm() {

source/libs/sync/test/syncConfigChangeTest.cpp

@@ -78,8 +78,8 @@ int32_t GetSnapshotCb(struct SSyncFSM* pFsm, SSnapshot* pSnapshot) {
 void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
 
 void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta) {
-  sTrace("==callback== ==ReConfigCb== flag:0x%lX, isDrop:%d, index:%ld, code:%d, currentTerm:%lu, term:%lu",
-         cbMeta.flag, cbMeta.isDrop, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
+  sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%ld, code:%d, currentTerm:%lu, term:%lu", cbMeta.flag,
+         cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term);
 }
 
 SSyncFSM* createFsm() {

source/libs/sync/test/syncRaftCfgTest.cpp

@@ -137,6 +137,6 @@ int main() {
   test3();
   test4();
   test5();
-  
+
   return 0;
 }

source/libs/sync/test/syncReconfigFinishTest.cpp

+#include <gtest/gtest.h>
+#include <stdio.h>
+#include "syncIO.h"
+#include "syncInt.h"
+#include "syncMessage.h"
+#include "syncUtil.h"
+
+void logTest() {
+  sTrace("--- sync log test: trace");
+  sDebug("--- sync log test: debug");
+  sInfo("--- sync log test: info");
+  sWarn("--- sync log test: warn");
+  sError("--- sync log test: error");
+  sFatal("--- sync log test: fatal");
+}
+
+SSyncCfg* createSyncOldCfg() {
+  SSyncCfg* pCfg = (SSyncCfg*)taosMemoryMalloc(sizeof(SSyncCfg));
+  memset(pCfg, 0, sizeof(SSyncCfg));
+
+  pCfg->replicaNum = 3;
+  pCfg->myIndex = 1;
+  for (int i = 0; i < pCfg->replicaNum; ++i) {
+    ((pCfg->nodeInfo)[i]).nodePort = i * 100;
+    snprintf(((pCfg->nodeInfo)[i]).nodeFqdn, sizeof(((pCfg->nodeInfo)[i]).nodeFqdn), "100.200.300.%d", i);
+  }
+
+  return pCfg;
+}
+
+SSyncCfg* createSyncNewCfg() {
+  SSyncCfg* pCfg = (SSyncCfg*)taosMemoryMalloc(sizeof(SSyncCfg));
+  memset(pCfg, 0, sizeof(SSyncCfg));
+
+  pCfg->replicaNum = 3;
+  pCfg->myIndex = 1;
+  for (int i = 0; i < pCfg->replicaNum; ++i) {
+    ((pCfg->nodeInfo)[i]).nodePort = i * 100;
+    snprintf(((pCfg->nodeInfo)[i]).nodeFqdn, sizeof(((pCfg->nodeInfo)[i]).nodeFqdn), "500.600.700.%d", i);
+  }
+
+  return pCfg;
+}
+
+SyncReconfigFinish *createMsg() {
+  SyncReconfigFinish *pMsg = syncReconfigFinishBuild(1234);
+ 
+  SSyncCfg* pOld = createSyncOldCfg();
+  SSyncCfg* pNew = createSyncNewCfg();
+  pMsg->oldCfg = *pOld;
+  pMsg->newCfg = *pNew;
+
+  pMsg->newCfgIndex = 11;
+  pMsg->newCfgTerm = 22;
+  pMsg->newCfgSeqNum = 33;
+
+  taosMemoryFree(pOld);
+  taosMemoryFree(pNew);
+
+  return pMsg;
+}
+
+
+void test1() {
+  SyncReconfigFinish *pMsg = createMsg();
+  syncReconfigFinishLog2((char *)"test1:", pMsg);
+  syncReconfigFinishDestroy(pMsg);
+}
+
+
+void test2() {
+  SyncReconfigFinish *pMsg = createMsg();
+  uint32_t         len = pMsg->bytes;
+  char *           serialized = (char *)taosMemoryMalloc(len);
+  syncReconfigFinishSerialize(pMsg, serialized, len);
+  SyncReconfigFinish *pMsg2 = syncReconfigFinishBuild(1000);
+  syncReconfigFinishDeserialize(serialized, len, pMsg2);
+  syncReconfigFinishLog2((char *)"test2: syncReconfigFinishSerialize -> syncReconfigFinishDeserialize ", pMsg2);
+
+  taosMemoryFree(serialized);
+  syncReconfigFinishDestroy(pMsg);
+  syncReconfigFinishDestroy(pMsg2);
+}
+
+void test3() {
+  SyncReconfigFinish *pMsg = createMsg();
+  uint32_t         len;
+  char *           serialized = syncReconfigFinishSerialize2(pMsg, &len);
+  SyncReconfigFinish *pMsg2 = syncReconfigFinishDeserialize2(serialized, len);
+  syncReconfigFinishLog2((char *)"test3: SyncReconfigFinishSerialize2 -> syncReconfigFinishDeserialize2 ", pMsg2);
+
+  taosMemoryFree(serialized);
+  syncReconfigFinishDestroy(pMsg);
+  syncReconfigFinishDestroy(pMsg2);
+}
+
+void test4() {
+  SyncReconfigFinish *pMsg = createMsg();
+  SRpcMsg          rpcMsg;
+  syncReconfigFinish2RpcMsg(pMsg, &rpcMsg);
+  SyncReconfigFinish *pMsg2 = (SyncReconfigFinish *)taosMemoryMalloc(rpcMsg.contLen);
+  syncReconfigFinishFromRpcMsg(&rpcMsg, pMsg2);
+  syncReconfigFinishLog2((char *)"test4: syncReconfigFinish2RpcMsg -> syncReconfigFinishFromRpcMsg ", pMsg2);
+
+  rpcFreeCont(rpcMsg.pCont);
+  syncReconfigFinishDestroy(pMsg);
+  syncReconfigFinishDestroy(pMsg2);
+}
+
+void test5() {
+  SyncReconfigFinish *pMsg = createMsg();
+  SRpcMsg          rpcMsg;
+  syncReconfigFinish2RpcMsg(pMsg, &rpcMsg);
+  SyncReconfigFinish *pMsg2 = syncReconfigFinishFromRpcMsg2(&rpcMsg);
+  syncReconfigFinishLog2((char *)"test5: syncReconfigFinish2RpcMsg -> syncReconfigFinishFromRpcMsg2 ", pMsg2);
+
+  rpcFreeCont(rpcMsg.pCont);
+  syncReconfigFinishDestroy(pMsg);
+  syncReconfigFinishDestroy(pMsg2);
+}
+
+int main() {
+  gRaftDetailLog = true;
+  tsAsyncLog = 0;
+  sDebugFlag = DEBUG_TRACE + DEBUG_SCREEN + DEBUG_FILE;
+  logTest();
+
+  test1();
+  test2();
+  test3();
+  test4();
+  test5();
+
+  return 0;
+}

source/libs/sync/test/syncTestTool.cpp

@@ -148,8 +148,8 @@ void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFini
 
 void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta) {
   char* s = syncCfg2Str(&(cbMeta.newCfg));
-  sTrace("==callback== ==ReConfigCb== flag:0x%lX, isDrop:%d, index:%ld, code:%d, currentTerm:%lu, term:%lu, newCfg:%s",
-         cbMeta.flag, cbMeta.isDrop, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term, s);
+  sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%ld, code:%d, currentTerm:%lu, term:%lu, newCfg:%s",
+         cbMeta.flag, cbMeta.index, cbMeta.code, cbMeta.currentTerm, cbMeta.term, s);
   taosMemoryFree(s);
 }
 

source/util/src/terror.c

@@ -423,6 +423,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_IS_LEADER,                "Sync is leader")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_LEADER,               "Sync not leader")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_ONE_REPLICA,              "Sync one replica")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_IN_NEW_CONFIG,        "Sync not in new config")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NEW_CONFIG_ERROR,         "Sync new config error")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_RECONFIG_NOT_READY,       "Sync not ready for reconfig")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_PROPOSE_NOT_READY,        "Sync not ready for propose")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR,           "Sync internal error")
 
 // wal