From e8391ee3bef1062de98e02737093ff05e7eb1fe9 Mon Sep 17 00:00:00 2001
From: liwenhui-soul <38217397+liwenhui-soul@users.noreply.github.com>
Date: Mon, 21 Mar 2022 16:52:31 +0800
Subject: [PATCH] remove learner parts when removing space (#4040)

---
 src/kvstore/NebulaStore.cpp | 18 ++++++++++++++----
 src/kvstore/NebulaStore.h   |  4 +++-
 src/kvstore/PartManager.h   |  2 +-
 3 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/src/kvstore/NebulaStore.cpp b/src/kvstore/NebulaStore.cpp
index 79b902aa25a..4729a4071e3 100644
--- a/src/kvstore/NebulaStore.cpp
+++ b/src/kvstore/NebulaStore.cpp
@@ -39,12 +39,12 @@ NebulaStore::~NebulaStore() {
   raftService_->stop();
   LOG(INFO) << "Waiting for the raft service stop...";
   raftService_->waitUntilStop();
-  spaces_.clear();
-  spaceListeners_.clear();
   bgWorkers_->stop();
   bgWorkers_->wait();
   storeWorker_->stop();
   storeWorker_->wait();
+  spaces_.clear();
+  spaceListeners_.clear();
   LOG(INFO) << "~NebulaStore()";
 }
 
@@ -437,6 +437,13 @@ void NebulaStore::removeSpace(GraphSpaceID spaceId, bool isListener) {
   if (!isListener) {
     auto spaceIt = this->spaces_.find(spaceId);
     if (spaceIt != this->spaces_.end()) {
+      for (auto& [partId, part] : spaceIt->second->parts_) {
+        // before calling removeSpace, meta client would call removePart to remove all parts in
+        // meta cache, which do not contain learners, so we remove them here
+        if (part->isLearner()) {
+          removePart(spaceId, partId, false);
+        }
+      }
       auto& engines = spaceIt->second->engines_;
       for (auto& engine : engines) {
         auto parts = engine->allParts();
@@ -490,8 +497,11 @@ nebula::cpp2::ErrorCode NebulaStore::clearSpace(GraphSpaceID spaceId) {
   return nebula::cpp2::ErrorCode::SUCCEEDED;
 }
 
-void NebulaStore::removePart(GraphSpaceID spaceId, PartitionID partId) {
-  folly::RWSpinLock::WriteHolder wh(&lock_);
+void NebulaStore::removePart(GraphSpaceID spaceId, PartitionID partId, bool needLock) {
+  folly::RWSpinLock::WriteHolder wh(nullptr);
+  if (needLock) {
+    wh.reset(&lock_);
+  }
   auto spaceIt = this->spaces_.find(spaceId);
   if (spaceIt != this->spaces_.end()) {
     auto partIt = spaceIt->second->parts_.find(partId);
diff --git a/src/kvstore/NebulaStore.h b/src/kvstore/NebulaStore.h
index 0bfbfba53e6..90a64c5a3c3 100644
--- a/src/kvstore/NebulaStore.h
+++ b/src/kvstore/NebulaStore.h
@@ -581,8 +581,10 @@ class NebulaStore : public KVStore, public Handler {
    *
    * @param spaceId
    * @param partId
+   * @param needLock if lock_ has already been locked, we need
+   * to set needLock as false, or we set it as true
    */
-  void removePart(GraphSpaceID spaceId, PartitionID partId) override;
+  void removePart(GraphSpaceID spaceId, PartitionID partId, bool needLock = true) override;
 
   /**
    * @brief Retrive the leader distribution
diff --git a/src/kvstore/PartManager.h b/src/kvstore/PartManager.h
index 20d4575eaf6..f89e143e083 100644
--- a/src/kvstore/PartManager.h
+++ b/src/kvstore/PartManager.h
@@ -77,7 +77,7 @@ class Handler {
    * @param spaceId
    * @param partId
    */
-  virtual void removePart(GraphSpaceID spaceId, PartitionID partId) = 0;
+  virtual void removePart(GraphSpaceID spaceId, PartitionID partId, bool needLock = true) = 0;
 
   /**
    * @brief Add a partition as listener