remove job entries when drop space && check space exist when check jo…

src/meta/processors/job/JobManager.cpp

@@ -66,6 +66,16 @@ JobManager::~JobManager() {
   shutDown();
 }
 
+bool JobManager::spaceExist(GraphSpaceID spaceId) {
+  auto spaceKey = MetaKeyUtils::spaceKey(spaceId);
+  std::string val;
+  auto retCode = kvStore_->get(kDefaultSpaceId, kDefaultPartId, spaceKey, &val);
+  if (retCode == nebula::cpp2::ErrorCode::E_KEY_NOT_FOUND) {
+    return false;
+  }
+  return true;
+}
+
 nebula::cpp2::ErrorCode JobManager::handleRemainingJobs() {
   std::unique_ptr<kvstore::KVIterator> iter;
   auto retCode =
@@ -972,6 +982,7 @@ ErrorOr<nebula::cpp2::ErrorCode, bool> JobManager::checkTypeJobRunning(
     return retCode;
   }
 
+  std::unordered_map<GraphSpaceID, bool> spaceExistCache;
   for (; iter->valid(); iter->next()) {
     auto jobKey = iter->key();
     if (MetaKeyUtils::isJobKey(jobKey)) {
@@ -985,6 +996,21 @@ ErrorOr<nebula::cpp2::ErrorCode, bool> JobManager::checkTypeJobRunning(
         continue;
       }
 
+      // In some corner cases, there maybe job entry exist but the space has been dropped
+      auto spaceId = jobDesc.getSpace();
+      if (spaceExistCache.find(spaceId) != spaceExistCache.end()) {
+        if (!spaceExistCache[spaceId]) {
+          continue;
+        }
+      } else {
+        if (!spaceExist(spaceId)) {
+          spaceExistCache[spaceId] = false;
+          continue;
+        } else {
+          spaceExistCache[spaceId] = true;
+        }
+      }
+
       auto status = jobDesc.getStatus();
       if (status == cpp2::JobStatus::QUEUE || status == cpp2::JobStatus::RUNNING) {
         return true;

src/meta/processors/job/JobManager.h

@@ -300,6 +300,13 @@ class JobManager : public boost::noncopyable, public nebula::cpp::NonMovable {
    */
   void resetSpaceRunning(GraphSpaceID spaceId);
 
+  /**
+   * @brief check if the space exist
+   *
+   * @return
+   */
+  bool spaceExist(GraphSpaceID spaceId);
+
  private:
   using PriorityQueue = folly::PriorityUMPSCQueueSet<std::tuple<JbOp, JobID, GraphSpaceID>, true>;
   // Each PriorityQueue contains high and low priority queues.

src/meta/processors/job/ReportTaskProcessor.cpp

@@ -13,6 +13,8 @@ namespace meta {
 #include "meta/processors/job/JobManager.h"
 
 void ReportTaskProcessor::process(const cpp2::ReportTaskReq& req) {
+  CHECK_SPACE_ID_AND_RETURN(req.get_space_id());
+
   JobManager* jobMgr = JobManager::getInstance();
   auto rc = jobMgr->reportTaskFinish(req);
   if (rc != nebula::cpp2::ErrorCode::SUCCEEDED) {

src/meta/processors/parts/DropSpaceProcessor.cpp

@@ -129,6 +129,23 @@ void DropSpaceProcessor::process(const cpp2::DropSpaceReq& req) {
   auto localIdkey = MetaKeyUtils::localIdKey(spaceId);
   batchHolder->remove(std::move(localIdkey));
 
+  // 8. Delete all job data
+  auto jobPrefix = MetaKeyUtils::jobPrefix(spaceId);
+  auto jobRet = doPrefix(jobPrefix);
+  if (!nebula::ok(jobRet)) {
+    auto retCode = nebula::error(jobRet);
+    LOG(INFO) << "Loading Job Failed" << apache::thrift::util::enumNameSafe(retCode);
+    handleErrorCode(retCode);
+    onFinished();
+    return;
+  }
+
+  auto jobIter = nebula::value(jobRet).get();
+  while (jobIter->valid()) {
+    batchHolder->remove(jobIter->key().str());
+    jobIter->next();
+  }
+
   auto timeInMilliSec = time::WallClock::fastNowInMilliSec();
   LastUpdateTimeMan::update(batchHolder.get(), timeInMilliSec);
   auto batch = encodeBatchValue(std::move(batchHolder)->getBatch());