diff --git a/.github/workflows/interactive.yml b/.github/workflows/interactive.yml
index b62a57073d28..5dde2d417aee 100644
--- a/.github/workflows/interactive.yml
+++ b/.github/workflows/interactive.yml
@@ -367,6 +367,28 @@ jobs:
         export ENGINE_TYPE=hiactor
         bash hqps_adhoc_test.sh ${INTERACTIVE_WORKSPACE} graph_algo RBO
 
+    - name: Run kafka wal test
+      env:
+        GS_TEST_DIR: ${{ github.workspace }}/gstest
+      run: |
+        git submodule update --init
+        cd ${GITHUB_WORKSPACE}/flex/build
+        cmake .. -DBUILD_KAFKA_WAL_WRITER_PARSER=ON && make -j$(nproc)
+        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
+        KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
+        bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties
+        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: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
+        ps aux | grep kafka | grep -v grep | awk '{print $2}' | xargs kill -9
+
     - name: Run Gremlin test on modern graph
       env:
         GS_TEST_DIR: ${{ github.workspace }}/gstest
diff --git a/docs/flex/interactive/development/dev_and_test.md b/docs/flex/interactive/development/dev_and_test.md
index e831f00e9591..52707165157f 100644
--- a/docs/flex/interactive/development/dev_and_test.md
+++ b/docs/flex/interactive/development/dev_and_test.md
@@ -262,3 +262,39 @@ In Interactive's execution engine, transactions such as `ReadTransaction`, `Upda
 2. If a transaction returns `false` during the `commit()` process, the error occurred prior to applying the WAL to the graph data. This type of failure could arise during the construction of the WAL or during its writing phase.
 
 3. It is important to note that errors can still occur when replaying the WAL to the graph database. Replaying might fail due to limitations in resources or due to unforeseen bugs. **However,** any errors encountered during this stage will be handled via exceptions or may result in process failure. Currently, there is no established mechanism to handle such failures. Future improvements should focus on implementing failover strategies, potentially allowing the GraphDB to continue replaying the WAL until it succeeds.
