revert src/kvstore

src/kvstore/DiskManager.cpp

@@ -20,8 +20,6 @@ DiskManager::DiskManager(const std::vector<std::string>& dataPaths,
  Paths* paths = new Paths();
  paths_.store(paths);
  size_t index = 0;
-
- // TODO(vee): Add initialize function to avoid using LOG(FATAL) in constructor.
  for (const auto& path : dataPaths) {
  auto absolute = boost::filesystem::absolute(path);
  if (!boost::filesystem::exists(absolute)) {
@@ -95,7 +93,7 @@ void DiskManager::addPartToPath(GraphSpaceID spaceId, PartitionID partId, const
  paths_.store(newPaths, std::memory_order_release);
  folly::rcu_retire(oldPaths, std::default_delete<Paths>());
  } catch (boost::filesystem::filesystem_error& e) {
- LOG(DFATAL) << "Invalid path: " << e.what();
+ LOG(FATAL) << "Invalid path: " << e.what();
  }
 }
 
@@ -116,7 +114,7 @@ void DiskManager::removePartFromPath(GraphSpaceID spaceId,
  paths_.store(newPaths, std::memory_order_release);
  folly::rcu_retire(oldPaths, std::default_delete<Paths>());
  } catch (boost::filesystem::filesystem_error& e) {
- LOG(DFATAL) << "Invalid path: " << e.what();
+ LOG(FATAL) << "Invalid path: " << e.what();
  }
 }
 

src/kvstore/NebulaStore.cpp

@@ -343,7 +343,7 @@ std::unique_ptr<KVEngine> NebulaStore::newEngine(GraphSpaceID spaceId,
  return std::make_unique<RocksEngine>(
  spaceId, vIdLen, dataPath, walPath, options_.mergeOp_, cfFactory);
  } else {
- LOG(DFATAL) << "Unknown engine type " << FLAGS_engine_type;
+ LOG(FATAL) << "Unknown engine type " << FLAGS_engine_type;
  return nullptr;
  }
 }
@@ -627,7 +627,7 @@ std::shared_ptr<Listener> NebulaStore::newListener(GraphSpaceID spaceId,
  listener = std::make_shared<ESListener>(
  spaceId, partId, raftAddr_, walPath, ioPool_, bgWorkers_, workers_, options_.schemaMan_);
  } else {
- LOG(DFATAL) << "Should not reach here";
+ LOG(FATAL) << "Should not reach here";
  return nullptr;
  }
  raftService_->addPartition(listener);

src/kvstore/RocksEngine.cpp

@@ -47,8 +47,6 @@ RocksEngine::RocksEngine(GraphSpaceID spaceId,
  walPath_ = folly::stringPrintf("%s/nebula/%d", walPath.c_str(), spaceId);
  }
  auto path = folly::stringPrintf("%s/data", dataPath_.c_str());
