Add Streams on top of Kafka Consumer (#172)

* Stop the Consumer grafefully when it is destroyed * Add Streams * Add Streams to InterpreterContext * Remove options to limit processed batches in Consumer * Add Streams unit tests * Stop waiting for a full batch if the Consumer stopped * Add ReadLock functionality to Synchronized * Use per Consumer-based locking * Replace shared_mutex with RWLock
2021-06-22 08:43:19 +02:00
parent 4004e94ca1
commit d6a6d280dd
23 changed files with 1007 additions and 112 deletions
--- a/tests/benchmark/expansion.cpp
+++ b/tests/benchmark/expansion.cpp
@@ -37,7 +37,7 @@ class ExpansionBenchFixture : public benchmark::Fixture {

    MG_ASSERT(db->CreateIndex(label));

-    interpreter_context.emplace(&*db, query::InterpreterConfig{}, data_directory);
+    interpreter_context.emplace(&*db, query::InterpreterConfig{}, data_directory, "non existing bootstrap servers");
    interpreter.emplace(&*interpreter_context);
  }

--- a/tests/manual/single_query.cpp
+++ b/tests/manual/single_query.cpp
@@ -17,7 +17,8 @@ int main(int argc, char *argv[]) {
  storage::Storage db;
  auto data_directory = std::filesystem::temp_directory_path() / "single_query_test";
  utils::OnScopeExit([&data_directory] { std::filesystem::remove_all(data_directory); });
-  query::InterpreterContext interpreter_context{&db, query::InterpreterConfig{}, data_directory};
+  query::InterpreterContext interpreter_context{&db, query::InterpreterConfig{}, data_directory,
+                                                "non existing bootstrap servers"};
  query::Interpreter interpreter{&interpreter_context};

  ResultStreamFaker stream(&db);
--- a/tests/unit/CMakeLists.txt
+++ b/tests/unit/CMakeLists.txt
@@ -44,6 +44,19 @@ function(_add_unit_test test_cpp custom_main)
 endfunction(_add_unit_test)


+# Test integrations-kafka
+
+add_library(kafka-mock STATIC kafka_mock.cpp)
+target_link_libraries(kafka-mock mg-utils librdkafka++ librdkafka Threads::Threads zlib gtest)
+# Include directories are intentionally not set, because kafka-mock isn't meant to be used apart from unit tests
+
+add_unit_test(integrations_kafka_consumer.cpp kafka_mock.cpp)
+target_link_libraries(${test_prefix}integrations_kafka_consumer kafka-mock mg-integrations-kafka)
+
+add_unit_test(mgp_kafka_c_api.cpp)
+target_link_libraries(${test_prefix}mgp_kafka_c_api mg-query mg-integrations-kafka)
+
+
 # Test mg-query

 add_unit_test(bfs_single_node.cpp)
@@ -101,6 +114,9 @@ target_link_libraries(${test_prefix}query_trigger mg-query)
 add_unit_test(query_serialization_property_value.cpp)
 target_link_libraries(${test_prefix}query_serialization_property_value mg-query)

+add_unit_test(query_streams.cpp)
+target_link_libraries(${test_prefix}query_streams mg-query kafka-mock)
+

 # Test query/procedure
 add_unit_test(query_procedure_mgp_type.cpp)
@@ -327,16 +343,3 @@ add_custom_command(
 add_custom_target(test_lcp ALL DEPENDS ${CMAKE_CURRENT_BINARY_DIR}/test_lcp)
 add_test(test_lcp ${CMAKE_CURRENT_BINARY_DIR}/test_lcp)
 add_dependencies(memgraph__unit test_lcp)
-
-
-# Test integrations-kafka
-
-add_library(kafka-mock STATIC kafka_mock.cpp)
-target_link_libraries(kafka-mock mg-utils librdkafka++ librdkafka Threads::Threads zlib gtest)
-# Include directories are intentionally not set, because kafka-mock isn't meant to be used apart from unit tests
-
-add_unit_test(integrations_kafka_consumer.cpp kafka_mock.cpp)
-target_link_libraries(${test_prefix}integrations_kafka_consumer kafka-mock mg-integrations-kafka)
-
-add_unit_test(mgp_kafka_c_api.cpp)
-target_link_libraries(${test_prefix}mgp_kafka_c_api mg-query mg-integrations-kafka)
--- a/tests/unit/integrations_kafka_consumer.cpp
+++ b/tests/unit/integrations_kafka_consumer.cpp
@@ -3,20 +3,22 @@
 #include <string>
 #include <string_view>
 #include <thread>
+#include <vector>

 #include <fmt/core.h>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
 #include <spdlog/common.h>
 #include <spdlog/spdlog.h>
-#include "gmock/gmock.h"
-#include "gtest/gtest.h"
 #include "integrations/kafka/consumer.hpp"
 #include "integrations/kafka/exceptions.hpp"
 #include "kafka_mock.hpp"
-#include "utils/timer.hpp"
+#include "utils/string.hpp"

 using namespace integrations::kafka;

 namespace {
+const auto kDummyConsumerFunction = [](const auto & /*messages*/) {};
 int SpanToInt(std::span<const char> span) {
  int result{0};
  if (span.size() != sizeof(int)) {
@@ -33,9 +35,7 @@ struct ConsumerTest : public ::testing::Test {
  ConsumerInfo CreateDefaultConsumerInfo() const {
    const auto test_name = std::string{::testing::UnitTest::GetInstance()->current_test_info()->name()};
    return ConsumerInfo{
-        .consumer_function = [](const std::vector<Message> &) {},
        .consumer_name = "Consumer" + test_name,
-        .bootstrap_servers = cluster.Bootstraps(),
        .topics = {kTopicName},
        .consumer_group = "ConsumerGroup " + test_name,
        .batch_interval = std::nullopt,
@@ -43,25 +43,32 @@ struct ConsumerTest : public ::testing::Test {
    };
  };

-  std::unique_ptr<Consumer> CreateConsumer(ConsumerInfo &&info) {
-    auto custom_consumer_function = std::move(info.consumer_function);
+  std::unique_ptr<Consumer> CreateConsumer(ConsumerInfo &&info, ConsumerFunction consumer_function) {
+    EXPECT_EQ(1, info.topics.size());
+    EXPECT_EQ(info.topics.at(0), kTopicName);
    auto last_received_message = std::make_shared<std::atomic<int>>(0);
-    info.consumer_function = [weak_last_received_message = std::weak_ptr{last_received_message},
-                              custom_consumer_function =
-                                  std::move(custom_consumer_function)](const std::vector<Message> &messages) {
+    const auto consumer_function_wrapper = [weak_last_received_message = std::weak_ptr{last_received_message},
+                                            consumer_function =
+                                                std::move(consumer_function)](const std::vector<Message> &messages) {
      auto last_received_message = weak_last_received_message.lock();
+
+      EXPECT_FALSE(messages.empty());
+      for (const auto &message : messages) {
+        EXPECT_EQ(message.TopicName(), kTopicName);
+      }
      if (last_received_message != nullptr) {
        *last_received_message = SpanToInt(messages.back().Payload());
      } else {
-        custom_consumer_function(messages);
+        consumer_function(messages);
      }
    };

-    auto consumer = std::make_unique<Consumer>(std::move(info));
+    auto consumer =
+        std::make_unique<Consumer>(cluster.Bootstraps(), std::move(info), std::move(consumer_function_wrapper));
    int sent_messages{1};
    SeedTopicWithInt(kTopicName, sent_messages);

-    consumer->Start(std::nullopt);
+    consumer->Start();
    if (!consumer->IsRunning()) {
      return nullptr;
    }
@@ -104,15 +111,15 @@ TEST_F(ConsumerTest, BatchInterval) {
  std::vector<std::pair<size_t, std::chrono::steady_clock::time_point>> received_timestamps{};
  info.batch_interval = kBatchInterval;
  auto expected_messages_received = true;
-  info.consumer_function = [&](const std::vector<Message> &messages) mutable {
+  auto consumer_function = [&](const std::vector<Message> &messages) mutable {
    received_timestamps.push_back({messages.size(), std::chrono::steady_clock::now()});
    for (const auto &message : messages) {
      expected_messages_received &= (kMessage == std::string_view(message.Payload().data(), message.Payload().size()));
    }
  };

-  auto consumer = CreateConsumer(std::move(info));
-  consumer->Start(std::nullopt);
+  auto consumer = CreateConsumer(std::move(info), std::move(consumer_function));
+  consumer->Start();
  ASSERT_TRUE(consumer->IsRunning());

  constexpr auto kMessageCount = 7;
@@ -149,13 +156,13 @@ TEST_F(ConsumerTest, BatchInterval) {
 }

 TEST_F(ConsumerTest, StartStop) {
-  Consumer consumer{CreateDefaultConsumerInfo()};
+  Consumer consumer{cluster.Bootstraps(), CreateDefaultConsumerInfo(), kDummyConsumerFunction};

  auto start = [&consumer](const bool use_conditional) {
    if (use_conditional) {
      consumer.StartIfStopped();
    } else {
-      consumer.Start(std::nullopt);
+      consumer.Start();
    }
  };

@@ -178,7 +185,7 @@ TEST_F(ConsumerTest, StartStop) {

    start(use_conditional_start);
    EXPECT_TRUE(consumer.IsRunning());
-    EXPECT_THROW(consumer.Start(std::nullopt), ConsumerRunningException);
+    EXPECT_THROW(consumer.Start(), ConsumerRunningException);
    consumer.StartIfStopped();
    EXPECT_TRUE(consumer.IsRunning());

@@ -207,15 +214,15 @@ TEST_F(ConsumerTest, BatchSize) {
  info.batch_size = kBatchSize;
  constexpr std::string_view kMessage = "BatchSizeTestMessage";
  auto expected_messages_received = true;
-  info.consumer_function = [&](const std::vector<Message> &messages) mutable {
+  auto consumer_function = [&](const std::vector<Message> &messages) mutable {
    received_timestamps.push_back({messages.size(), std::chrono::steady_clock::now()});
    for (const auto &message : messages) {
      expected_messages_received &= (kMessage == std::string_view(message.Payload().data(), message.Payload().size()));
    }
  };

-  auto consumer = CreateConsumer(std::move(info));
-  consumer->Start(std::nullopt);
+  auto consumer = CreateConsumer(std::move(info), std::move(consumer_function));
+  consumer->Start();
  ASSERT_TRUE(consumer->IsRunning());

  constexpr auto kLastBatchMessageCount = 1;
@@ -259,14 +266,15 @@ TEST_F(ConsumerTest, BatchSize) {

 TEST_F(ConsumerTest, InvalidBootstrapServers) {
  auto info = CreateDefaultConsumerInfo();
-  info.bootstrap_servers = "non.existing.host:9092";
-  EXPECT_THROW(Consumer(std::move(info)), ConsumerFailedToInitializeException);
+
+  EXPECT_THROW(Consumer("non.existing.host:9092", std::move(info), kDummyConsumerFunction),
+               ConsumerFailedToInitializeException);
 }

 TEST_F(ConsumerTest, InvalidTopic) {
  auto info = CreateDefaultConsumerInfo();
  info.topics = {"Non existing topic"};
-  EXPECT_THROW(Consumer(std::move(info)), TopicNotFoundException);
+  EXPECT_THROW(Consumer(cluster.Bootstraps(), std::move(info), kDummyConsumerFunction), TopicNotFoundException);
 }

 TEST_F(ConsumerTest, StartsFromPreviousOffset) {
@@ -276,8 +284,9 @@ TEST_F(ConsumerTest, StartsFromPreviousOffset) {
  std::atomic<int> received_message_count{0};
  const std::string kMessagePrefix{"Message"};
  auto expected_messages_received = true;
-  info.consumer_function = [&](const std::vector<Message> &messages) mutable {
+  auto consumer_function = [&](const std::vector<Message> &messages) mutable {
    auto message_count = received_message_count.load();
+    EXPECT_EQ(messages.size(), 1);
    for (const auto &message : messages) {
      std::string message_payload = kMessagePrefix + std::to_string(message_count++);
      expected_messages_received &=
@@ -287,33 +296,34 @@ TEST_F(ConsumerTest, StartsFromPreviousOffset) {
  };

  // This test depends on CreateConsumer starts and stops the consumer, so the offset is stored
-  auto consumer = CreateConsumer(std::move(info));
+  auto consumer = CreateConsumer(std::move(info), std::move(consumer_function));
  ASSERT_FALSE(consumer->IsRunning());

-  constexpr auto kMessageCount = 4;
-  for (auto sent_messages = 0; sent_messages < kMessageCount; ++sent_messages) {
-    cluster.SeedTopic(kTopicName, std::string_view{kMessagePrefix + std::to_string(sent_messages)});
-  }
-
-  auto do_batches = [&](int64_t batch_count) {
+  auto send_and_consume_messages = [&](int batch_count) {
    SCOPED_TRACE(fmt::format("Already received messages: {}", received_message_count.load()));
-    consumer->Start(batch_count);
+    auto expected_total_messages = received_message_count + batch_count;
+    for (auto sent_messages = 0; sent_messages < batch_count; ++sent_messages) {
+      cluster.SeedTopic(kTopicName,
+                        std::string_view{kMessagePrefix + std::to_string(received_message_count + sent_messages)});
+    }
+    consumer->Start();
    const auto start = std::chrono::steady_clock::now();
    ASSERT_TRUE(consumer->IsRunning());
    constexpr auto kMaxWaitTime = std::chrono::seconds(5);

-    while (consumer->IsRunning() && (std::chrono::steady_clock::now() - start) < kMaxWaitTime) {
+    while (expected_total_messages != received_message_count.load() &&
+           (std::chrono::steady_clock::now() - start) < kMaxWaitTime) {
      std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
    // it is stopped because of limited batches
+    EXPECT_EQ(expected_total_messages, received_message_count);
+    consumer->Stop();
    ASSERT_FALSE(consumer->IsRunning());
+    EXPECT_TRUE(expected_messages_received) << "Some unexpected message have been received";
  };

-  ASSERT_NO_FATAL_FAILURE(do_batches(kMessageCount / 2));
-  ASSERT_NO_FATAL_FAILURE(do_batches(kMessageCount / 2));
-
-  EXPECT_TRUE(expected_messages_received) << "Some unexpected message have been received";
-  EXPECT_EQ(received_message_count, kMessageCount);
+  ASSERT_NO_FATAL_FAILURE(send_and_consume_messages(2));
+  ASSERT_NO_FATAL_FAILURE(send_and_consume_messages(2));
 }

 TEST_F(ConsumerTest, TestMethodWorks) {
@@ -321,10 +331,10 @@ TEST_F(ConsumerTest, TestMethodWorks) {
  auto info = CreateDefaultConsumerInfo();
  info.batch_size = kBatchSize;
  const std::string kMessagePrefix{"Message"};
-  info.consumer_function = [](const std::vector<Message> &messages) mutable {};
+  auto consumer_function = [](const std::vector<Message> &messages) mutable {};

  // This test depends on CreateConsumer starts and stops the consumer, so the offset is stored
-  auto consumer = CreateConsumer(std::move(info));
+  auto consumer = CreateConsumer(std::move(info), std::move(consumer_function));

  constexpr auto kMessageCount = 4;
  for (auto sent_messages = 0; sent_messages < kMessageCount; ++sent_messages) {
@@ -362,3 +372,38 @@ TEST_F(ConsumerTest, TestMethodWorks) {
    EXPECT_NO_FATAL_FAILURE(check_test_method());
  }
 }
+
+TEST_F(ConsumerTest, ConsumerStatus) {
+  const std::string kConsumerName = "ConsumerGroupNameAAAA";
+  const std::vector<std::string> topics = {"Topic1QWER", "Topic2XCVBB"};
+  const std::string kConsumerGroupName = "ConsumerGroupTestAsdf";
+  constexpr auto kBatchInterval = std::chrono::milliseconds{111};
+  constexpr auto kBatchSize = 222;
+
+  for (const auto &topic : topics) {
+    cluster.CreateTopic(topic);
+  }
+
+  auto check_info = [&](const ConsumerInfo &info) {
+    EXPECT_EQ(kConsumerName, info.consumer_name);
+    EXPECT_EQ(kConsumerGroupName, info.consumer_group);
+    EXPECT_EQ(kBatchInterval, info.batch_interval);
+    EXPECT_EQ(kBatchSize, info.batch_size);
+    EXPECT_EQ(2, info.topics.size()) << utils::Join(info.topics, ",");
+    ASSERT_LE(2, info.topics.size());
+    EXPECT_EQ(topics[0], info.topics[0]);
+    EXPECT_EQ(topics[1], info.topics[1]);
+  };
+
+  Consumer consumer{cluster.Bootstraps(),
+                    ConsumerInfo{kConsumerName, topics, kConsumerGroupName, kBatchInterval, kBatchSize},
+                    kDummyConsumerFunction};
+
+  check_info(consumer.Info());
+  consumer.Start();
+  check_info(consumer.Info());
+  consumer.StartIfStopped();
+  check_info(consumer.Info());
+  consumer.StopIfRunning();
+  check_info(consumer.Info());
+}
--- a/tests/unit/interpreter.cpp
+++ b/tests/unit/interpreter.cpp
@@ -30,7 +30,8 @@ auto ToEdgeList(const communication::bolt::Value &v) {
 struct InterpreterFaker {
  explicit InterpreterFaker(storage::Storage *db, const query::InterpreterConfig config,
                            const std::filesystem::path &data_directory)
-      : interpreter_context(db, config, data_directory), interpreter(&interpreter_context) {}
+      : interpreter_context(db, config, data_directory, "not used bootstrap servers"),
+        interpreter(&interpreter_context) {}

  auto Prepare(const std::string &query, const std::map<std::string, storage::PropertyValue> &params = {}) {
    ResultStreamFaker stream(interpreter_context.db);
--- a/tests/unit/kafka_mock.cpp
+++ b/tests/unit/kafka_mock.cpp
@@ -51,18 +51,22 @@ KafkaClusterMock::KafkaClusterMock(const std::vector<std::string> &topics) {
  }

  for (const auto &topic : topics) {
-    constexpr auto partition_count = 1;
-    constexpr auto replication_factor = 1;
-    rd_kafka_resp_err_t topic_err =
-        rd_kafka_mock_topic_create(cluster_.get(), topic.c_str(), partition_count, replication_factor);
-    if (RD_KAFKA_RESP_ERR_NO_ERROR != topic_err) {
-      throw std::runtime_error("Failed to create the mock topic (" + topic + "): " + rd_kafka_err2str(topic_err));
-    }
+    CreateTopic(topic);
  }
 };

 std::string KafkaClusterMock::Bootstraps() const { return rd_kafka_mock_cluster_bootstraps(cluster_.get()); };

+void KafkaClusterMock::CreateTopic(const std::string &topic_name) {
+  constexpr auto partition_count = 1;
+  constexpr auto replication_factor = 1;
+  rd_kafka_resp_err_t topic_err =
+      rd_kafka_mock_topic_create(cluster_.get(), topic_name.c_str(), partition_count, replication_factor);
+  if (RD_KAFKA_RESP_ERR_NO_ERROR != topic_err) {
+    throw std::runtime_error("Failed to create the mock topic (" + topic_name + "): " + rd_kafka_err2str(topic_err));
+  }
+}
+
 void KafkaClusterMock::SeedTopic(const std::string &topic_name, std::string_view message) {
  SeedTopic(topic_name, std::span{message.data(), message.size()});
 }
--- a/tests/unit/kafka_mock.hpp
+++ b/tests/unit/kafka_mock.hpp
@@ -28,6 +28,7 @@ class KafkaClusterMock {
  explicit KafkaClusterMock(const std::vector<std::string> &topics);

  std::string Bootstraps() const;
+  void CreateTopic(const std::string &topic_name);
  void SeedTopic(const std::string &topic_name, std::span<const char> message);
  void SeedTopic(const std::string &topic_name, std::string_view message);

--- a/tests/unit/query_dump.cpp
+++ b/tests/unit/query_dump.cpp
@@ -190,7 +190,7 @@ DatabaseState GetState(storage::Storage *db) {

 auto Execute(storage::Storage *db, const std::string &query) {
  auto data_directory = std::filesystem::temp_directory_path() / "MG_tests_unit_query_dump";
-  query::InterpreterContext context(db, query::InterpreterConfig{}, data_directory);
+  query::InterpreterContext context(db, query::InterpreterConfig{}, data_directory, "non existing bootstrap servers");
  query::Interpreter interpreter(&context);
  ResultStreamFaker stream(db);

@@ -704,7 +704,9 @@ TEST(DumpTest, ExecuteDumpDatabase) {
 class StatefulInterpreter {
 public:
  explicit StatefulInterpreter(storage::Storage *db)
-      : db_(db), context_(db_, query::InterpreterConfig{}, data_directory_), interpreter_(&context_) {}
+      : db_(db),
+        context_(db_, query::InterpreterConfig{}, data_directory_, "non existing bootstrap servers"),
+        interpreter_(&context_) {}

  auto Execute(const std::string &query) {
    ResultStreamFaker stream(db_);
--- a/tests/unit/query_plan_edge_cases.cpp
+++ b/tests/unit/query_plan_edge_cases.cpp
@@ -24,7 +24,7 @@ class QueryExecution : public testing::Test {

  void SetUp() {
    db_.emplace();
-    interpreter_context_.emplace(&*db_, query::InterpreterConfig{}, data_directory);
+    interpreter_context_.emplace(&*db_, query::InterpreterConfig{}, data_directory, "non existing bootstrap servers");
    interpreter_.emplace(&*interpreter_context_);
  }

--- a/tests/unit/query_streams.cpp
+++ b/tests/unit/query_streams.cpp
@@ -0,0 +1,231 @@
+#include <algorithm>
+#include <filesystem>
+#include <optional>
+#include <utility>
+
+#include <gtest/gtest.h>
+#include "kafka_mock.hpp"
+#include "query/config.hpp"
+#include "query/interpreter.hpp"
+#include "query/streams.hpp"
+#include "storage/v2/storage.hpp"
+
+using Streams = query::Streams;
+using StreamInfo = query::StreamInfo;
+using StreamStatus = query::StreamStatus;
+namespace {
+const static std::string kTopicName{"TrialTopic"};
+
+struct StreamCheckData {
+  std::string name;
+  StreamInfo info;
+  bool is_running;
+};
+
+std::string GetDefaultStreamName() {
+  return std::string{::testing::UnitTest::GetInstance()->current_test_info()->name()};
+}
+
+StreamInfo CreateDefaultStreamInfo() {
+  return StreamInfo{
+      .topics = {kTopicName},
+      .consumer_group = "ConsumerGroup " + GetDefaultStreamName(),
+      .batch_interval = std::nullopt,
+      .batch_size = std::nullopt,
+      // TODO(antaljanosbenjamin) Add proper reference once Streams supports that
+      .transformation_name = "not yet used",
+  };
+}
+
+StreamCheckData CreateDefaultStreamCheckData() { return {GetDefaultStreamName(), CreateDefaultStreamInfo(), false}; }
+
+std::filesystem::path GetCleanDataDirectory() {
+  const auto path = std::filesystem::temp_directory_path() / "query-streams";
+  std::filesystem::remove_all(path);
+  return path;
+}
+}  // namespace
+
+class StreamsTest : public ::testing::Test {
+ public:
+  StreamsTest() { ResetStreamsObject(); }
+
+ protected:
+  storage::Storage db_;
+  std::filesystem::path data_directory_{GetCleanDataDirectory()};
+  KafkaClusterMock mock_cluster_{std::vector<std::string>{kTopicName}};
+  // Though there is a Streams object in interpreter context, it makes more sense to use a separate object to test,
+  // because that provides a way to recreate the streams object and also give better control over the arguments of the
+  // Streams constructor.
+  query::InterpreterContext interpreter_context_{&db_, query::InterpreterConfig{}, data_directory_,
+                                                 "dont care bootstrap servers"};
+  std::filesystem::path streams_data_directory_{data_directory_ / "separate-dir-for-test"};
+  std::optional<Streams> streams_;
+
+  void ResetStreamsObject() {
+    streams_.emplace(&interpreter_context_, mock_cluster_.Bootstraps(), streams_data_directory_);
+  }
+
+  void CheckStreamStatus(const StreamCheckData &check_data) {
+    SCOPED_TRACE(fmt::format("Checking status of '{}'", check_data.name));
+    const auto &stream_statuses = streams_->Show();
+    auto it = std::find_if(stream_statuses.begin(), stream_statuses.end(),
+                           [&check_data](const auto &stream_status) { return stream_status.name == check_data.name; });
+    ASSERT_NE(it, stream_statuses.end());
+    const auto &status = *it;
+    // the order don't have to be strictly the same, but based on the implementation it shouldn't change
+    EXPECT_TRUE(std::equal(check_data.info.topics.begin(), check_data.info.topics.end(), status.info.topics.begin(),
+                           status.info.topics.end()));
+    EXPECT_EQ(check_data.info.consumer_group, status.info.consumer_group);
+    EXPECT_EQ(check_data.info.batch_interval, status.info.batch_interval);
+    EXPECT_EQ(check_data.info.batch_size, status.info.batch_size);
+    // TODO(antaljanosbenjamin) Add proper reference once Streams supports that
+    // EXPECT_EQ(check_data.info.transformation_name, status.info.transformation_name);
+    EXPECT_EQ(check_data.is_running, status.is_running);
+  }
+
+  void StartStream(StreamCheckData &check_data) {
+    streams_->Start(check_data.name);
+    check_data.is_running = true;
+  }
+
+  void StopStream(StreamCheckData &check_data) {
+    streams_->Stop(check_data.name);
+    check_data.is_running = false;
+  }
+
+  void Clear() {
+    if (!std::filesystem::exists(data_directory_)) return;
+    std::filesystem::remove_all(data_directory_);
+  }
+};
+
+TEST_F(StreamsTest, SimpleStreamManagement) {
+  auto check_data = CreateDefaultStreamCheckData();
+  streams_->Create(check_data.name, check_data.info);
+  EXPECT_NO_FATAL_FAILURE(CheckStreamStatus(check_data));
+
+  streams_->Start(check_data.name);
+  check_data.is_running = true;
+  EXPECT_NO_FATAL_FAILURE(CheckStreamStatus(check_data));
+
+  streams_->StopAll();
+  check_data.is_running = false;
+  EXPECT_NO_FATAL_FAILURE(CheckStreamStatus(check_data));
+
+  streams_->StartAll();
+  check_data.is_running = true;
+  EXPECT_NO_FATAL_FAILURE(CheckStreamStatus(check_data));
+
+  streams_->Stop(check_data.name);
+  check_data.is_running = false;
+  EXPECT_NO_FATAL_FAILURE(CheckStreamStatus(check_data));
+
+  streams_->Drop(check_data.name);
+  EXPECT_TRUE(streams_->Show().empty());
+}
+
+TEST_F(StreamsTest, CreateAlreadyExisting) {
+  auto stream_info = CreateDefaultStreamInfo();
+  auto stream_name = GetDefaultStreamName();
+  streams_->Create(stream_name, stream_info);
+
+  try {
+    streams_->Create(stream_name, stream_info);
+    FAIL() << "Creating already existing stream should throw\n";
+  } catch (query::StreamsException &exception) {
+    EXPECT_EQ(exception.what(), fmt::format("Stream already exists with name '{}'", stream_name));
+  }
+}
+
+TEST_F(StreamsTest, DropNotExistingStream) {
+  const auto stream_info = CreateDefaultStreamInfo();
+  const auto stream_name = GetDefaultStreamName();
+  const std::string not_existing_stream_name{"ThisDoesn'tExists"};
+  streams_->Create(stream_name, stream_info);
+
+  try {
+    streams_->Drop(not_existing_stream_name);
+    FAIL() << "Dropping not existing stream should throw\n";
+  } catch (query::StreamsException &exception) {
+    EXPECT_EQ(exception.what(), fmt::format("Couldn't find stream '{}'", not_existing_stream_name));
+  }
+}
+
+TEST_F(StreamsTest, RestoreStreams) {
+  std::array stream_check_datas{
+      CreateDefaultStreamCheckData(),
+      CreateDefaultStreamCheckData(),
+      CreateDefaultStreamCheckData(),
+      CreateDefaultStreamCheckData(),
+  };
+
+  // make the stream infos unique
+  for (auto i = 0; i < stream_check_datas.size(); ++i) {
+    auto &stream_check_data = stream_check_datas[i];
+    auto &stream_info = stream_check_data.info;
+    auto iteration_postfix = std::to_string(i);
+
+    stream_check_data.name += iteration_postfix;
+    stream_info.topics[0] += iteration_postfix;
+    stream_info.consumer_group += iteration_postfix;
+    stream_info.transformation_name += iteration_postfix;
+    if (i > 0) {
+      stream_info.batch_interval = std::chrono::milliseconds((i + 1) * 10);
+      stream_info.batch_size = 1000 + i;
+    }
+
+    mock_cluster_.CreateTopic(stream_info.topics[0]);
+  }
+  stream_check_datas[1].info.batch_interval = {};
+  stream_check_datas[2].info.batch_size = {};
+
+  const auto check_restore_logic = [&stream_check_datas, this]() {
+    // Reset the Streams object to trigger reloading
+    ResetStreamsObject();
+    EXPECT_TRUE(streams_->Show().empty());
+    streams_->RestoreStreams();
+    EXPECT_EQ(stream_check_datas.size(), streams_->Show().size());
+    for (const auto &check_data : stream_check_datas) {
+      ASSERT_NO_FATAL_FAILURE(CheckStreamStatus(check_data));
+    }
+  };
+
+  streams_->RestoreStreams();
+  EXPECT_TRUE(streams_->Show().empty());
+
+  for (auto &check_data : stream_check_datas) {
+    streams_->Create(check_data.name, check_data.info);
+  }
+  {
+    SCOPED_TRACE("After streams are created");
+    check_restore_logic();
+  }
+
+  for (auto &check_data : stream_check_datas) {
+    StartStream(check_data);
+  }
+  {
+    SCOPED_TRACE("After starting streams");
+    check_restore_logic();
+  }
+
+  // Stop two of the streams
+  StopStream(stream_check_datas[1]);
+  StopStream(stream_check_datas[3]);
+  {
+    SCOPED_TRACE("After stopping two streams");
+    check_restore_logic();
+  }
+
+  // Stop the rest of the streams
+  StopStream(stream_check_datas[0]);
+  StopStream(stream_check_datas[2]);
+  check_restore_logic();
+  {
+    SCOPED_TRACE("After stopping all streams");
+    check_restore_logic();
+  }
+}
+
+// TODO(antaljanosbenjamin) Add tests for Streams::Test method and transformation
--- a/tests/unit/utils_synchronized.cpp
+++ b/tests/unit/utils_synchronized.cpp
@@ -2,8 +2,13 @@

 #include "gtest/gtest.h"

+#include "utils/rw_lock.hpp"
 #include "utils/synchronized.hpp"

+static_assert(utils::SharedMutex<std::shared_mutex>, "std::shared_mutex must be considered as shared mutex!");
+static_assert(utils::SharedMutex<utils::RWLock>, "utils::RWLock must be considered as shared mutex!");
+static_assert(!utils::SharedMutex<std::mutex>, "std::mutex must not be considered as shared mutex!");
+
 class NoMoveNoCopy {
 public:
  NoMoveNoCopy(int, int) {}
@@ -15,7 +20,6 @@ class NoMoveNoCopy {
  ~NoMoveNoCopy() = default;
 };

-// NOLINTNEXTLINE(hicpp-special-member-functions)
 TEST(Synchronized, Constructors) {
  {
    utils::Synchronized<std::vector<int>> vec;
@@ -31,7 +35,6 @@ TEST(Synchronized, Constructors) {
    std::vector<int> data = {1, 2, 3};
    utils::Synchronized<std::vector<int>> vec(std::move(data));
    // data is guaranteed by the standard to be empty after move
-    // NOLINTNEXTLINE(bugprone-use-after-move, hicpp-invalid-access-moved)
    EXPECT_TRUE(data.empty());
    EXPECT_EQ(vec->size(), 3);
  }
@@ -52,11 +55,33 @@ class TestLock {
  }
 };

-// NOLINTNEXTLINE(hicpp-special-member-functions)
-TEST(Synchronized, Usage) {
-  utils::Synchronized<std::vector<int>, TestLock> my_vector;
+bool test_shared_lock_locked = false;
+
+class TestSharedLock {
+ public:
+  void lock() {
+    ASSERT_FALSE(test_lock_locked);
+    test_lock_locked = true;
+  }
+  void unlock() {
+    ASSERT_TRUE(test_lock_locked);
+    test_lock_locked = false;
+  }
+  void lock_shared() {
+    ASSERT_FALSE(test_shared_lock_locked);
+    test_shared_lock_locked = true;
+  }
+  void unlock_shared() {
+    ASSERT_TRUE(test_shared_lock_locked);
+    test_shared_lock_locked = false;
+  }
+};
+
+template <typename TSynchronizedVector>
+void CheckWriteLock(TSynchronizedVector &my_vector) {
+  ASSERT_TRUE(my_vector->empty()) << "Cannot use not empty vector";
  {
-    // LockedPtr
+    SCOPED_TRACE("LockedPtr");
    auto ptr = my_vector.Lock();
    ASSERT_TRUE(test_lock_locked);
    ptr->push_back(5);
@@ -64,12 +89,12 @@ TEST(Synchronized, Usage) {
  ASSERT_FALSE(test_lock_locked);

  {
-    // Indirection operator
+    SCOPED_TRACE("Indirection operator");
    my_vector->push_back(6);
  }

  {
-    // Lambda
+    SCOPED_TRACE("Lambda");
    my_vector.WithLock([](auto &my_vector) {
      ASSERT_TRUE(test_lock_locked);
      EXPECT_EQ(my_vector.size(), 2);
@@ -79,3 +104,88 @@ TEST(Synchronized, Usage) {
    ASSERT_FALSE(test_lock_locked);
  }
 }
+
+template <typename TSynchronizedVector>
+void CheckReadLock(TSynchronizedVector &my_vector, const size_t expected_size) {
+  {
+    SCOPED_TRACE("ReadLockPtr");
+    auto ptr = my_vector.ReadLock();
+    ASSERT_TRUE(test_shared_lock_locked);
+    EXPECT_EQ(expected_size, ptr->size());
+  }
+  ASSERT_FALSE(test_shared_lock_locked);
+
+  {
+    auto ptr1 = my_vector.ReadLock();
+    ASSERT_TRUE(test_shared_lock_locked);
+
+    {
+      test_shared_lock_locked = false;
+      auto ptr2 = my_vector.ReadLock();
+      EXPECT_EQ(expected_size, ptr1->size());
+      EXPECT_EQ(expected_size, ptr2->size());
+    }
+    test_shared_lock_locked = true;
+  }
+  ASSERT_FALSE(test_shared_lock_locked);
+
+  {
+    SCOPED_TRACE("Indirection operator");
+    EXPECT_EQ(expected_size, my_vector->size());
+  }
+  {
+    SCOPED_TRACE("Indirection operator with other ReadLock");
+
+    [[maybe_unused]] auto ptr = my_vector.ReadLock();
+    ASSERT_TRUE(test_shared_lock_locked);
+    test_shared_lock_locked = false;
+    EXPECT_EQ(expected_size, my_vector->size());
+    ASSERT_FALSE(test_shared_lock_locked);
+    test_shared_lock_locked = true;
+  }
+  ASSERT_FALSE(test_shared_lock_locked);
+
+  {
+    SCOPED_TRACE("Lambda");
+    my_vector.WithReadLock([&expected_size](auto &my_vector) {
+      ASSERT_TRUE(test_shared_lock_locked);
+      EXPECT_EQ(my_vector.size(), expected_size);
+    });
+    ASSERT_FALSE(test_shared_lock_locked);
+  }
+  {
+    SCOPED_TRACE("Lambda with other ReadLock");
+    auto ptr1 = my_vector.ReadLock();
+    ASSERT_TRUE(test_shared_lock_locked);
+
+    {
+      test_shared_lock_locked = false;
+      my_vector.WithReadLock([&expected_size](const auto &my_vector) {
+        ASSERT_TRUE(test_shared_lock_locked);
+        EXPECT_EQ(my_vector.size(), expected_size);
+      });
+      ASSERT_FALSE(test_shared_lock_locked);
+    }
+    test_shared_lock_locked = true;
+  }
+  ASSERT_FALSE(test_shared_lock_locked);
+}
+
+TEST(Synchronized, Usage) {
+  utils::Synchronized<std::vector<int>, TestLock> my_vector;
+  CheckWriteLock(my_vector);
+}
+
+TEST(Synchronized, SharedUsage) {
+  utils::Synchronized<std::vector<int>, TestSharedLock> my_vector;
+  CheckWriteLock(my_vector);
+  {
+    SCOPED_TRACE("Non const reference");
+    ASSERT_NO_FATAL_FAILURE(CheckReadLock(my_vector, 2));
+  }
+  {
+    const utils::Synchronized<std::vector<int>, TestSharedLock> &my_const_vector = my_vector;
+    SCOPED_TRACE("Const reference");
+    ASSERT_NO_FATAL_FAILURE(CheckReadLock(my_const_vector, 2));
+  }
+}