+
+## Persisting WAL to kafka
+
+Kafka-based WAL storage is also provided. Follows [kafka-quick-start](https://kafka.apache.org/quickstart).
+### Install kafka
+
+```bash
+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
+```
+
+### kafka with kraft
+
+```bash
+KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
+bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties
+bin/kafka-server-start.sh config/kraft/reconfig-server.properties
+```
+
+### Create topic
+
+```bash
+bin/kafka-topics.sh --create --topic kafka-test --bootstrap-server localhost:9092
+# describe the topic
+bin/kafka-topics.sh --describe --topic kafka-test --bootstrap-server localhost:9092
+```
+
+### Test KafkaWalWriter and KafkaWalParser
+
+```bash
+cd flex && mkdir build
+cd build && cmake .. -DBUILD_TEST=ON && make -j
+./tests/hqps/kafka_test localhost:902 kafka-test
+# run the kafka test
+```
diff --git a/flex/CMakeLists.txt b/flex/CMakeLists.txt
index f05e2190205e..f546c67f3215 100644
--- a/flex/CMakeLists.txt
+++ b/flex/CMakeLists.txt
@@ -17,6 +17,8 @@ option(USE_PTHASH "Whether to use pthash" OFF)
 option(OPTIMIZE_FOR_HOST "Whether to optimize on host" ON) # Whether to build optimized code on host
 option(USE_STATIC_ARROW "Whether to use static arrow" OFF) # Whether to link arrow statically, default is OFF
 option(BUILD_WITH_OTEL "Whether to build with opentelemetry-cpp" OFF) # Whether to build with opentelemetry-cpp, default is OFF
+
+option(BUILD_KAFKA_WAL_WRITER_PARSER "Whether to build kafka wal writer and wal parser" OFF) # Whether to build kafka wal writer and wal parser, default is OFF
 option(BUILD_WITH_OSS "Whether to build with oss support" OFF) # Whether to build with oss support, default is OFF
 
 #print options
@@ -129,6 +131,21 @@ if (NOT yaml-cpp_FOUND)
     message(FATAL_ERROR "yaml-cpp not found, please install the yaml-cpp library")
 endif ()
 
+#find CppKafka-------------------------------------------------------------------
+if (BUILD_KAFKA_WAL_WRITER_PARSER)
+    find_package(CppKafka)
+    if (NOT CppKafka_FOUND)
+        message(FATAL_ERROR "cppkafka not found, please install cppkafka library")
+    else()
+        include_directories(SYSTEM ${CppKafka_INCLUDE_DIRS})
+        message(STATUS "cpp kafka include dir: ${CppKafka_INCLUDE_DIRS}")
+        set(CppKafka_LIBRARIES CppKafka::cppkafka)
+        add_definitions(-DBUILD_KAFKA_WAL_WRITER_PARSER)
+        message(STATUS "cppkafka found")
+    endif ()
+endif()
+
+
 #find boost----------------------------------------------------------------------
 find_package(Boost REQUIRED COMPONENTS system filesystem
              # required by folly
diff --git a/flex/bin/rt_server.cc b/flex/bin/rt_server.cc
index c6bbbbc8cf00..91684c2e11c4 100644
--- a/flex/bin/rt_server.cc
+++ b/flex/bin/rt_server.cc
@@ -108,6 +108,7 @@ int main(int argc, char** argv) {
   service_config.query_port = http_port;
   service_config.start_admin_service = false;
   service_config.start_compiler = false;
+  service_config.wal_uri = config.wal_uri;
   service_config.set_sharding_mode(vm["sharding-mode"].as<std::string>());
   server::GraphDBService::get().init(service_config);
 
diff --git a/flex/engines/graph_db/CMakeLists.txt b/flex/engines/graph_db/CMakeLists.txt
index fb20dbf56750..740d3466c9c0 100644
--- a/flex/engines/graph_db/CMakeLists.txt
+++ b/flex/engines/graph_db/CMakeLists.txt
@@ -4,11 +4,19 @@ file(GLOB_RECURSE GRAPH_DB_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/app/*.cc"
                                      "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/*.cc"
                                      "${CMAKE_CURRENT_SOURCE_DIR}/app/builtin/*.cc")
 
+if (NOT BUILD_KAFKA_WAL_WRITER_PARSER)
+        list(REMOVE_ITEM GRAPH_DB_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/kafka_wal_writer.cc"
+                                            "${CMAKE_CURRENT_SOURCE_DIR}/database/wal/kafka_wal_parser.cc")   
+endif()
+
 add_library(flex_graph_db SHARED ${GRAPH_DB_SRC_FILES})
 
 target_include_directories(flex_graph_db PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>)
 target_link_libraries(flex_graph_db flex_rt_mutable_graph ${LIBGRAPELITE_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
 target_link_libraries(flex_graph_db runtime_execute)
+if (BUILD_KAFKA_WAL_WRITER_PARSER)
+        target_link_libraries(flex_graph_db ${CppKafka_LIBRARIES})
+endif()
 install_flex_target(flex_graph_db)
 
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/database/graph_db.h
diff --git a/flex/engines/graph_db/app/kafka_wal_ingester_app.cc b/flex/engines/graph_db/app/kafka_wal_ingester_app.cc
new file mode 100644
index 000000000000..8496aa63187e
--- /dev/null
+++ b/flex/engines/graph_db/app/kafka_wal_ingester_app.cc
@@ -0,0 +1,218 @@
+/** 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 "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_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_;
+  timestamp_t ingested_plus_one_;
+  std::vector<std::string> data_;
+  // 0: not exist, 1: exist, 2: ingested
+  std::vector<uint8_t> states_;
+
+  void resize() {
+    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 %= 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_(BUFFSIZ),
+        states_(BUFFSIZ, 0) {}
+
+  bool empty() const { return ingested_plus_one_ == end_; }
+
+  void ingest_impl(const std::string& data) {
+    auto header = reinterpret_cast<const WalHeader*>(data.data());
+    if (header->type == 0) {
+      InsertTransaction::IngestWal(
+          session_.graph(), header->timestamp,
+          const_cast<char*>(data.data()) + sizeof(WalHeader), header->length,
+          session_.allocator());
+    } else {
+      auto txn = session_.GetUpdateTransaction();
+      auto header = reinterpret_cast<const WalHeader*>(data.data());
+      txn.IngestWal(session_.graph(), session_.db().work_dir(),
+                    header->timestamp,
+                    const_cast<char*>(data.data()) + sizeof(WalHeader),
+                    header->length, session_.allocator());
+      txn.Commit();
+    }
+  }
+
+  void ingest() {
+    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) {
+        ingest_impl(data_[idx]);
+      }
+      states_[idx] = 0;
+      ++ingested_plus_one_;
+      ++idx;
+      idx %= len;
+      flag = true;
+    }
+    if (flag) {
+      session_.commit(ingested_plus_one_);
+    }
+  }
+  void push(const std::string& data) {
+    auto header = reinterpret_cast<const WalHeader*>(data.data());
+    if (header->timestamp < begin_) {
+      LOG(ERROR) << "Invalid timestamp: " << header->timestamp;
+      return;
+    }
+    size_t index;
+    size_t n = data_.size();
+    if (header->timestamp < end_) {
+      index = (header->timestamp - begin_) % n;
+    } else if (header->timestamp - begin_ < n) {
+      index = header->timestamp - begin_;
+      end_ = header->timestamp + 1;
+    } else {
+      ingest();
+      while (header->timestamp - ingested_plus_one_ + 1 > states_.size()) {
+        resize();
+      }
+      index = (header->timestamp - begin_) % states_.size();
+      end_ = header->timestamp + 1;
+    }
+    if (header->length == 0) {
+      states_[index] = 2;
+    } else if (header->type == 0) {
+      ingest_impl(data);
+      states_[index] = 2;
+    } else {
+      states_[index] = 1;
+      data_[index] = data;
+    }
+  }
+};
+
+class KafkaWalConsumer {
+ public:
+  static constexpr const std::chrono::milliseconds POLL_TIMEOUT =
+      std::chrono::milliseconds(100);
+
+  // always track all partitions and from begining
+  KafkaWalConsumer(WalIngester& ingester, cppkafka::Configuration config,
+                   const std::string& topic_name, int32_t thread_num)
+      : ingester_(ingester) {
+    auto topic_partitions = get_all_topic_partitions(config, topic_name, false);
+    consumers_.reserve(topic_partitions.size());
+    for (size_t i = 0; i < topic_partitions.size(); ++i) {
+      consumers_.emplace_back(std::make_unique<cppkafka::Consumer>(config));
+      consumers_.back()->assign({topic_partitions[i]});
+    }
+  }
+
+  void poll() {
+    for (auto& consumer : consumers_) {
+      auto msgs = consumer->poll_batch(1024);
+      for (const auto& msg : msgs) {
+        if (msg) {
+          if (msg.get_error()) {
+            if (!msg.is_eof()) {
+              LOG(INFO) << "[+] Received error notification: "
+                        << msg.get_error();
+            }
+          } else {
+            std::string payload = msg.get_payload();
+            ingester_.push(payload);
+            consumer->commit(msg);
+          }
+        }
+      }
+    }
+  }
+
+ private:
+  std::vector<std::unique_ptr<cppkafka::Consumer>> consumers_;
+  WalIngester& ingester_;
+};
+
+void Ingest(const std::string& data, GraphDBSession& session) {}
+
+bool KafkaWalIngesterApp::Query(GraphDBSession& graph, Decoder& input,
+                                Encoder& output) {
+  cppkafka::Configuration config;
+  std::string topic_name;
+  while (!input.empty()) {
+    auto key = input.get_string();
+    auto value = input.get_string();
+    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 (!force_stop_.load()) {
+    consumer.poll();
+    ingester.ingest();
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  }
+  while (!ingester.empty()) {
+    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);
+}
+#endif
+}  // namespace gs
diff --git a/flex/engines/graph_db/app/kafka_wal_ingester_app.h b/flex/engines/graph_db/app/kafka_wal_ingester_app.h
new file mode 100644
index 000000000000..1ac63bdd5cdc
--- /dev/null
+++ b/flex/engines/graph_db/app/kafka_wal_ingester_app.h
@@ -0,0 +1,48 @@
+/** 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.
+ */
+
+#ifndef ENGINES_KAFKA_WAL_INGESTER_APP_H_
+#define ENGINES_KAFKA_WAL_INGESTER_APP_H_
+
+#include "flex/engines/graph_db/app/app_base.h"
+#include "flex/engines/graph_db/database/graph_db_session.h"
+
+namespace gs {
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+// Ingest wal from kafka
+class KafkaWalIngesterApp : public WriteAppBase {
+ public:
+  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 {
+ public:
+  KafkaWalIngesterAppFactory() = default;
+  ~KafkaWalIngesterAppFactory() = default;
+
+  AppWrapper CreateApp(const GraphDB& db) override;
+};
+#endif
+
+}  // namespace gs
+
+#endif  // ENGINES_KAFKA_WAL_INGESTER_APP_H_
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/graph_db.cc b/flex/engines/graph_db/database/graph_db.cc
index 6c7508af5e43..4a80183fb82b 100644
--- a/flex/engines/graph_db/database/graph_db.cc
+++ b/flex/engines/graph_db/database/graph_db.cc
@@ -52,7 +52,10 @@ struct SessionLocalContext {
   char _padding2[4096 - sizeof(GraphDBSession) % 4096];
 };
 
-GraphDB::GraphDB() = default;
+GraphDB::GraphDB()
+    : monitor_thread_running_(false),
+      last_ingested_wal_ts_(0),
+      compact_thread_running_(false) {}
 GraphDB::~GraphDB() {
   if (compact_thread_running_) {
     compact_thread_running_ = false;
@@ -486,6 +489,7 @@ void GraphDB::openWalAndCreateContexts(const GraphDBConfig& config,
   }
   auto wal_parser = WalParserFactory::CreateWalParser(wal_uri);
   ingestWals(*wal_parser, data_dir, thread_num_);
+  last_ingested_wal_ts_ = wal_parser->last_ts();
 
   for (int i = 0; i < thread_num_; ++i) {
     contexts_[i].logger->open(wal_uri, i);
diff --git a/flex/engines/graph_db/database/graph_db.h b/flex/engines/graph_db/database/graph_db.h
index 2481d403c4d8..53844556eb89 100644
--- a/flex/engines/graph_db/database/graph_db.h
+++ b/flex/engines/graph_db/database/graph_db.h
@@ -169,6 +169,9 @@ class GraphDB {
 
   inline const GraphDBConfig& config() const { return config_; }
 
+  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; }
+
  private:
   bool registerApp(const std::string& path, uint8_t index = 0);
 
@@ -204,6 +207,8 @@ class GraphDB {
   std::thread monitor_thread_;
   bool monitor_thread_running_;
 
+  uint64_t last_ingested_wal_ts_;
+
   timestamp_t last_compaction_ts_;
   bool compact_thread_running_ = false;
   std::thread compact_thread_;
diff --git a/flex/engines/graph_db/database/graph_db_session.cc b/flex/engines/graph_db/database/graph_db_session.cc
index 03d34b4d43b4..b2b70a9d1187 100644
--- a/flex/engines/graph_db/database/graph_db_session.cc
+++ b/flex/engines/graph_db/database/graph_db_session.cc
@@ -67,6 +67,8 @@ const MutablePropertyFragment& GraphDBSession::graph() const {
 
 const GraphDB& GraphDBSession::db() const { return db_; }
 
+void GraphDBSession::commit(timestamp_t ts) { db_.version_manager_.commit(ts); }
+
 MutablePropertyFragment& GraphDBSession::graph() { return db_.graph(); }
 
 const Schema& GraphDBSession::schema() const { return db_.schema(); }
@@ -248,6 +250,8 @@ AppBase* GraphDBSession::GetApp(int type) {
   return app;
 }
 
+Allocator& GraphDBSession::allocator() { return alloc_; }
+
 #undef likely  // likely
 
 Result<std::pair<uint8_t, std::string_view>>
diff --git a/flex/engines/graph_db/database/graph_db_session.h b/flex/engines/graph_db/database/graph_db_session.h
index 84aeed45623a..8fce571c02cf 100644
--- a/flex/engines/graph_db/database/graph_db_session.h
+++ b/flex/engines/graph_db/database/graph_db_session.h
@@ -108,6 +108,10 @@ class GraphDBSession {
 
   AppBase* GetApp(const std::string& name);
 
+  Allocator& allocator();
+
+  void commit(timestamp_t ts);
+
  private:
   Result<std::pair<uint8_t, std::string_view>>
   parse_query_type_from_cypher_json(const std::string_view& input);
diff --git a/flex/engines/graph_db/database/insert_transaction.cc b/flex/engines/graph_db/database/insert_transaction.cc
index a6cc61a942dd..dea4afe769f9 100644
--- a/flex/engines/graph_db/database/insert_transaction.cc
+++ b/flex/engines/graph_db/database/insert_transaction.cc
@@ -173,6 +173,13 @@ bool InsertTransaction::Commit() {
 
 void InsertTransaction::Abort() {
   if (timestamp_ != std::numeric_limits<timestamp_t>::max()) {
+    auto header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
+    header->length = 0;
+    header->timestamp = timestamp_;
+    header->type = 0;
+    if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+      LOG(ERROR) << "Failed to append wal log";
+    }
     LOG(ERROR) << "aborting " << timestamp_ << "-th transaction (insert)";
     vm_.release_insert_timestamp(timestamp_);
     clear();
diff --git a/flex/engines/graph_db/database/single_edge_insert_transaction.cc b/flex/engines/graph_db/database/single_edge_insert_transaction.cc
index f9c00dca0772..f8f4b6a904e3 100644
--- a/flex/engines/graph_db/database/single_edge_insert_transaction.cc
+++ b/flex/engines/graph_db/database/single_edge_insert_transaction.cc
@@ -98,6 +98,13 @@ bool SingleEdgeInsertTransaction::AddEdge(label_t src_label, const Any& src,
 
 void SingleEdgeInsertTransaction::Abort() {
   if (timestamp_ != std::numeric_limits<timestamp_t>::max()) {
+    auto header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
+    header->length = 0;
+    header->timestamp = timestamp_;
+    header->type = 0;
+    if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+      LOG(ERROR) << "Failed to append wal log";
+    }
     LOG(ERROR) << "aborting " << timestamp_
                << "-th transaction (single edge insert)";
     vm_.release_insert_timestamp(timestamp_);
diff --git a/flex/engines/graph_db/database/single_vertex_insert_transaction.cc b/flex/engines/graph_db/database/single_vertex_insert_transaction.cc
index 188d1bfe72b4..c0c201c68430 100644
--- a/flex/engines/graph_db/database/single_vertex_insert_transaction.cc
+++ b/flex/engines/graph_db/database/single_vertex_insert_transaction.cc
@@ -178,6 +178,13 @@ bool SingleVertexInsertTransaction::Commit() {
 
 void SingleVertexInsertTransaction::Abort() {
   if (timestamp_ != std::numeric_limits<timestamp_t>::max()) {
+    auto header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
+    header->length = 0;
+    header->timestamp = timestamp_;
+    header->type = 0;
+    if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+      LOG(ERROR) << "Failed to append wal log";
+    }
     LOG(ERROR) << "aborting " << timestamp_
                << "-th transaction (single vertex insert)";
     vm_.release_insert_timestamp(timestamp_);
diff --git a/flex/engines/graph_db/database/update_transaction.cc b/flex/engines/graph_db/database/update_transaction.cc
index 7d8a53eefdb9..3ad80ec6043a 100644
--- a/flex/engines/graph_db/database/update_transaction.cc
+++ b/flex/engines/graph_db/database/update_transaction.cc
@@ -98,7 +98,7 @@ UpdateTransaction::UpdateTransaction(const GraphDBSession& session,
   updated_edge_data_.resize(csr_num);
 }
 
-UpdateTransaction::~UpdateTransaction() { release(); }
+UpdateTransaction::~UpdateTransaction() { Abort(); }
 
 timestamp_t UpdateTransaction::timestamp() const { return timestamp_; }
 
@@ -127,7 +127,18 @@ bool UpdateTransaction::Commit() {
   return true;
 }
 
-void UpdateTransaction::Abort() { release(); }
+void UpdateTransaction::Abort() {
+  if (timestamp_ != std::numeric_limits<timestamp_t>::max()) {
+    auto header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
+    header->length = 0;
+    header->timestamp = timestamp_;
+    header->type = 0;
+    if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+      LOG(ERROR) << "Failed to append wal log";
+    }
+    release();
+  }
+}
 
 bool UpdateTransaction::AddVertex(label_t label, const Any& oid,
                                   const std::vector<Any>& props) {
diff --git a/flex/engines/graph_db/database/version_manager.cc b/flex/engines/graph_db/database/version_manager.cc
index 9ff538b67505..38f261dba05d 100644
--- a/flex/engines/graph_db/database/version_manager.cc
+++ b/flex/engines/graph_db/database/version_manager.cc
@@ -98,6 +98,13 @@ void VersionManager::release_insert_timestamp(uint32_t ts) {
   pending_reqs_.fetch_sub(1);
 }
 
+void VersionManager::commit(uint32_t ts) {
+  lock_.lock();
+  read_ts_.store(ts);
+  write_ts_.store(ts + 1);
+  lock_.unlock();
+}
+
 uint32_t VersionManager::acquire_update_timestamp() {
   int expected_update_reqs = 0;
   while (
diff --git a/flex/engines/graph_db/database/version_manager.h b/flex/engines/graph_db/database/version_manager.h
index 2a7d4f3fd85a..41112087da1f 100644
--- a/flex/engines/graph_db/database/version_manager.h
+++ b/flex/engines/graph_db/database/version_manager.h
@@ -51,6 +51,8 @@ class VersionManager {
   void release_update_timestamp(uint32_t ts);
   bool revert_update_timestamp(uint32_t ts);
 
+  void commit(uint32_t ts);
+
  private:
   std::atomic<uint32_t> write_ts_{1};
   std::atomic<uint32_t> read_ts_{0};
diff --git a/flex/engines/graph_db/database/wal/kafka_wal_parser.cc b/flex/engines/graph_db/database/wal/kafka_wal_parser.cc
new file mode 100644
index 000000000000..694989fada79
--- /dev/null
+++ b/flex/engines/graph_db/database/wal/kafka_wal_parser.cc
@@ -0,0 +1,142 @@
+/** 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 "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::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"
+  auto res = parse_uri(uri);
+  if (!res) {
+    LOG(FATAL) << "Failed to parse uri: " << uri;
+    return nullptr;
+  }
+  gs::Decoder decoder(res.value().data(), res.value().size());
+  std::string topic_name;
+  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));
+    }
+  }
+  auto parser = std::unique_ptr<IWalParser>(new KafkaWalParser(config));
+  parser->open(topic_name);
+  return parser;
+}
+
+KafkaWalParser::KafkaWalParser(const cppkafka::Configuration& config)
+    : consumer_(nullptr), last_ts_(0), config_(config) {
+  consumer_ = std::make_unique<cppkafka::Consumer>(config);
+}
+
+void KafkaWalParser::open(const std::string& topic_name) {
+  auto topic_partitions = get_all_topic_partitions(config_, topic_name);
+  open(topic_partitions);
+}
+
+void KafkaWalParser::open(
+    const std::vector<cppkafka::TopicPartition>& topic_partitions) {
+  consumer_->assign(topic_partitions);
+  insert_wal_list_.resize(4096);
+  uint32_t cnt = 0;
+  while (true) {
+    auto msgs = consumer_->poll_batch(MAX_BATCH_SIZE);
+    if (msgs.empty() && cnt == last_ts_) {
+      LOG(INFO) << "No message are polled, the topic has been all consumed.";
+      break;
+    }
+    for (auto& msg : msgs) {
+      if (msg) {
+        if (msg.get_error()) {
+          if (!msg.is_eof()) {
+            LOG(INFO) << "[+] Received error notification: " << msg.get_error();
+          }
+        } else {
+          message_vector_.emplace_back(msg.get_payload());
+          const std::string& wal = message_vector_.back();
+          auto header = reinterpret_cast<const WalHeader*>(wal.data());
+          if (header->timestamp == 0) {
+            LOG(WARNING) << "Invalid timestamp 0, skip";
+            continue;
+          }
+          if (header->type) {
+            UpdateWalUnit unit;
+            unit.timestamp = header->timestamp;
+            unit.ptr = const_cast<char*>(wal.data() + sizeof(WalHeader));
+            unit.size = header->length;
+            update_wal_list_.push_back(unit);
+            cnt++;
+          } else {
+            if (header->timestamp >= insert_wal_list_.size()) {
+              insert_wal_list_.resize(header->timestamp + 1);
+            }
+            if (insert_wal_list_[header->timestamp].ptr) {
+              LOG(WARNING) << "Duplicated timestamp " << header->timestamp
+                           << ", skip";
+              continue;
+            }
+            cnt++;
+            insert_wal_list_[header->timestamp].ptr =
+                const_cast<char*>(wal.data() + sizeof(WalHeader));
+            insert_wal_list_[header->timestamp].size = header->length;
+          }
+          last_ts_ = std::max(header->timestamp, last_ts_);
+        }
+      }
+    }
+    consumer_->commit();
+  }
+
+  LOG(INFO) << "last_ts: " << last_ts_;
+  if (!update_wal_list_.empty()) {
+    std::sort(update_wal_list_.begin(), update_wal_list_.end(),
+              [](const UpdateWalUnit& lhs, const UpdateWalUnit& rhs) {
+                return lhs.timestamp < rhs.timestamp;
+              });
+  }
+}
+
+void KafkaWalParser::close() {
+  if (consumer_) {
+    consumer_.reset();
+  }
+  insert_wal_list_.clear();
+}
+
+uint32_t KafkaWalParser::last_ts() const { return last_ts_; }
+
+const WalContentUnit& KafkaWalParser::get_insert_wal(uint32_t ts) const {
+  if (insert_wal_list_[ts].ptr == NULL) {
+    LOG(WARNING) << "No WAL for timestamp " << ts;
+  }
+  return insert_wal_list_[ts];
+}
+const std::vector<UpdateWalUnit>& KafkaWalParser::get_update_wals() const {
+  return update_wal_list_;
+}
+
+const bool KafkaWalParser::registered_ = WalParserFactory::RegisterWalParser(
+    "kafka", static_cast<WalParserFactory::wal_parser_initializer_t>(
+                 &KafkaWalParser::Make));
+
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/wal/kafka_wal_parser.h b/flex/engines/graph_db/database/wal/kafka_wal_parser.h
new file mode 100644
index 000000000000..78eeaac401e6
--- /dev/null
+++ b/flex/engines/graph_db/database/wal/kafka_wal_parser.h
@@ -0,0 +1,61 @@
+/** 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.
+ */
+
+#ifndef ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_
+#define ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_
+
+#include <vector>
+#include "cppkafka/cppkafka.h"
+#include "flex/engines/graph_db/database/wal/wal.h"
+
+namespace gs {
+
+class KafkaWalParser : public IWalParser {
+ public:
+  static constexpr const std::chrono::milliseconds POLL_TIMEOUT =
+      std::chrono::milliseconds(100);
+  static constexpr const size_t MAX_BATCH_SIZE = 1000;
+
+  static std::unique_ptr<IWalParser> Make(const std::string&);
+
+  // always track all partitions and from begining
+  KafkaWalParser(const cppkafka::Configuration& config);
+  ~KafkaWalParser() { close(); }
+
+  void open(const std::string& topic_name) override;
+  void close() override;
+
+  uint32_t last_ts() const override;
+  const WalContentUnit& get_insert_wal(uint32_t ts) const override;
+  const std::vector<UpdateWalUnit>& get_update_wals() const override;
+
+  //////Kafka specific methods
+  void open(const std::vector<cppkafka::TopicPartition>& partitions);
+
+ private:
+  std::unique_ptr<cppkafka::Consumer> consumer_;
+  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
+
+#endif  // ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_PARSER_H_
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/wal/kafka_wal_utils.cc b/flex/engines/graph_db/database/wal/kafka_wal_utils.cc
new file mode 100644
index 000000000000..90d2430bd41d
--- /dev/null
+++ b/flex/engines/graph_db/database/wal/kafka_wal_utils.cc
@@ -0,0 +1,97 @@
+/** 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 "flex/engines/graph_db/database/wal/kafka_wal_utils.h"
+#include "glog/logging.h"
+
+namespace gs {
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+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::optional<std::vector<char>> parse_uri(const std::string& wal_uri) {
+  std::vector<char> buf;
+  gs::Encoder encoder(buf);
+
+  const std::string prefix = "kafka://";
+  if (wal_uri.find(prefix) != 0) {
+    LOG(ERROR) << "Invalid uri: " << wal_uri;
+    return std::nullopt;
+  }
+
+  std::string hosts_part = wal_uri.substr(prefix.length());
+  size_t query_pos = hosts_part.find('/');
+  std::string hosts;
+  std::string query;
+  hosts = hosts_part.substr(0, query_pos);
+  query = hosts_part.substr(query_pos + 1);
+
+  std::string kafka_brokers = hosts;
+  encoder.put_string(std::string("metadata.broker.list"));
+  encoder.put_string(kafka_brokers);
+  size_t top_pos = query.find('?');
+  std::string topic;
+  if (top_pos != std::string::npos) {
+    topic = query.substr(0, top_pos);
+    query = query.substr(top_pos + 1);
+  } else {
+    topic = query;
+    query = "";
+  }
+  encoder.put_string(std::string("topic_name"));
+  encoder.put_string(topic);
+  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);
+      encoder.put_string(key);
+      encoder.put_string(value);
+    }
+  }
+  encoder.put_string(std::string("enable.auto.commit"));
+  encoder.put_string(std::string("false"));
+
+  return buf;
+}
+#endif
+
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/wal/kafka_wal_utils.h b/flex/engines/graph_db/database/wal/kafka_wal_utils.h
new file mode 100644
index 000000000000..2354ed235ac9
--- /dev/null
+++ b/flex/engines/graph_db/database/wal/kafka_wal_utils.h
@@ -0,0 +1,38 @@
+/** 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.
+ */
+
+#ifndef FLEX_ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_UTILS_H_
+#define FLEX_ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_UTILS_H_
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+#include <optional>
+#include <string>
+#include <vector>
+#include "cppkafka/cppkafka.h"
+#include "flex/utils/app_utils.h"
+
+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);
+
+std::optional<std::vector<char>> parse_uri(const std::string& wal_uri);
+
+}  // namespace gs
+#endif
+
+#endif  // FLEX_ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_UTILS_H_
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/wal/kafka_wal_writer.cc b/flex/engines/graph_db/database/wal/kafka_wal_writer.cc
new file mode 100644
index 000000000000..340befc4ed31
--- /dev/null
+++ b/flex/engines/graph_db/database/wal/kafka_wal_writer.cc
@@ -0,0 +1,89 @@
+/** 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 "flex/engines/graph_db/database/wal/kafka_wal_writer.h"
+#include "flex/engines/graph_db/database/wal/kafka_wal_utils.h"
+#include "flex/engines/graph_db/database/wal/wal.h"
+
+#include <chrono>
+#include <filesystem>
+
+namespace gs {
+
+std::unique_ptr<IWalWriter> KafkaWalWriter::Make() {
+  return std::unique_ptr<IWalWriter>(new KafkaWalWriter());
+}
+
+void KafkaWalWriter::open(const std::string& uri, int thread_id) {
+  auto res = parse_uri(uri);
+  if (!res) {
+    LOG(FATAL) << "Failed to parse uri: " << uri;
+  }
+  gs::Decoder decoder(res.value().data(), res.value().size());
+  while (!decoder.empty()) {
+    auto key = decoder.get_string();
+    auto value = decoder.get_string();
+    if (key == "metadata.broker.list") {
+      kafka_brokers_ = value;
+    } else if (key == "topic_name") {
+      kafka_topic_ = value;
+    }
+  }
+
+  if (thread_id_ != -1 || producer_) {
+    LOG(FATAL) << "KafkaWalWriter has been opened";
+  }
+  thread_id_ = thread_id;
+  if (!kafka_brokers_.empty()) {
+    if (kafka_topic_.empty()) {
+      LOG(FATAL) << "Kafka topic is empty";
+    }
+    cppkafka::Configuration config = {{"metadata.broker.list", kafka_brokers_}};
+    producer_ =
+        std::make_shared<cppkafka::BufferedProducer<std::string>>(config);
+    producer_->set_max_number_retries(DEFAULT_MAX_RETRIES);
+    builder_.topic(kafka_topic_).partition(thread_id_);
+  } else {
+    LOG(FATAL) << "Kafka brokers is empty";
+  }
+}
+
+void KafkaWalWriter::close() {
+  if (producer_) {
+    producer_->flush();
+    producer_.reset();
+    thread_id_ = -1;
+    kafka_topic_.clear();
+  }
+}
+
+bool KafkaWalWriter::append(const char* data, size_t length) {
+  try {
+    producer_->sync_produce(builder_.payload({data, length}));
+    producer_->flush(true);
+  } catch (const cppkafka::HandleException& e) {
+    LOG(ERROR) << "Failed to send to kafka: " << e.what();
+    return false;
+  }
+  VLOG(10) << "Finished sending to kafka with message size: " << length
+           << ", partition: " << thread_id_;
+  return true;
+}
+
+const bool KafkaWalWriter::registered_ = WalWriterFactory::RegisterWalWriter(
+    "kafka", static_cast<WalWriterFactory::wal_writer_initializer_t>(
+                 &KafkaWalWriter::Make));
+
+}  // namespace gs
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/wal/kafka_wal_writer.h b/flex/engines/graph_db/database/wal/kafka_wal_writer.h
new file mode 100644
index 000000000000..83e98afe466b
--- /dev/null
+++ b/flex/engines/graph_db/database/wal/kafka_wal_writer.h
@@ -0,0 +1,78 @@
+/** 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.
+ */
+
+#ifndef ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_WRITER_H_
+#define ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_WRITER_H_
+
+#include <memory>
+#include <unordered_map>
+#include "flex/engines/graph_db/database/wal/wal.h"
+
+#include "cppkafka/cppkafka.h"
+
+namespace gs {
+
+/**
+ * KafkaWalWriter is a concrete implementation of IWalWriter that writes WAL
+ * entries to a Kafka topic.
+ * It uses cppkafka to produce messages to Kafka.
+ * The Kafka brokers are specified by the environment variable
+ * KAFKA_BROKER_LIST. The Kafka topic is specified by the open() method, the
+ * thread_id is used as the partition number.
+ *
+ * After restarting the GraphDB service, the KafkaWalWriter will continue to
+ * write to the same Kafka topic and partition. Consumers should be able to
+ * select the WAL entries by the timestamp.
+ */
+class KafkaWalWriter : public IWalWriter {
+ public:
+  static constexpr const int DEFAULT_MAX_RETRIES = 10;
+
+  static std::unique_ptr<IWalWriter> Make();
+
+  KafkaWalWriter()
+      : thread_id_(-1),
+        kafka_brokers_(""),
+        kafka_topic_(""),
+        producer_(nullptr),
+        builder_("") {}  // brokers could be a list of brokers
+
+  ~KafkaWalWriter() { close(); }
+
+  void open(const std::string& kafka_topic, int thread_id) override;
+  void close() override;
+  bool append(const char* data, size_t length) override;
+  std::string type() const override { return "kafka"; }
+
+  //////Kafka specific methods
+
+  inline std::tuple<int64_t, int64_t> getCurrentOffset() {
+    return producer_->get_producer().query_offsets(
+        cppkafka::TopicPartition(kafka_topic_, thread_id_));
+  }
+
+ private:
+  int32_t thread_id_;
+  std::string kafka_brokers_;
+  std::string kafka_topic_;
+  std::shared_ptr<cppkafka::BufferedProducer<std::string>> producer_;
+  cppkafka::MessageBuilder builder_;
+
+  static const bool registered_;
+};
+
+}  // namespace gs
+
+#endif  // ENGINES_GRAPH_DB_DATABASE_WAL_KAFKA_WAL_WRITER_H_
\ No newline at end of file
diff --git a/flex/engines/graph_db/database/wal/wal.cc b/flex/engines/graph_db/database/wal/wal.cc
index 14d9c147651d..3053c3191185 100644
--- a/flex/engines/graph_db/database/wal/wal.cc
+++ b/flex/engines/graph_db/database/wal/wal.cc
@@ -101,7 +101,7 @@ std::unique_ptr<IWalParser> WalParserFactory::CreateWalParser(
 bool WalParserFactory::RegisterWalParser(
     const std::string& wal_writer_type,
     WalParserFactory::wal_parser_initializer_t initializer) {
-  LOG(INFO) << "Registering wal writer of type: " << wal_writer_type;
+  LOG(INFO) << "Registering wal parser of type: " << wal_writer_type;
   auto& known_parsers_ = getKnownWalParsers();
   known_parsers_.emplace(wal_writer_type, initializer);
   return true;
diff --git a/flex/engines/http_server/graph_db_service.cc b/flex/engines/http_server/graph_db_service.cc
index eb27be41b1e4..025be75d4e35 100644
--- a/flex/engines/http_server/graph_db_service.cc
+++ b/flex/engines/http_server/graph_db_service.cc
@@ -425,6 +425,23 @@ bool GraphDBService::stop_compiler_subprocess() {
   return true;
 }
 
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+bool GraphDBService::start_kafka_wal_ingester() {
+  kafka_wal_ingester_ = std::make_unique<KafkaWalIngester>();
+  kafka_wal_ingester_->open(gs::GraphDB::get(), service_config_.wal_uri);
+  return true;
+}
+
+bool GraphDBService::stop_kafka_wal_ingester() {
+  if (kafka_wal_ingester_) {
+    kafka_wal_ingester_->close();
+    kafka_wal_ingester_.reset();
+    return true;
+  }
+  return false;
+}
+#endif
+
 std::string GraphDBService::find_interactive_class_path() {
   std::string interactive_home = DEFAULT_INTERACTIVE_HOME;
   if (std::getenv("INTERACTIVE_HOME")) {
diff --git a/flex/engines/http_server/graph_db_service.h b/flex/engines/http_server/graph_db_service.h
index 0689cf1003ba..56cefcc1f6ff 100644
--- a/flex/engines/http_server/graph_db_service.h
+++ b/flex/engines/http_server/graph_db_service.h
@@ -28,6 +28,9 @@
 #include "flex/storages/metadata/metadata_store_factory.h"
 #include "flex/utils/result.h"
 #include "flex/utils/service_utils.h"
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+#include "flex/engines/http_server/kafka_wal_ingester.h"
+#endif
 
 #include <yaml-cpp/yaml.h>
 #include <boost/process.hpp>
@@ -175,6 +178,12 @@ class GraphDBService {
 
   bool check_compiler_ready() const;
 
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+  bool start_kafka_wal_ingester();
+
+  bool stop_kafka_wal_ingester();
+#endif
+
  private:
   GraphDBService() = default;
 
@@ -197,6 +206,9 @@ class GraphDBService {
   boost::process::child compiler_process_;
   // handler for metadata store
   std::shared_ptr<gs::IGraphMetaStore> metadata_store_;
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+  std::unique_ptr<KafkaWalIngester> kafka_wal_ingester_;
+#endif
 };
 
 }  // namespace server
diff --git a/flex/engines/http_server/kafka_wal_ingester.cc b/flex/engines/http_server/kafka_wal_ingester.cc
new file mode 100644
index 000000000000..cf6ab53b5f2b
--- /dev/null
+++ b/flex/engines/http_server/kafka_wal_ingester.cc
@@ -0,0 +1,52 @@
+/** 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 "flex/engines/http_server/kafka_wal_ingester.h"
+#include "flex/engines/graph_db/database/graph_db.h"
+#include "flex/engines/graph_db/database/graph_db_session.h"
+#include "flex/engines/graph_db/database/wal/kafka_wal_utils.h"
+
+namespace server {
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+
+bool KafkaWalIngester::open(gs::GraphDB& db, const std::string& wal_uri) {
+  ingester_ = std::make_unique<gs::KafkaWalIngesterApp>();
+  ingester_thread_ = std::thread([&]() {
+    auto res = gs::parse_uri(wal_uri);
+    if (!res) {
+      LOG(ERROR) << "Failed to parse uri: " << wal_uri;
+      return;
+    }
+    gs::Decoder decoder(res.value().data(), res.value().size());
+    std::vector<char> buf;
+    gs::Encoder encoder(buf);
+    ingester_->Query(db.GetSession(0), decoder, encoder);
+    gs::Decoder output(buf.data(), buf.size());
+    db.set_last_ingested_wal_ts(output.get_long());
+  });
+  return true;
+}
+
+bool KafkaWalIngester::close() {
+  if (ingester_) {
+    ingester_->terminal();
+  }
+  if (ingester_thread_.joinable()) {
+    ingester_thread_.join();
+  }
+  return true;
+}
+#endif
+
+}  // namespace server
\ No newline at end of file
diff --git a/flex/engines/http_server/kafka_wal_ingester.h b/flex/engines/http_server/kafka_wal_ingester.h
new file mode 100644
index 000000000000..16f02b0460c8
--- /dev/null
+++ b/flex/engines/http_server/kafka_wal_ingester.h
@@ -0,0 +1,34 @@
+/** 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 <cppkafka/cppkafka.h>
+#include <thread>
+#include <vector>
+
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+#include "flex/engines/graph_db/app/kafka_wal_ingester_app.h"
+namespace server {
+class KafkaWalIngester {
+ public:
+  KafkaWalIngester() : force_stop_(false), ingester_(nullptr) {}
+
+  bool open(gs::GraphDB&, const std::string& uri);
+
+  bool close();
+  std::atomic<bool> force_stop_{false};
+  std::unique_ptr<gs::KafkaWalIngesterApp> ingester_;
+  std::thread ingester_thread_;
+};
+}  // namespace server
+#endif
\ No newline at end of file
diff --git a/flex/tests/hqps/CMakeLists.txt b/flex/tests/hqps/CMakeLists.txt
index 2aa5ac05bf61..857f9dce7e06 100644
--- a/flex/tests/hqps/CMakeLists.txt
+++ b/flex/tests/hqps/CMakeLists.txt
@@ -10,7 +10,7 @@ foreach(f ${GS_TEST_FILES})
         set(T_NAME ${CMAKE_MATCH_1})
         message(STATUS "Found graphscope test - " ${T_NAME})
         add_executable(${T_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/${T_NAME}.cc)
-        target_link_libraries(${T_NAME} flex_plan_proto  flex_graph_db flex_utils)
+        target_link_libraries(${T_NAME} flex_plan_proto  flex_graph_db flex_utils flex_server)
         if (BUILD_WITH_OSS)
                 target_link_libraries(${T_NAME} cpp-sdk)
         endif()
diff --git a/flex/tests/hqps/kafka_test.cc b/flex/tests/hqps/kafka_test.cc
new file mode 100644
index 000000000000..168124f7e2d1
--- /dev/null
+++ b/flex/tests/hqps/kafka_test.cc
@@ -0,0 +1,91 @@
+/** 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>
+
+#include "grape/serialization/in_archive.h"
+#include "grape/serialization/out_archive.h"
+#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 <glog/logging.h>
+
+int main(int argc, char** argv) {
+  if (argc < 3) {
+    std::cerr << "usage: kafka_test <kafka_brokers> <kafka_topic> "
+              << std::endl;
+    return -1;
+  }
+#ifdef BUILD_KAFKA_WAL_WRITER_PARSER
+
+  std::string kafka_brokers = argv[1];
+  std::string kafka_topic = argv[2];
+  LOG(INFO) << "Kafka brokers: " << kafka_brokers;
+
+  // Write messages to the specified kafka topic, and read them back.
+  gs::KafkaWalWriter writer;
+  cppkafka::Configuration config = {{"metadata.broker.list", kafka_brokers},
+                                    {"group.id", "test"},
+                                    {"enable.auto.commit", false}};
+  gs::KafkaWalParser parser(config);
+  std::string uri = "kafka://" + kafka_brokers + "/" + kafka_topic;
+  writer.open(uri, 0);
+
+  // Let user enter number of messages to write
+  LOG(INFO) << "Enter number of messages to write: ";
+  int num_messages = 3;
+  for (int i = 0; i < num_messages; ++i) {
+    std::string message = "message " + std::to_string(i);
+    grape::InArchive in_archive;
+    in_archive.Resize(sizeof(gs::WalHeader));
+    in_archive << message;
+    auto header = reinterpret_cast<gs::WalHeader*>(in_archive.GetBuffer());
+    header->timestamp = i + 1;
+    header->type = 0;
+    header->length = in_archive.GetSize() - sizeof(gs::WalHeader);
+    LOG(INFO) << "Writing wal: " << header->timestamp << ", " << header->length;
+    writer.append(in_archive.GetBuffer(), in_archive.GetSize());
+  }
+  auto offset = writer.getCurrentOffset();
+  LOG(INFO) << "Current offset: " << std::get<0>(offset) << ", "
+            << std::get<1>(offset);
+  writer.close();
+  LOG(INFO) << "Messages have been written to Kafka topic: " << argv[2]
+            << std::endl;
+
+  // Digest the messages
+  std::vector<cppkafka::TopicPartition> topic_partitions = {
+      {kafka_topic, 0, 0}};
+
+  // parser.open(topic_partitions);
+  parser.open(kafka_topic);
+  auto last_ts = parser.last_ts();
+  for (int i = 0; i < num_messages; ++i) {
+    const gs::WalContentUnit& wal = parser.get_insert_wal(last_ts + i);
+    if (wal.ptr) {
+      grape::OutArchive out_archive;
+      out_archive.SetSlice(wal.ptr, wal.size);
+      std::string message;
+      out_archive >> message;
+      LOG(INFO) << "Read message: " << message;
+    } else {
+      LOG(ERROR) << "No message for timestamp " << last_ts + i;
+    }
+  }
+#endif
+
+  return 0;
+}
\ No newline at end of file
diff --git a/flex/tests/hqps/kafka_wal_ingester_test.cc b/flex/tests/hqps/kafka_wal_ingester_test.cc
new file mode 100644
index 000000000000..2b32b8d0fb95
--- /dev/null
+++ b/flex/tests/hqps/kafka_wal_ingester_test.cc
@@ -0,0 +1,128 @@
+/** 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>
+#include <random>
+#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>
+#include "flex/engines/http_server/kafka_wal_ingester.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];
+  gs::GraphDBConfig config(schema, work_dir, "", 1);
+  std::string uri =
+      "kafka://" + kafka_brokers + "/" + kafka_topic + "?group.id=test";
+  config.wal_uri = uri;
+  db.Open(config);
+  {
+    for (int i = 0; i < 100; ++i) {
+      auto txn = db.GetInsertTransaction(0);
+      gs::label_t label = db.schema().get_vertex_label_id("PERSON");
+      int64_t id = i;
+      int64_t weight = i * 2 + 1;
+      txn.AddVertex(label, id, {gs::Any(weight)});
+      txn.Commit();
+    }
+  }
+
+  gs::GraphDB db2;
+  db2.Open(config);
+
+  CHECK(db2.GetReadTransaction(0).GetVertexNum(0) == 100)
+      << "Vertex num: " << db2.GetReadTransaction(0).GetVertexNum(0);
+  cppkafka::Configuration config1 = {{"metadata.broker.list", kafka_brokers},
+                                     {"group.id", "test"},
+                                     {"enable.auto.commit", false},
+                                     {"auto.offset.reset", "earliest"}};
+  server::KafkaWalIngester ingester;
+  ingester.open(db2, uri);
+  {
+    std::vector<std::thread> threads;
+    for (int i = 100; i < 200; ++i) {
+      threads.emplace_back([&db, i] {
+        auto txn = db.GetInsertTransaction(0);
+        gs::label_t label = db.schema().get_vertex_label_id("PERSON");
+        int64_t id = i;
+        int64_t weight = 200;
+        txn.AddVertex(label, id, {gs::Any(weight)});
+        std::random_device rd;
+        std::mt19937 gen(rd());
+        std::this_thread::sleep_for(std::chrono::milliseconds(gen() % 1000));
+        if (i % 20 == 0) {
+          txn.Abort();
+        } else {
+          txn.Commit();
+        }
+        {
+          auto txn = db.GetUpdateTransaction(0);
+          txn.AddVertex(label, id - 100, {gs::Any(weight - 1)});
+          txn.Commit();
+        }
+      });
+    }
+    for (auto& thrd : threads) {
+      thrd.join();
+    }
+  }
+  LOG(INFO) << db.GetReadTransaction(0).GetVertexNum(0);
+
+  std::this_thread::sleep_for(std::chrono::seconds(4));
+  {
+    auto txn = db2.GetReadTransaction(0);
+    CHECK(txn.GetVertexNum(0) == 195) << "Vertex num: " << txn.GetVertexNum(0);
+    gs::vid_t lid;
+    CHECK(txn.GetVertexIndex(0, 90L, lid));
+    LOG(INFO) << "Vertex id: " << lid;
+    CHECK(txn.GetVertexIndex(0, 188L, lid));
+    LOG(INFO) << "Vertex id: " << lid;
+    auto iter = txn.GetVertexIterator(0);
+    int cnt = 0;
+    while (iter.IsValid()) {
+      if (cnt < 100) {
+        CHECK(iter.GetField(0).AsInt64() == 199);
+      } else {
+        CHECK(iter.GetField(0).AsInt64() == 200);
+      }
+      cnt++;
+      iter.Next();
+    }
+    ingester.close();
+
+    // db2.stop_kafka_wal_ingester();
+  }
+#endif
+  return 0;
+}
\ No newline at end of file
diff --git a/k8s/dockerfiles/flex-interactive.Dockerfile b/k8s/dockerfiles/flex-interactive.Dockerfile
index 72447d90fd69..b86151cc09ea 100644
--- a/k8s/dockerfiles/flex-interactive.Dockerfile
+++ b/k8s/dockerfiles/flex-interactive.Dockerfile
@@ -27,7 +27,9 @@ COPY --chown=graphscope:graphscope . /home/graphscope/GraphScope
 
 # install flex
 RUN . ${HOME}/.cargo/env  && cd ${HOME}/GraphScope/flex && \
-    git submodule update --init && mkdir build && cd build && cmake .. -DCMAKE_INSTALL_PREFIX=/opt/flex -DBUILD_DOC=OFF -DBUILD_TEST=OFF -DOPTIMIZE_FOR_HOST=${OPTIMIZE_FOR_HOST} -DUSE_STATIC_ARROW=ON && make -j ${PARALLEL} && make install && \
+    git submodule update --init && mkdir build && cd build && \
+    cmake .. -DCMAKE_INSTALL_PREFIX=/opt/flex -DBUILD_DOC=OFF -DBUILD_TEST=OFF \
+        -DOPTIMIZE_FOR_HOST=${OPTIMIZE_FOR_HOST} -DUSE_STATIC_ARROW=ON -DBUILD_KAFKA_WAL_WRITER_PARSER=ON && make -j ${PARALLEL} && make install && \
     cd ~/GraphScope/interactive_engine/ && mvn clean package -Pexperimental -DskipTests && \
     cd ~/GraphScope/interactive_engine/compiler && cp target/compiler-0.0.1-SNAPSHOT.jar /opt/flex/lib/ && \
     cp target/libs/*.jar /opt/flex/lib/ && \
@@ -103,6 +105,7 @@ RUN find /opt/flex/lib/ -name "*.a" -type f -delete
 # include 
 COPY --from=builder /opt/flex/include/ /opt/graphscope/include/ /opt/vineyard/include/ /opt/flex/include/
 COPY --from=builder /opt/graphscope/lib/libgrape-lite.so /opt/flex/lib/
+COPY --from=builder /opt/graphscope/lib/libcppkafka.so* /opt/flex/lib/
 
 # copy the builtin graph, modern_graph
 RUN mkdir -p /opt/flex/share/gs_interactive_default_graph/
@@ -146,6 +149,7 @@ COPY --from=builder /usr/lib/$PLATFORM-linux-gnu/libprotobuf* /usr/lib/$PLATFORM
     /usr/lib/$PLATFORM-linux-gnu/libglog*.so* \
     /usr/lib/$PLATFORM-linux-gnu/libgflags*.so* \
     /usr/lib/$PLATFORM-linux-gnu/libicudata.so* \
+    /usr/lib/$PLATFORM-linux-gnu/librdkafka* \
     /usr/lib/$PLATFORM-linux-gnu/
 
 RUN sudo rm -rf /usr/lib/$PLATFORM-linux-gnu/libLLVM*.so* && sudo rm -rf /opt/flex/lib/libseastar.a && \
diff --git a/python/graphscope/gsctl/scripts/install_deps.sh b/python/graphscope/gsctl/scripts/install_deps.sh
index 397e3b76b6a0..a8169c532b32 100755
--- a/python/graphscope/gsctl/scripts/install_deps.sh
+++ b/python/graphscope/gsctl/scripts/install_deps.sh
@@ -866,9 +866,9 @@ install_analytical_java_dependencies() {
   fi
 }
 
-INTERACTIVE_MACOS=("apache-arrow" "rapidjson" "boost" "glog" "gflags")
-INTERACTIVE_UBUNTU=("rapidjson-dev" "libgoogle-glog-dev" "libgflags-dev")
-INTERACTIVE_CENTOS=("rapidjson-devel")
+INTERACTIVE_MACOS=("apache-arrow" "rapidjson" "boost" "glog" "gflags" "librdkafka")
+INTERACTIVE_UBUNTU=("rapidjson-dev" "libgoogle-glog-dev" "libgflags-dev" "librdkafka-dev")
+INTERACTIVE_CENTOS=("rapidjson-devel" "librdkafka-devel")
 
 install_interactive_dependencies() {
   # dependencies package
@@ -881,6 +881,7 @@ install_interactive_dependencies() {
     # hiactor is only supported on ubuntu
     install_hiactor
     install_mimalloc
+    install_cppkafka
     ${SUDO} sh -c 'echo "fs.aio-max-nr = 1048576" >> /etc/sysctl.conf'
     ${SUDO} sysctl -p /etc/sysctl.conf
   else
@@ -888,6 +889,7 @@ install_interactive_dependencies() {
     install_arrow
     install_boost
     install_mimalloc
+    install_cppkafka
   fi
   # libgrape-lite
   install_libgrape_lite "v0.3.2"