-
- // TODO(vee): Add initialize function to avoid using LOG(FATAL) in constructor.
  if (FileUtils::fileType(path.c_str()) == FileType::NOTEXIST) {
  if (readonly) {
  LOG(FATAL) << "Path " << path << " not exist";

src/kvstore/RocksEngineConfig.cpp

@@ -248,8 +248,7 @@ rocksdb::Status initRocksdbOptions(rocksdb::Options& baseOpts,
  auto walDir = folly::stringPrintf("%s/rocksdb_wal/%d", FLAGS_rocksdb_wal_dir.c_str(), spaceId);
  if (fs::FileUtils::fileType(walDir.c_str()) == fs::FileType::NOTEXIST) {
  if (!fs::FileUtils::makeDir(walDir)) {
- LOG(DFATAL) << "makeDir " << walDir << " failed";
- return rocksdb::Status::InvalidArgument();
+ LOG(FATAL) << "makeDir " << walDir << " failed";
  }
  }
  LOG(INFO) << "set rocksdb wal of space " << spaceId << " to " << walDir;

src/kvstore/listener/Listener.cpp

@@ -39,10 +39,7 @@ Listener::Listener(GraphSpaceID spaceId,
 void Listener::start(std::vector<HostAddr>&& peers, bool) {
  std::lock_guard<std::mutex> g(raftLock_);
 
- if (!init()) {
- // TODO(vee): return bool to avoid using LOG(FATAL)
- LOG(FATAL) << "Listener init failed";
- }
+ init();
 
  lastLogId_ = wal_->lastLogId();
  lastLogTerm_ = wal_->lastLogTerm();

src/kvstore/listener/Listener.h

@@ -75,7 +75,7 @@ using RaftClient = thrift::ThriftClientManager<raftex::cpp2::RaftexServiceAsyncC
  *
  * * Must implement in derived class
  * // extra initialize work could do here
- * bool init()
+ * void init()
  *
  * // Main interface to process logs, listener need to apply the committed log entry to their
  * // state machine. Once apply succeeded, user should call persist() to make their progress
@@ -166,7 +166,7 @@ class Listener : public raftex::RaftPart {
  /**
  * @brief extra initialize work could do here
  */
- virtual bool init() = 0;
+ virtual void init() = 0;
 
  /**
  * @brief Get last apply id from persistence storage, used in initialization
@@ -194,7 +194,7 @@ class Listener : public raftex::RaftPart {
  */
  void onLostLeadership(TermID term) override {
  UNUSED(term);
- LOG(DFATAL) << "Should not reach here";
+ LOG(FATAL) << "Should not reach here";
  }
 
  /**
@@ -204,7 +204,7 @@ class Listener : public raftex::RaftPart {
  */
  void onElected(TermID term) override {
  UNUSED(term);
- LOG(DFATAL) << "Should not reach here";
+ LOG(FATAL) << "Should not reach here";
  }
 
  /**
@@ -214,7 +214,7 @@ class Listener : public raftex::RaftPart {
  */
  void onLeaderReady(TermID term) override {
  UNUSED(term);
- LOG(DFATAL) << "Should not reach here";
+ LOG(FATAL) << "Should not reach here";
  }
 
  /**

src/kvstore/listener/elasticsearch/ESListener.cpp

@@ -14,11 +14,10 @@ DEFINE_int32(listener_commit_batch_size, 1000, "Max batch size when listener com
 
 namespace nebula {
 namespace kvstore {
-bool ESListener::init() {
+void ESListener::init() {
  auto vRet = schemaMan_->getSpaceVidLen(spaceId_);
  if (!vRet.ok()) {
- LOG(DFATAL) << "vid length error";
- return false;
+ LOG(FATAL) << "vid length error";
  }
  vIdLen_ = vRet.value();
  auto vidTypeRet = schemaMan_->getSpaceVidType(spaceId_);
@@ -29,8 +28,7 @@ bool ESListener::init() {
 
  auto cRet = schemaMan_->getServiceClients(meta::cpp2::ExternalServiceType::ELASTICSEARCH);
  if (!cRet.ok() || cRet.value().empty()) {
- LOG(DFATAL) << "elasticsearch clients error";
- return false;
+ LOG(FATAL) << "elasticsearch clients error";
  }
  std::vector<nebula::plugin::ESClient> esClients;
  for (const auto& c : cRet.value()) {
@@ -46,11 +44,9 @@ bool ESListener::init() {
  esAdapter_.setClients(std::move(esClients));
  auto sRet = schemaMan_->toGraphSpaceName(spaceId_);
  if (!sRet.ok()) {
- LOG(DFATAL) << "space name error";
- return false;
+ LOG(FATAL) << "space name error";
  }
  spaceName_ = std::make_unique<std::string>(sRet.value());
- return true;
 }
 
 bool ESListener::apply(const BatchHolder& batch) {
@@ -67,7 +63,7 @@ bool ESListener::apply(const BatchHolder& batch) {
  } else if (type == BatchLogType::OP_BATCH_REMOVE) {
  bulk.delete_(index, vid, src, dst, rank);
  } else {
- LOG(DFATAL) << "Unexpect";
+ LOG(FATAL) << "Unexpect";
  }
  };
  for (const auto& log : batch.getBatch()) {
@@ -156,8 +152,7 @@ void ESListener::pickTagAndEdgeData(BatchLogType type,
 
 bool ESListener::persist(LogID lastId, TermID lastTerm, LogID lastApplyLogId) {
  if (!writeAppliedId(lastId, lastTerm, lastApplyLogId)) {
- LOG(DFATAL) << "last apply ids write failed";
- return false;
+ LOG(FATAL) << "last apply ids write failed";
  }
  return true;
 }
@@ -169,9 +164,8 @@ std::pair<LogID, TermID> ESListener::lastCommittedLogId() {
  }
  int32_t fd = open(lastApplyLogFile_->c_str(), O_RDONLY);
  if (fd < 0) {
- LOG(DFATAL) << "Failed to open the file \"" << lastApplyLogFile_->c_str() << "\" (" << errno
- << "): " << strerror(errno);
- return {0, 0};
+ LOG(FATAL) << "Failed to open the file \"" << lastApplyLogFile_->c_str() << "\" (" << errno
+ << "): " << strerror(errno);
  }
  // read last logId from listener wal file.
  LogID logId;
@@ -193,9 +187,8 @@ LogID ESListener::lastApplyLogId() {
  }
  int32_t fd = open(lastApplyLogFile_->c_str(), O_RDONLY);
  if (fd < 0) {
- LOG(DFATAL) << "Failed to open the file \"" << lastApplyLogFile_->c_str() << "\" (" << errno
- << "): " << strerror(errno);
- return 0;
+ LOG(FATAL) << "Failed to open the file \"" << lastApplyLogFile_->c_str() << "\" (" << errno
+ << "): " << strerror(errno);
  }
  // read last applied logId from listener wal file.
  LogID logId;

src/kvstore/listener/elasticsearch/ESListener.h

@@ -46,7 +46,7 @@ class ESListener : public Listener {
  /**
  * @brief Init work: get vid length, get es client
  */
- bool init() override;
+ void init() override;
 
  /**
  * @brief Send data by es client

src/kvstore/listener/test/NebulaListenerTest.cpp

@@ -211,9 +211,7 @@ class DummyListener : public Listener {
  }
 
  protected:
- bool init() override {
- return true;
- }
+ void init() override {}
 
  bool apply(const BatchHolder& batch) {
  for (auto& log : batch.getBatch()) {

src/kvstore/plugins/hbase/HBaseStore.cpp

@@ -212,8 +212,7 @@ ResultCode HBaseStore::rangeWithPrefix(GraphSpaceID spaceId,
 ResultCode HBaseStore::sync(GraphSpaceID spaceId, PartitionID partId) {
  UNUSED(spaceId);
  UNUSED(partId);
- LOG(DFATAL) << "Unimplement";
- return ResultCode::ERR_UNSUPPORTED;
+ LOG(FATAL) << "Unimplement";
 }
 
 ResultCode HBaseStore::multiRemove(GraphSpaceID spaceId, std::vector<std::string>& keys) {
@@ -404,13 +403,11 @@ void HBaseStore::asyncRemovePrefix(GraphSpaceID spaceId,
 }
 
 ResultCode HBaseStore::ingest(GraphSpaceID) {
- LOG(DFATAL) << "Unimplement";
- return ResultCode::ERR_UNSUPPORTED;
+ LOG(FATAL) << "Unimplement";
 }
 
 int32_t HBaseStore::allLeader(std::unordered_map<GraphSpaceID, std::vector<PartitionID>>&) {
- LOG(DFATAL) << "Unimplement";
- return 0;
+ LOG(FATAL) << "Unimplement";
 }
 
 } // namespace kvstore

src/kvstore/plugins/hbase/HBaseStore.h

@@ -181,11 +181,11 @@ class HBaseStore : public KVStore {
  KVCallback cb);
 
  void asyncAtomicOp(GraphSpaceID, PartitionID, raftex::AtomicOp, KVCallback) override {
- LOG(DFATAL) << "Not supported yet!";
+ LOG(FATAL) << "Not supported yet!";
  }
 
  void asyncAtomicOp(GraphSpaceID, PartitionID, std::string&& multiValues, KVCallback) override {
- LOG(DFATAL) << "Not supported yet!";
+ LOG(FATAL) << "Not supported yet!";
  }
 
  ResultCode ingest(GraphSpaceID spaceId) override;

src/kvstore/raftex/RaftPart.cpp

@@ -95,7 +95,7 @@ class AppendLogsIterator final : public LogIterator {
  }
  }
  if (notFulfilledPromise > 0) {
- LOG(DFATAL) << "notFulfilledPromise == " << notFulfilledPromise;
+ LOG(FATAL) << "notFulfilledPromise == " << notFulfilledPromise;
  }
  }
  }
@@ -394,7 +394,7 @@ const char* RaftPart::roleStr(Role role) const {
  case Role::LEARNER:
  return "Learner";
  default:
- LOG(DFATAL) << idStr_ << "Invalid role";
+ LOG(FATAL) << idStr_ << "Invalid role";
  }
  return nullptr;
 }
@@ -1091,8 +1091,7 @@ void RaftPart::processAppendLogResponses(const AppendLogResponses& resps,
  [self = shared_from_this(), term = term_] { self->onLeaderReady(term); });
  }
  } else {
- LOG(DFATAL) << idStr_ << "Failed to commit logs";
- return;
+ LOG(FATAL) << idStr_ << "Failed to commit logs";
  }
  VLOG(4) << idStr_ << "Leader succeeded in committing the logs " << committedId + 1 << " to "
  << lastLogId;

src/kvstore/wal/FileBasedWal.cpp

@@ -479,9 +479,8 @@ bool FileBasedWal::appendLogInternal(LogID id,
 
  ssize_t bytesWritten = write(currFd_, strBuf.data(), strBuf.size());
  if (bytesWritten != (ssize_t)strBuf.size()) {
- LOG(DFATAL) << idStr_ << "bytesWritten:" << bytesWritten << ", expected:" << strBuf.size()
- << ", error:" << strerror(errno);
- return false;
+ LOG(FATAL) << idStr_ << "bytesWritten:" << bytesWritten << ", expected:" << strBuf.size()
+ << ", error:" << strerror(errno);
  }
 
  if (policy_.sync && ::fsync(currFd_) == -1) {