fix ci

Author: liulx20

.github/workflows/interactive.yml

@@ -371,6 +371,7 @@ jobs:
       env:
         GS_TEST_DIR: ${{ github.workspace }}/gstest
       run: |
+        cd ${GITHUB_WORKSPACE}/flex/build
         wget https://dlcdn.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
         tar -zxf kafka_2.13-3.9.0.tgz
         cd kafka_2.13-3.9.0
@@ -379,10 +380,12 @@ jobs:
         bin/kafka-server-start.sh config/kraft/reconfig-server.properties &
 
         bin/kafka-topics.sh --create --topic kafka-test --bootstrap-server localhost:9092
-
-        ./tests/hqps/kafka_test localhost:902 kafka-test
-
-        pkill -f kafka-server-start.sh
+        ../tests/hqps/kafka_test localhost:9092 kafka-test
+        bin/kafka-topics.sh --delete --topic kafka-test --bootstrap-server localhost:9092
+        bin/kafka-topics.sh --create --topic kafka-test --bootstrap-server localhost:9092
+        ../tests/hqps/kafka_wal_ingester_test .. localhost:9092 kafka-test
+        bin/kafka-topics.sh --delete --topic kafka-test --bootstrap-server localhost:9092
+        pkill -f kafka
 
     - name: Run Gremlin test on modern graph
       env:

flex/engines/graph_db/app/kafka_wal_ingester_app.cc

@@ -77,20 +77,18 @@ bool KafkaWalIngesterApp::Query(GraphDBSession& graph, Decoder& input,
                                 Encoder& output) {
   // TODO: get value from config
   std::string kafka_brokers = std::string(input.get_string());
-  std::string brokers = std::string(input.get_string());
-  std::string topic_name = std::string(input.get_string());
   std::string group_id = std::string(input.get_string());
-  std::string engine_endpoint = std::string(input.get_string());
+  bool enable_auto_commit = input.get_byte();
+  std::string topic_name = std::string(input.get_string());
   LOG(INFO) << "Kafka brokers: " << kafka_brokers;
-  LOG(INFO) << "engine endpoint: " << engine_endpoint;
 
   cppkafka::Configuration config = {{"metadata.broker.list", kafka_brokers},
                                     {"group.id", group_id},
                                     // Disable auto commit
-                                    {"enable.auto.commit", false}};
+                                    {"enable.auto.commit", enable_auto_commit}};
   gs::KafkaWalConsumer consumer(config, topic_name, 1);
   // TODO: how to make it stop
