fix

Author: liulx20

flex/engines/graph_db/app/kafka_wal_ingester_app.cc

@@ -16,12 +16,13 @@
 #include "flex/engines/graph_db/app/kafka_wal_ingester_app.h"
 #include "cppkafka/cppkafka.h"
 #include "flex/engines/graph_db/database/graph_db.h"
-#include "flex/engines/graph_db/database/wal/kafka_wal_parser.h"
+#include "flex/engines/graph_db/database/wal/kafka_wal_utils.h"
 
 namespace gs {
 #ifdef BUILD_KAFKA_WAL_WRITER_PARSER
 
 struct WalIngester {
+  constexpr static size_t BUFFSIZ = 4096;
   GraphDBSession& session_;
   timestamp_t begin_;
   timestamp_t end_;
@@ -31,30 +32,32 @@ struct WalIngester {
   std::vector<uint8_t> states_;
 
   void resize() {
-    size_t n = data_.size();
-    std::vector<std::string> new_data(n + 4096);
-    std::vector<uint8_t> new_states(n + 4096, 0);
-    size_t idx = (ingested_plus_one_ - begin_) % n;
-    for (size_t i = 0; i < n; ++i) {
+    size_t origin_len = data_.size();
+    std::vector<std::string> new_data(origin_len + BUFFSIZ);
+    std::vector<uint8_t> new_states(origin_len + BUFFSIZ, 0);
+    size_t idx = (ingested_plus_one_ - begin_) % origin_len;
+    for (size_t i = 0; i < origin_len; ++i) {
       new_data[i] = data_[idx];
       new_states[i] = states_[idx];
       if (states_[idx]) {
         end_ = ingested_plus_one_ + i + 1;
       }
       ++idx;
-      idx %= n;
+      idx %= origin_len;
     }
     data_ = std::move(new_data);
     states_ = std::move(new_states);
     begin_ = ingested_plus_one_;
   }
+
+  timestamp_t last_ingested() const { return ingested_plus_one_ - 1; }
   WalIngester(GraphDBSession& session, timestamp_t cur)
       : session_(session),
         begin_(cur),
         end_(cur),
         ingested_plus_one_(cur),
-        data_(4096),
-        states_(4096, 0) {}
+        data_(BUFFSIZ),
+        states_(BUFFSIZ, 0) {}
 
   bool empty() const { return ingested_plus_one_ == end_; }
 
@@ -77,7 +80,8 @@ struct WalIngester {
   }
 
   void ingest() {
-    size_t idx = (ingested_plus_one_ - begin_) % data_.size();
+    size_t len = data_.size();
+    size_t idx = (ingested_plus_one_ - begin_) % len;
     bool flag = false;
     while (states_[idx] == 2 || states_[idx] == 1) {
       if (states_[idx] == 1) {
@@ -86,7 +90,7 @@ struct WalIngester {
       states_[idx] = 0;
       ++ingested_plus_one_;
       ++idx;
-      idx %= data_.size();
+      idx %= len;
       flag = true;
     }
     if (flag) {
@@ -180,14 +184,15 @@ bool KafkaWalIngesterApp::Query(GraphDBSession& graph, Decoder& input,
     if (key == "topic_name") {
       topic_name = value;
     } else {
+      LOG(INFO) << "Kafka config: " << key << " = " << value;
       config.set(std::string(key), std::string(value));
     }
   }
   LOG(INFO) << "Kafka brokers: " << config.get("metadata.broker.list");
   timestamp_t cur_ts = graph.db().get_last_ingested_wal_ts() + 1;
   gs::WalIngester ingester(graph, cur_ts);
   gs::KafkaWalConsumer consumer(ingester, config, topic_name, 1);
-  while (graph.db().kafka_wal_ingester_state()) {
+  while (!force_stop_.load()) {
     consumer.poll();
     ingester.ingest();
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
@@ -196,8 +201,16 @@ bool KafkaWalIngesterApp::Query(GraphDBSession& graph, Decoder& input,
     consumer.poll();
     ingester.ingest();
   }
+  int64_t ts = ingester.last_ingested();
+  output.put_long(ts);
+  return true;
+}
+
+bool KafkaWalIngesterApp::terminal() {
+  force_stop_.store(true);
   return true;
 }
+
 AppWrapper KafkaWalIngesterAppFactory::CreateApp(const GraphDB& db) {
   return AppWrapper(new KafkaWalIngesterApp(), NULL);
 }

flex/engines/graph_db/app/kafka_wal_ingester_app.h

@@ -24,11 +24,14 @@ namespace gs {
 // Ingest wal from kafka
 class KafkaWalIngesterApp : public WriteAppBase {
  public:
-  KafkaWalIngesterApp() {}
+  KafkaWalIngesterApp() : force_stop_(false) {}
 
   AppType type() const override { return AppType::kBuiltIn; }
 
   bool Query(GraphDBSession& graph, Decoder& input, Encoder& output) override;
+
+  bool terminal();
+  std::atomic<bool> force_stop_{false};
 };
 
 class KafkaWalIngesterAppFactory : public AppFactoryBase {

flex/engines/graph_db/database/graph_db.cc

@@ -28,10 +28,6 @@
 #include "flex/engines/graph_db/runtime/utils/cypher_runner_impl.h"
 #include "flex/utils/yaml_utils.h"
 
-#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
-#include "flex/engines/graph_db/app/kafka_wal_ingester_app.h"
-#endif
-
 #include "flex/third_party/httplib.h"
 
 namespace gs {
@@ -58,7 +54,6 @@ struct SessionLocalContext {
 
 GraphDB::GraphDB()
     : monitor_thread_running_(false),
-      kafka_wal_ingester_thread_running_(false),
       last_ingested_wal_ts_(0),
       compact_thread_running_(false) {}
 GraphDB::~GraphDB() {
@@ -282,54 +277,6 @@ void GraphDB::Close() {
   std::fill(app_factories_.begin(), app_factories_.end(), nullptr);
 }
 
-#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
-bool GraphDB::kafka_wal_ingester_state() const {
-  return kafka_wal_ingester_thread_running_.load();
-}
-
-void GraphDB::start_kafka_wal_ingester(const cppkafka::Configuration& config,
-                                       const std::string& topic_name) {
-  if (kafka_wal_ingester_thread_running_) {
-    kafka_wal_ingester_thread_running_ = false;
-    if (kafka_wal_ingester_thread_.joinable()) {
-      kafka_wal_ingester_thread_.join();
-    }
-  }
-  kafka_wal_ingester_thread_running_ = true;
-
-  kafka_wal_ingester_thread_ = std::thread([&]() {
-    std::vector<char> buffer;
-    gs::Encoder encoder(buffer);
-    encoder.put_string("topic_name");
-    encoder.put_string(topic_name);
-    if (config.has_property("metadata.broker.list")) {
-      encoder.put_string("metadata.broker.list");
-      encoder.put_string(config.get("metadata.broker.list"));
-    }
-    if (config.has_property("group.id")) {
-      encoder.put_string("group.id");
-      encoder.put_string(config.get("group.id"));
-    }
-    if (config.has_property("enable.auto.commit")) {
-      encoder.put_string("enable.auto.commit");
-      encoder.put_string(config.get("enable.auto.commit"));
-    }
-    if (config.has_property("auto.offset.reset")) {
-      encoder.put_string("auto.offset.reset");
-      encoder.put_string(config.get("auto.offset.reset"));
-    }
-    gs::Decoder decoder(buffer.data(), buffer.size());
-    KafkaWalIngesterApp().Query(GetSession(0), decoder, encoder);
-  });
-}
-
-void GraphDB::stop_kafka_wal_ingester() {
-  kafka_wal_ingester_thread_running_ = false;
-  kafka_wal_ingester_thread_.join();
-}
-
-#endif
-
 ReadTransaction GraphDB::GetReadTransaction(int thread_id) {
   return contexts_[thread_id].session.GetReadTransaction();
 }

flex/engines/graph_db/database/graph_db.h

@@ -33,9 +33,6 @@
 #include "flex/storages/rt_mutable_graph/loader/loader_factory.h"
 #include "flex/storages/rt_mutable_graph/loading_config.h"
 #include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
-#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
-#include "cppkafka/cppkafka.h"
-#endif
 
 namespace gs {
 
@@ -172,15 +169,6 @@ class GraphDB {
 
   inline const GraphDBConfig& config() const { return config_; }
 
-#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
-  bool kafka_wal_ingester_state() const;
-
-  void start_kafka_wal_ingester(const cppkafka::Configuration& config,
-                                const std::string& topic_name);
-
-  void stop_kafka_wal_ingester();
-#endif
-
   uint64_t get_last_ingested_wal_ts() const { return last_ingested_wal_ts_; }
   void set_last_ingested_wal_ts(uint64_t ts) { last_ingested_wal_ts_ = ts; }
 
@@ -219,8 +207,6 @@ class GraphDB {
   std::thread monitor_thread_;
   bool monitor_thread_running_;
 
-  std::thread kafka_wal_ingester_thread_;
-  std::atomic<bool> kafka_wal_ingester_thread_running_;
   uint64_t last_ingested_wal_ts_;
 
   timestamp_t last_compaction_ts_;

flex/engines/graph_db/database/wal/kafka_wal_parser.cc

@@ -14,81 +14,31 @@
  */
 
 #include "flex/engines/graph_db/database/wal/kafka_wal_parser.h"
+#include "flex/engines/graph_db/database/wal/kafka_wal_utils.h"
 #include "flex/engines/graph_db/database/wal/wal.h"
 
 namespace gs {
 
-std::vector<cppkafka::TopicPartition> get_all_topic_partitions(
-    const cppkafka::Configuration& config, const std::string& topic_name,
-    bool from_beginning) {
-  std::vector<cppkafka::TopicPartition> partitions;
-  cppkafka::Consumer consumer(config);  // tmp consumer
-  LOG(INFO) << config.get("metadata.broker.list");
-  LOG(INFO) << config.get("group.id");
-
-  LOG(INFO) << "Get metadata for topic " << topic_name;
-  auto meta_vector = consumer.get_metadata().get_topics({topic_name});
-  if (meta_vector.empty()) {
-    LOG(WARNING) << "Failed to get metadata for topic " << topic_name
-                 << ", maybe the topic does not exist";
-    return {};
-  }
-  auto metadata = meta_vector.front().get_partitions();
-  for (const auto& partition : metadata) {
-    if (from_beginning) {
-      partitions.push_back(
-          cppkafka::TopicPartition(topic_name, partition.get_id(), 0));
-    } else {
-      partitions.push_back(cppkafka::TopicPartition(
-          topic_name, partition.get_id()));  // from the beginning
-    }
-  }
-  return partitions;
-}
-
 std::unique_ptr<IWalParser> KafkaWalParser::Make(const std::string& uri) {
   // uri should be like
   // "kafka://localhost:9092,localhost:9093/my_topic?group.id=my_consumer_group"
-  const std::string prefix = "kafka://";
-  if (uri.find(prefix) != 0) {
-    LOG(FATAL) << "Invalid uri: " << uri;
-  }
-
-  std::string hosts_part = uri.substr(prefix.length());
-  size_t query_pos = hosts_part.find('/');
-  std::string hosts;
-  std::string query;
-  cppkafka::Configuration config;
-
-  if (query_pos != std::string::npos) {
-    hosts = hosts_part.substr(0, query_pos);
-    query = hosts_part.substr(query_pos + 1);
-  } else {
-    LOG(FATAL) << "Invalid uri: " << uri;
+  auto res = parse_uri(uri);
+  if (!res) {
+    LOG(FATAL) << "Failed to parse uri: " << uri;
+    return nullptr;
   }
-  config.set("metadata.broker.list", hosts);
-  size_t top_pos = query.find('?');
+  gs::Decoder decoder(res.value().data(), res.value().size());
   std::string topic_name;
-  if (top_pos != std::string::npos) {
-    topic_name = query.substr(0, top_pos);
-    query = query.substr(top_pos + 1);
-  } else {
-    LOG(FATAL) << "Invalid uri: " << uri;
-  }
-  std::istringstream query_stream(query);
-  std::string pair;
-  while (std::getline(query_stream, pair, '&')) {
-    size_t eq_pos = pair.find('=');
-    if (eq_pos != std::string::npos) {
-      std::string key = pair.substr(0, eq_pos);
-      std::string value = pair.substr(eq_pos + 1);
-      if (key == "group.id") {
-        config.set("group.id", value);
-      }
+  cppkafka::Configuration config;
+  while (!decoder.empty()) {
+    auto key = decoder.get_string();
+    auto value = decoder.get_string();
+    if (key == "topic_name") {
+      topic_name = value;
+    } else {
+      config.set(std::string(key), std::string(value));
     }
   }
-  config.set("enable.auto.commit", false);
-
   auto parser = std::unique_ptr<IWalParser>(new KafkaWalParser(config));
   parser->open(topic_name);
   return parser;

flex/engines/graph_db/database/wal/kafka_wal_parser.h

@@ -22,13 +22,6 @@
 
 namespace gs {
 
-/*
- * Get all partitions of the given topic.
- */
-std::vector<cppkafka::TopicPartition> get_all_topic_partitions(
-    const cppkafka::Configuration& config, const std::string& topic_name,
-    bool from_beginning = true);
-
 class KafkaWalParser : public IWalParser {
  public:
   static constexpr const std::chrono::milliseconds POLL_TIMEOUT =