add reader test

Committed-by: xiaolei.zl from Dev container

.github/workflows/interactive.yml

@@ -568,8 +568,8 @@ jobs:
 
         # write to the kafka brokers : localhost:9092
         cd ${GITHUB_WORKSPACE}/flex/build/
-        ./tests/wal_writer/wal_writer_test localhost:9092 kafka topic_1 1 10
-        ./bin/wal_consumer -b localhost:9092 -t topic_1
+        ./tests/wal/wal_writer_test localhost:9092 kafka topic_1 1 10
+        ./tests/wal/wal_reader_test localhost:9092 kafka topic_1 1
 
         # kill the kafka server
         kill $(ps aux | grep 'kafka_2.13-3.8.0' | awk '{print $2}')

docs/flex/interactive/development/dev_and_test.md

@@ -124,8 +124,12 @@ You need to deploy a kafka cluster first. For details, please refer to [Kafka Do
 
 ##### Settings
 
+We Compare the performance of `LocalWalWriter` and `KafkaWalWriter` on a host with 
+
 ##### Producing Wals
 
+
+
 ##### Consuming Wals
 
 ## Testing

flex/tests/CMakeLists.txt

@@ -1,5 +1,5 @@
 add_subdirectory(hqps)
 add_subdirectory(rt_mutable_graph)
 if (BUILD_KAFKA_WAL_WRITER)
-    add_subdirectory(wal_writer)
+    add_subdirectory(wal)
 endif()

flex/tests/wal/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_executable(wal_writer_test ${CMAKE_CURRENT_SOURCE_DIR}/wal_writer_test.cc)
+target_link_libraries(wal_writer_test PUBLIC ${CppKafka_LIBRARIES} ${LIBGRAPELITE_LIBRARIES} flex_graph_db)
+
+add_executable(wal_reader_test ${CMAKE_CURRENT_SOURCE_DIR}/wal_reader_test.cc)
+target_link_libraries(wal_reader_test PUBLIC ${CppKafka_LIBRARIES} ${LIBGRAPELITE_LIBRARIES} flex_graph_db)

flex/tests/wal/wal_reader_test.cc

@@ -0,0 +1,48 @@
+
+#include "cppkafka/cppkafka.h"
+#include "librdkafka/rdkafka.h"
+
+#include <iostream>
+#include <string>
+#include "flex/engines/graph_db/database/wal.h"
+#include "grape/grape.h"
+
+int main(int argc, char** argv) {
+  if (argc != 5) {
+    std::cerr << "Usage: " << argv[0]
+              << "<kafka brokers> <local/kafka> <topic/directory> <thread_num>"
+              << std::endl;
+    return 1;
+  }
+  std::string brokers = argv[1];
+  std::string type = argv[2];
+  std::string topic_name = argv[3];
+  int thread_num = std::stoi(argv[4]);
+
+  double t = -grape::GetCurrentTime();
+
+  if (type == "local") {
+    std::cout << "Consuming message from directory " << topic_name
+              << " thread num " << thread_num;
+    std::vector<std::string> wals;
+    for (const auto& entry : std::filesystem::directory_iterator(topic_name)) {
+      wals.push_back(entry.path().string());
+    }
+    std::unique_ptr<gs::IWalsParser> parser =
+        std::make_unique<gs::LocalWalsParser>(wals);
+  } else {
+    std::cout << "Consuming message from topic " << topic_name << " thread num "
+              << thread_num;
+    cppkafka::Configuration config = {{"metadata.broker.list", brokers},
+                                      {"group.id", "primary_group"},
+                                      // Disable auto commit
+                                      {"enable.auto.commit", false}};
+    std::unique_ptr<gs::IWalsParser> parser =
+        std::make_unique<gs::KafkaWalsParser>(config, topic_name);
+  }
+
+  t += grape::GetCurrentTime();
+  std::cout << "Consuming message took " << t << " seconds" << std::endl;
+
+  return 0;
+}

flex/tests/wal_writer/wal_writer_test.cc → flex/tests/wal/wal_writer_test.cc

@@ -37,9 +37,8 @@ void test_local_wal_writer(const std::string& topic_name, int thread_num,
     kafka_writers.emplace_back(new gs::LocalWalWriter());
   }
   for (int i = 0; i < thread_num; ++i) {
-    std::string dst_path = "/tmp/";
     // check whether files exist
-    kafka_writers[i]->open(dst_path, i);
+    kafka_writers[i]->open(topic_name, i);
   }
   run(kafka_writers, payload, message_cnt);
 }