cherry-pick from 1.0 for checkpoint

src/kvstore/NebulaStore.cpp

@@ -770,23 +770,21 @@ ResultCode NebulaStore::setWriteBlocking(GraphSpaceID spaceId, bool sign) {
                 return error(partRet);
             }
             auto p = nebula::value(partRet);
-            if (p->isLeader()) {
-                auto ret = ResultCode::SUCCEEDED;
-                p->setBlocking(sign);
-                if (sign) {
-                    folly::Baton<true, std::atomic> baton;
-                    p->sync([&ret, &baton] (kvstore::ResultCode code) {
-                        if (kvstore::ResultCode::SUCCEEDED != code) {
-                            ret = code;
-                        }
-                        baton.post();
-                    });
-                    baton.wait();
-                }
-                if (ret != ResultCode::SUCCEEDED) {
-                    LOG(ERROR) << "Part sync failed. space : " << spaceId << " Part : " << part;
-                    return ret;
-                }
+            auto ret = ResultCode::SUCCEEDED;
+            p->setBlocking(sign);
+            if (sign) {
+                folly::Baton<true, std::atomic> baton;
+                p->sync([&ret, &baton] (kvstore::ResultCode code) {
+                    if (kvstore::ResultCode::SUCCEEDED != code) {
+                        ret = code;
+                    }
+                    baton.post();
+                });
+                baton.wait();
+            }
+            if (ret != ResultCode::SUCCEEDED) {
+                 LOG(ERROR) << "Part sync failed. space : " << spaceId << " Part : " << part;
+                 return ret;
             }
         }
     }

src/kvstore/RocksEngine.cpp

@@ -432,9 +432,10 @@ ResultCode RocksEngine::createCheckpoint(const std::string& name) {
     auto checkpointPath = folly::stringPrintf("%s/checkpoints/%s/data",
                                               dataPath_.c_str(), name.c_str());
     LOG(INFO) << "Target checkpoint path : " << checkpointPath;
-    if (fs::FileUtils::exist(checkpointPath)) {
-        LOG(ERROR) << "The snapshot file already exists: " << checkpointPath;
-        return ResultCode::ERR_CHECKPOINT_ERROR;
+    if (fs::FileUtils::exist(checkpointPath) &&
+        !fs::FileUtils::remove(checkpointPath.data(), true)) {
+        LOG(ERROR) << "Remove exist dir failed of checkpoint : " << checkpointPath;
+        return ResultCode::ERR_IO_ERROR;
     }
 
     auto parent = checkpointPath.substr(0, checkpointPath.rfind('/'));

src/kvstore/raftex/RaftPart.cpp

@@ -556,8 +556,11 @@ folly::Future<AppendLogResult> RaftPart::appendLogAsync(ClusterID source,
                                                         LogType logType,
                                                         std::string log,
                                                         AtomicOp op) {
-    if (blocking_ && (logType == LogType::NORMAL || logType == LogType::ATOMIC_OP)) {
-        return AppendLogResult::E_WRITE_BLOCKING;
+    if (blocking_) {
+        // No need to block heartbeats and empty log.
+         if ((logType == LogType::NORMAL && !log.empty()) || logType == LogType::ATOMIC_OP) {
+             return AppendLogResult::E_WRITE_BLOCKING;
+         }
     }
 
     LogCache swappedOutLogs;

src/kvstore/wal/FileBasedWal.cpp

@@ -603,6 +603,11 @@ bool FileBasedWal::linkCurrentWAL(const char* newPath) {
         LOG(INFO) << idStr_ << "No wal files found, skip link";
         return true;
     }
+    if (fs::FileUtils::exist(newPath) &&
+        !fs::FileUtils::remove(newPath, true)) {
+        LOG(ERROR) << "Remove exist dir failed of wal : " << newPath;
+        return false;
+     }
     if (!fs::FileUtils::makeDir(newPath)) {
         LOG(INFO) << idStr_ << "Link file parent dir make failed : " << newPath;
         return false;

src/meta/processors/admin/AdminClient.cpp

@@ -599,14 +599,10 @@ folly::Future<Status> AdminClient::createSnapshot(GraphSpaceID spaceId,
     folly::Promise<Status> pro;
     auto f = pro.getFuture();
 
-    /**
-     * Don't need retry.
-     * Because existing checkpoint directories leads to fail again.
-     **/
     getResponse({Utils::getAdminAddrFromStoreAddr(host)}, 0, std::move(req),
             [] (auto client, auto request) {
         return client->future_createCheckpoint(request);
-    }, 0, std::move(pro), 0);
+    }, 0, std::move(pro), 3 /*The snapshot operation need to retry 3 times*/);
     return f;
 }
 
@@ -626,7 +622,7 @@ folly::Future<Status> AdminClient::dropSnapshot(GraphSpaceID spaceId,
     getResponse({Utils::getAdminAddrFromStoreAddr(host)}, 0, std::move(req),
             [] (auto client, auto request) {
         return client->future_dropCheckpoint(request);
-    }, 0, std::move(pro), 1 /*The snapshot operation only needs to be retried twice*/);
+    }, 0, std::move(pro), 3 /*The snapshot operation need to retry 3 times*/);
     return f;
 }
 
@@ -641,7 +637,7 @@ folly::Future<Status> AdminClient::blockingWrites(GraphSpaceID spaceId,
     getResponse({Utils::getAdminAddrFromStoreAddr(host)}, 0, std::move(req),
             [] (auto client, auto request) {
         return client->future_blockingWrites(request);
-    }, 0, std::move(pro), 1 /*The blocking needs to be retried twice*/);
+    }, 0, std::move(pro), 32 /*The blocking need to retry 32 times*/);
     return f;
 }
 

src/meta/processors/admin/CreateSnapshotProcessor.cpp

@@ -13,7 +13,6 @@ namespace nebula {
 namespace meta {
 
 void CreateSnapshotProcessor::process(const cpp2::CreateSnapshotReq&) {
-    folly::SharedMutex::ReadHolder rHolder(LockUtils::spaceLock());
     // check the index rebuild. not allowed to create snapshot when index rebuilding.
     auto prefix = MetaServiceUtils::rebuildIndexStatusPrefix();
     std::unique_ptr<kvstore::KVIterator> iter;

src/meta/processors/listenerMan/ListenerProcessor.cpp

@@ -82,7 +82,7 @@ void ListListenerProcessor::process(const cpp2::ListListenerReq& req) {
     std::string prefix = MetaServiceUtils::listenerPrefix(space);
     auto ret = kvstore_->prefix(kDefaultSpaceId, kDefaultPartId, prefix, &iter);
     if (ret != kvstore::ResultCode::SUCCEEDED) {
-        LOG(ERROR) << "Can't find any users.";
+        LOG(ERROR) << "Can't find any listener.";
         handleErrorCode(cpp2::ErrorCode::E_NOT_FOUND);
         onFinished();
         return;