-  while (!graph.db().kafka_wal_ingester_state()) {
+  while (graph.db().kafka_wal_ingester_state()) {
     auto res = consumer.poll();
     if (res.empty()) {
       std::this_thread::sleep_for(gs::KafkaWalConsumer::POLL_TIMEOUT);
@@ -103,11 +101,13 @@ bool KafkaWalIngesterApp::Query(GraphDBSession& graph, Decoder& input,
       txn.IngestWal(graph.graph(), txn.timestamp(),
                     const_cast<char*>(res.data()) + sizeof(WalHeader),
                     header->length, txn.allocator());
+      txn.Commit();
     } else if (header->type == 1) {
       auto txn = graph.GetUpdateTransaction();
       txn.IngestWal(graph.graph(), graph.db().work_dir(), txn.timestamp(),
                     const_cast<char*>(res.data()) + sizeof(WalHeader),
                     header->length, txn.allocator());
+      txn.Commit();
     } else {
       LOG(ERROR) << "Unknown wal type: " << header->type;
     }

flex/engines/graph_db/database/graph_db.cc

@@ -283,24 +283,25 @@ bool GraphDB::kafka_wal_ingester_state() const {
   return kafka_wal_ingester_thread_running_.load();
 }
 
-void GraphDB::start_kafka_wal_ingester(const std::string& kafka_brokers,
-                                       const std::string& brokers,
-                                       const std::string& topic_name,
-                                       const std::string& group_id,
-                                       const std::string& engine_endpoint) {
+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;
-    kafka_wal_ingester_thread_.join();
+    if (kafka_wal_ingester_thread_.joinable()) {
+      kafka_wal_ingester_thread_.join();
+    }
   }
-  std::vector<char> buffer;
-  gs::Encoder encoder(buffer);
-  encoder.put_string(kafka_brokers);
-  encoder.put_string(brokers);
-  encoder.put_string(topic_name);
-  encoder.put_string(group_id);
-  encoder.put_string(engine_endpoint);
-  gs::Decoder decoder(buffer.data(), buffer.size());
-  KafkaWalIngesterApp().Query(GetSession(0), decoder, encoder);
+  kafka_wal_ingester_thread_running_ = true;
+  kafka_wal_ingester_thread_ = std::thread([&]() {
+    std::vector<char> buffer;
+    gs::Encoder encoder(buffer);
+    encoder.put_string(config.get("metadata.broker.list"));
+    encoder.put_string(config.get("group.id"));
+    encoder.put_byte(config.get("enable.auto.commit") == "true");
+    encoder.put_string(topic_name);
+    gs::Decoder decoder(buffer.data(), buffer.size());
+    KafkaWalIngesterApp().Query(GetSession(0), decoder, encoder);
+  });
 }
 
 void GraphDB::stop_kafka_wal_ingester() {

flex/engines/graph_db/database/graph_db.h

@@ -33,6 +33,9 @@
 #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,11 +175,8 @@ class GraphDB {
 #ifdef BUILD_KAFKA_WAL_WRITER_PARSER
   bool kafka_wal_ingester_state() const;
 
-  void start_kafka_wal_ingester(const std::string& kafka_brokers,
-                                const std::string& brokers,
-                                const std::string& topic_name,
-                                const std::string& group_id,
-                                const std::string& engine_endpoint);
+  void start_kafka_wal_ingester(const cppkafka::Configuration& config,
+                                const std::string& topic_name);
 
   void stop_kafka_wal_ingester();
 #endif

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

@@ -14,10 +14,6 @@
  */
 
 #include "flex/engines/graph_db/database/wal/kafka_wal_parser.h"
-#include <fcntl.h>
-#include <sys/mman.h>
-#include <unistd.h>
-#include <filesystem>
 #include "flex/engines/graph_db/database/wal/wal.h"
 
 namespace gs {
@@ -40,7 +36,7 @@ std::vector<cppkafka::TopicPartition> get_all_topic_partitions(
   return partitions;
 }
 
-std::unique_ptr<IWalParser> KafkaWalParser::Make() {
+std::unique_ptr<IWalParser> KafkaWalParser::Make(const std::string&) {
   const char* broker_list = std::getenv("KAFKA_BROKER_LIST");
   if (broker_list == nullptr) {
     LOG(FATAL) << "KAFKA_BROKER_LIST is not set";
@@ -53,11 +49,7 @@ std::unique_ptr<IWalParser> KafkaWalParser::Make() {
 }
 
 KafkaWalParser::KafkaWalParser(const cppkafka::Configuration& config)
-    : consumer_(nullptr),
-      insert_wal_list_(NULL),
-      insert_wal_list_size_(0),
-      last_ts_(0),
-      config_(config) {
+    : consumer_(nullptr), last_ts_(0), config_(config) {
   consumer_ = std::make_unique<cppkafka::Consumer>(config);
 }
 
@@ -69,11 +61,7 @@ void KafkaWalParser::open(const std::string& topic_name) {
 void KafkaWalParser::open(
     const std::vector<cppkafka::TopicPartition>& topic_partitions) {
   consumer_->assign(topic_partitions);
-  insert_wal_list_ = static_cast<WalContentUnit*>(
-      mmap(NULL, IWalWriter::MAX_WALS_NUM * sizeof(WalContentUnit),
-           PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_NORESERVE,
-           -1, 0));
-  insert_wal_list_size_ = IWalWriter::MAX_WALS_NUM;
+  insert_wal_list_.resize(4096);
 
   while (true) {
     auto msgs = consumer_->poll_batch(MAX_BATCH_SIZE);
@@ -95,7 +83,7 @@ void KafkaWalParser::open(
   }
 
   for (auto& wal : message_vector_) {
-    LOG(INFO) << "Got wal:" << wal.size();
+    VLOG(1) << "Got wal:" << wal.size();
     const char* payload = wal.data();
     const WalHeader* header = reinterpret_cast<const WalHeader*>(payload);
     uint32_t cur_ts = header->timestamp;
@@ -111,6 +99,9 @@ void KafkaWalParser::open(
       unit.size = length;
       update_wal_list_.push_back(unit);
     } else {
+      if (cur_ts >= insert_wal_list_.size()) {
+        insert_wal_list_.resize(cur_ts + 1);
+      }
       if (insert_wal_list_[cur_ts].ptr) {
         LOG(WARNING) << "Duplicated timestamp " << cur_ts << ", skip";
       }
@@ -134,9 +125,7 @@ void KafkaWalParser::close() {
   if (consumer_) {
     consumer_.reset();
   }
-  if (insert_wal_list_ != NULL) {
-    munmap(insert_wal_list_, insert_wal_list_size_ * sizeof(WalContentUnit));
-  }
+  insert_wal_list_.clear();
 }
 
 uint32_t KafkaWalParser::last_ts() const { return last_ts_; }
@@ -151,4 +140,8 @@ const std::vector<UpdateWalUnit>& KafkaWalParser::get_update_wals() const {
   return update_wal_list_;
 }
 
+const bool KafkaWalParser::registered_ = WalParserFactory::RegisterWalParser(
+    "kafaka", static_cast<WalParserFactory::wal_parser_initializer_t>(
+                  &KafkaWalParser::Make));
+
 }  // namespace gs

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

@@ -17,9 +17,8 @@
 #define ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_
 
 #include <vector>
-#include "flex/engines/graph_db/database/wal/wal.h"
-
 #include "cppkafka/cppkafka.h"
+#include "flex/engines/graph_db/database/wal/wal.h"
 
 namespace gs {
 
@@ -35,7 +34,7 @@ class KafkaWalParser : public IWalParser {
       std::chrono::milliseconds(100);
   static constexpr const size_t MAX_BATCH_SIZE = 1000;
 
-  static std::unique_ptr<IWalParser> Make();
+  static std::unique_ptr<IWalParser> Make(const std::string&);
 
   // always track all partitions and from begining
   KafkaWalParser(const cppkafka::Configuration& config);
@@ -53,13 +52,14 @@ class KafkaWalParser : public IWalParser {
 
  private:
   std::unique_ptr<cppkafka::Consumer> consumer_;
-  WalContentUnit* insert_wal_list_;
-  size_t insert_wal_list_size_;
+  std::vector<WalContentUnit> insert_wal_list_;
   uint32_t last_ts_;
 
   std::vector<UpdateWalUnit> update_wal_list_;
   std::vector<std::string> message_vector_;  // used to hold the polled messages
   cppkafka::Configuration config_;
+
+  static const bool registered_;
 };
 
 }  // namespace gs

flex/tests/hqps/kafka_wal_ingester_test.cc

@@ -0,0 +1,77 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <iostream>
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+#include <flex/engines/graph_db/database/wal/kafka_wal_parser.h>
+#include <flex/engines/graph_db/database/wal/kafka_wal_writer.h>
+#endif
+#include "grape/serialization/in_archive.h"
+#include "grape/serialization/out_archive.h"
+
+#include "flex/engines/graph_db/database/graph_db.h"
+#include "flex/engines/graph_db/database/graph_db_session.h"
+#include "flex/storages/rt_mutable_graph/schema.h"
+
+int main(int argc, char** argv) {
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+  gs::Schema schema;
+  schema.add_vertex_label(
+      "PERSON",
+      {
+          gs::PropertyType::kInt64,  // version
+      },
+      {"weight"},
+      {std::tuple<gs::PropertyType, std::string, size_t>(
+          gs::PropertyType::kInt64, "id", 0)},
+      {gs::StorageStrategy::kMem, gs::StorageStrategy::kMem}, 4096);
+  gs::GraphDB db;
+  std::string work_dir = argv[1];
+  std::string kafka_brokers = argv[2];
+  std::string kafka_topic = argv[3];
+  db.Open(schema, work_dir, 1);
+
+  gs::KafkaWalWriter writer(kafka_brokers);
+  writer.open(kafka_topic, 0);
+  grape::InArchive in_archive;
+  in_archive.Resize(sizeof(gs::WalHeader));
+  gs::label_t label = db.schema().get_vertex_label_id("PERSON");
+  in_archive << static_cast<uint8_t>(0) << label;
+  int64_t id = 998244353;
+  in_archive << id;
+  int64_t weight = 100;
+  in_archive << weight;
+  auto header = reinterpret_cast<gs::WalHeader*>(in_archive.GetBuffer());
+  header->timestamp = 1;
+  header->type = 0;
+  header->length = in_archive.GetSize() - sizeof(gs::WalHeader);
+  writer.append(in_archive.GetBuffer(), in_archive.GetSize());
+  writer.close();
+  cppkafka::Configuration config = {{"metadata.broker.list", kafka_brokers},
+                                    {"group.id", "test"},
+                                    {"enable.auto.commit", false}};
+  db.start_kafka_wal_ingester(config, kafka_topic);
+  std::this_thread::sleep_for(std::chrono::seconds(1));
+  db.stop_kafka_wal_ingester();
+  auto txn = db.GetReadTransaction(0);
+  gs::vid_t lid;
+  CHECK(txn.GetVertexNum(label) == 1);
+  CHECK(txn.GetVertexIndex(label, id, lid));
+  auto iter = txn.GetVertexIterator(label);
+  CHECK(iter.GetField(0).AsInt64() == 100);
+  std::cout << "Vertex id: " << lid << std::endl;
+  db.Close();
+#endif
+  return 0;
+}