From fa7e214bcf7e1ae84b75e7ced145663deadd2545 Mon Sep 17 00:00:00 2001
From: Matija Santl <matija.santl@memgraph.com>
Date: Tue, 19 Jun 2018 14:37:02 +0200
Subject: [PATCH] Add kafka library and integrate it into memgraph

Summary:
Integrated kafka library into memgraph. This version supports all opencypher
features and will only output messages consumed from kafka.

Depends on D1434

Next steps are persisting stream metadata and transforming messages in order to
store them in the graph.

Reviewers: teon.banek, mtomic, mferencevic, buda

Reviewed By: teon.banek

Subscribers: mferencevic, pullbot, buda

Differential Revision: https://phabricator.memgraph.io/D1466
---
 .ycm_extra_conf.py                            |   1 +
 CMakeLists.txt                                |   1 +
 docs/feature_specs/kafka/opencypher.md        |  16 +-
 libs/CMakeLists.txt                           |  16 ++
 libs/setup.sh                                 |   5 +
 src/CMakeLists.txt                            |   7 +-
 src/database/graph_db.cpp                     |   7 +
 src/database/graph_db.hpp                     |   3 +
 src/integrations/CMakeLists.txt               |   7 +
 src/integrations/kafka/consumer.cpp           | 202 ++++++++++++++++++
 src/integrations/kafka/consumer.hpp           |  66 ++++++
 src/integrations/kafka/exceptions.hpp         |  59 +++++
 src/integrations/kafka/streams.cpp            |  68 ++++++
 src/integrations/kafka/streams.hpp            |  38 ++++
 src/query/frontend/ast/ast.capnp              |   6 +-
 src/query/frontend/ast/ast.cpp                |  34 ++-
 src/query/frontend/ast/ast.hpp                |  33 +--
 .../frontend/ast/cypher_main_visitor.cpp      |  33 ++-
 .../frontend/ast/cypher_main_visitor.hpp      |   3 +
 .../frontend/opencypher/grammar/Cypher.g4     |  18 +-
 .../frontend/stripped_lexer_constants.hpp     |   2 +-
 src/query/plan/operator.cpp                   | 150 +++++++++++--
 src/query/plan/operator.lcp                   |  35 ++-
 src/query/plan/rule_based_planner.hpp         |  12 +-
 tests/unit/CMakeLists.txt                     |   4 +-
 tests/unit/cypher_main_visitor.cpp            |  94 ++++++--
 tests/unit/query_common.hpp                   |   8 +-
 tests/unit/query_planner.cpp                  | 105 ++++++---
 28 files changed, 923 insertions(+), 110 deletions(-)
 create mode 100644 src/integrations/CMakeLists.txt
 create mode 100644 src/integrations/kafka/consumer.cpp
 create mode 100644 src/integrations/kafka/consumer.hpp
 create mode 100644 src/integrations/kafka/exceptions.hpp
 create mode 100644 src/integrations/kafka/streams.cpp
 create mode 100644 src/integrations/kafka/streams.hpp

diff --git a/.ycm_extra_conf.py b/.ycm_extra_conf.py
index 99b8f46ff..0e172c1b6 100644
--- a/.ycm_extra_conf.py
+++ b/.ycm_extra_conf.py
@@ -35,6 +35,7 @@ BASE_FLAGS = [
     '-I./libs/bzip2',
     '-I./libs/zlib',
     '-I./libs/rocksdb/include',
+    '-I./libs/librdkafka/include/librdkafka',
     '-I./build/include'
 ]
 
diff --git a/CMakeLists.txt b/CMakeLists.txt
index b8db6c080..010ab59ea 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -154,6 +154,7 @@ include_directories(SYSTEM ${BZIP2_INCLUDE_DIR})
 include_directories(SYSTEM ${ZLIB_INCLUDE_DIR})
 include_directories(SYSTEM ${ROCKSDB_INCLUDE_DIR})
 include_directories(SYSTEM ${CAPNP_INCLUDE_DIR})
+include_directories(SYSTEM ${LIBRDKAFKA_INCLUDE_DIR})
 # -----------------------------------------------------------------------------
 
 # openCypher parser -----------------------------------------------------------
diff --git a/docs/feature_specs/kafka/opencypher.md b/docs/feature_specs/kafka/opencypher.md
index 22f9dd2a1..7e485f17c 100644
--- a/docs/feature_specs/kafka/opencypher.md
+++ b/docs/feature_specs/kafka/opencypher.md
@@ -20,15 +20,29 @@ The full openCypher clause for creating a stream is:
 ```opencypher
 CREATE STREAM stream_name AS
   LOAD DATA KAFKA 'URI'
+  WITH TOPIC 'topic'
   WITH TRANSFORM 'URI'
-  [BATCH INTERVAL milliseconds]
+  [BATCH_INTERVAL milliseconds]
+  [BATCH_SIZE count]
 ```
 
+The `WITH TOPIC` parameter specifies the kafka topic from which we'll stream
+data.
+
 The `WITH TRANSFORM` parameter should contain a URI of the transform script.
 
 The `BATCH_INTERVAL` parameter defines the time interval in milliseconds
 that defines the time between two successive stream importing operations.
 
+The `BATCH_SIZE` parameter defines the count of kafka messages that will be
+batched together before import.
+
+If both `BATCH_INTERVAL` and `BATCH_SIZE` parameters are given, the condition
+that is satisfied first will trigger the batched import.
+
+Default values for `BATCH_INTERVAL` is 100 milliseconds, and the default value
+for `BATCH_SIZE` is 10;
+
 The `DROP` clause deletes a stream:
 ```opencypher
 DROP STREAM stream_name;
diff --git a/libs/CMakeLists.txt b/libs/CMakeLists.txt
index e17c04c91..167fccf92 100644
--- a/libs/CMakeLists.txt
+++ b/libs/CMakeLists.txt
@@ -233,3 +233,19 @@ set(CAPNP_EXE ${CMAKE_CURRENT_SOURCE_DIR}/capnproto/local/bin/capnp
 set(CAPNP_CXX_EXE ${CMAKE_CURRENT_SOURCE_DIR}/capnproto/local/bin/capnpc-c++
     CACHE FILEPATH "Path to capnproto c++ plugin executable" FORCE)
 mark_as_advanced(CAPNP_INCLUDE_DIR CAPNP_LIBRARY KJ_LIBRARY CAPNP_EXE CAPNP_CXX_EXE)
+
+# Setup librdkafka.
+import_external_library(librdkafka STATIC
+  ${CMAKE_CURRENT_SOURCE_DIR}/librdkafka/lib/librdkafka.a
+  ${CMAKE_CURRENT_SOURCE_DIR}/librdkafka/include/librdkafka
+  CMAKE_ARGS -DRDKAFKA_BUILD_STATIC=ON
+             -DRDKAFKA_BUILD_EXAMPLES=OFF
+             -DRDKAFKA_BUILD_TESTS=OFF
+             -DCMAKE_INSTALL_LIBDIR=lib
+             -DWITH_SSL=ON
+             # If we want SASL, we need to install it on build machines
+             -DWITH_SASL=OFF)
+
+import_library(librdkafka++ STATIC
+  ${CMAKE_CURRENT_SOURCE_DIR}/librdkafka/lib/librdkafka++.a
+  librdkafka-proj)
diff --git a/libs/setup.sh b/libs/setup.sh
index 4c72e246a..14c9e5ae0 100755
--- a/libs/setup.sh
+++ b/libs/setup.sh
@@ -127,3 +127,8 @@ tar -xzf capnproto.tar.gz
 rm -rf capnproto
 mv capnproto-c++-0.6.1 capnproto
 rm capnproto.tar.gz
+
+# kafka
+kafka_tag="c319b4e987d0bc4fe4f01cf91419d90b62061655" # Mar 8, 2018
+# git clone https://github.com/edenhill/librdkafka.git
+clone git://deps.memgraph.io/librdkafka.git librdkafka $kafka_tag
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 54c51ab23..486725bcd 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -1,7 +1,8 @@
 # CMake configuration for the main memgraph library and executable
 
-# add memgraph sub libraries
+# add memgraph sub libraries, ordered by dependency
 add_subdirectory(utils)
+add_subdirectory(integrations)
 add_subdirectory(io)
 add_subdirectory(telemetry)
 
@@ -196,7 +197,7 @@ set(MEMGRAPH_ALL_LIBS stdc++fs Threads::Threads fmt cppitertools
     ${OPENSSL_LIBRARIES}
     ${Boost_IOSTREAMS_LIBRARY_RELEASE}
     ${Boost_SERIALIZATION_LIBRARY_RELEASE}
-    mg-utils mg-io)
+    mg-utils mg-io mg-integrations)
 
 if (USE_LTALLOC)
     list(APPEND MEMGRAPH_ALL_LIBS ltalloc)
@@ -218,7 +219,7 @@ add_dependencies(memgraph_lib generate_capnp)
 
 # STATIC library used to store key-value pairs
 add_library(kvstore_lib STATIC storage/kvstore.cpp)
-target_link_libraries(kvstore_lib stdc++fs mg-utils rocksdb bzip2 zlib)
+target_link_libraries(kvstore_lib stdc++fs mg-utils rocksdb bzip2 zlib glog gflags)
 
 # STATIC library for dummy key-value storage
 add_library(kvstore_dummy_lib STATIC storage/kvstore_dummy.cpp)
diff --git a/src/database/graph_db.cpp b/src/database/graph_db.cpp
index bb5dce37f..ddf229b18 100644
--- a/src/database/graph_db.cpp
+++ b/src/database/graph_db.cpp
@@ -56,6 +56,9 @@ class PrivateBase : public GraphDb {
 
   Storage &storage() override { return *storage_; }
   durability::WriteAheadLog &wal() override { return wal_; }
+  integrations::kafka::Streams &kafka_streams() override {
+    return kafka_streams_;
+  }
   int WorkerId() const override { return config_.worker_id; }
 
   // Makes a local snapshot from the visibility of accessor
@@ -98,6 +101,7 @@ class PrivateBase : public GraphDb {
   durability::WriteAheadLog wal_{config_.worker_id,
                                  config_.durability_directory,
                                  config_.durability_enabled};
+  integrations::kafka::Streams kafka_streams_;
 };
 
 template <template <typename TId> class TMapper>
@@ -445,6 +449,9 @@ PublicBase::~PublicBase() {
 GraphDb::Type PublicBase::type() const { return impl_->type(); }
 Storage &PublicBase::storage() { return impl_->storage(); }
 durability::WriteAheadLog &PublicBase::wal() { return impl_->wal(); }
+integrations::kafka::Streams &PublicBase::kafka_streams() {
+  return impl_->kafka_streams();
+}
 tx::Engine &PublicBase::tx_engine() { return impl_->tx_engine(); }
 ConcurrentIdMapper<Label> &PublicBase::label_mapper() {
   return impl_->label_mapper();
diff --git a/src/database/graph_db.hpp b/src/database/graph_db.hpp
index a5de5bab0..85a22519e 100644
--- a/src/database/graph_db.hpp
+++ b/src/database/graph_db.hpp
@@ -8,6 +8,7 @@
 #include "database/storage.hpp"
 #include "database/storage_gc.hpp"
 #include "durability/wal.hpp"
+#include "integrations/kafka/streams.hpp"
 #include "io/network/endpoint.hpp"
 #include "storage/concurrent_id_mapper.hpp"
 #include "storage/types.hpp"
@@ -92,6 +93,7 @@ class GraphDb {
   virtual Type type() const = 0;
   virtual Storage &storage() = 0;
   virtual durability::WriteAheadLog &wal() = 0;
+  virtual integrations::kafka::Streams &kafka_streams() = 0;
   virtual tx::Engine &tx_engine() = 0;
   virtual storage::ConcurrentIdMapper<storage::Label> &label_mapper() = 0;
   virtual storage::ConcurrentIdMapper<storage::EdgeType>
@@ -149,6 +151,7 @@ class PublicBase : public GraphDb {
   Type type() const override;
   Storage &storage() override;
   durability::WriteAheadLog &wal() override;
+  integrations::kafka::Streams &kafka_streams() override;
   tx::Engine &tx_engine() override;
   storage::ConcurrentIdMapper<storage::Label> &label_mapper() override;
   storage::ConcurrentIdMapper<storage::EdgeType> &edge_type_mapper() override;
diff --git a/src/integrations/CMakeLists.txt b/src/integrations/CMakeLists.txt
new file mode 100644
index 000000000..da83b0e42
--- /dev/null
+++ b/src/integrations/CMakeLists.txt
@@ -0,0 +1,7 @@
+set(integrations_src_files
+    kafka/streams.cpp
+    kafka/consumer.cpp)
+
+add_library(mg-integrations STATIC ${integrations_src_files})
+target_link_libraries(mg-integrations Threads::Threads fmt glog gflags
+                      librdkafka++ librdkafka zlib)
diff --git a/src/integrations/kafka/consumer.cpp b/src/integrations/kafka/consumer.cpp
new file mode 100644
index 000000000..36a44c2d8
--- /dev/null
+++ b/src/integrations/kafka/consumer.cpp
@@ -0,0 +1,202 @@
+#include "integrations/kafka/consumer.hpp"
+
+#include <chrono>
+
+#include "glog/logging.h"
+
+#include "integrations/kafka/exceptions.hpp"
+
+namespace integrations {
+namespace kafka {
+
+using namespace std::chrono_literals;
+
+constexpr int64_t kDefaultBatchIntervalMillis = 100;
+constexpr int64_t kDefaultBatchSize = 10;
+
+void Consumer::event_cb(RdKafka::Event &event) {
+  switch (event.type()) {
+    case RdKafka::Event::Type::EVENT_ERROR:
+      LOG(WARNING) << "[Kafka] stream " << info_.stream_name << " ERROR ("
+                   << RdKafka::err2str(event.err()) << "): " << event.str();
+      break;
+    default:
+      break;
+  }
+}
+
+Consumer::Consumer(const StreamInfo &info) : info_(info) {
+  std::unique_ptr<RdKafka::Conf> conf(
+      RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
+  std::string error;
+
+  if (conf->set("event_cb", this, error) != RdKafka::Conf::CONF_OK) {
+    throw ConsumerFailedToInitializeException(info_.stream_name, error);
+  }
+
+  if (conf->set("enable.partition.eof", "false", error) !=
+      RdKafka::Conf::CONF_OK) {
+    throw ConsumerFailedToInitializeException(info_.stream_name, error);
+  }
+
+  if (conf->set("bootstrap.servers", info_.stream_uri, error) !=
+      RdKafka::Conf::CONF_OK) {
+    throw ConsumerFailedToInitializeException(info_.stream_name, error);
+  }
+
+  if (conf->set("group.id", "mg", error) != RdKafka::Conf::CONF_OK) {
+    throw ConsumerFailedToInitializeException(info_.stream_name, error);
+  }
+
+  consumer_ = std::unique_ptr<RdKafka::KafkaConsumer,
+                              std::function<void(RdKafka::KafkaConsumer *)>>(
+      RdKafka::KafkaConsumer::create(conf.get(), error),
+      [this](auto *consumer) {
+        this->StopConsuming();
+        consumer->close();
+        delete consumer;
+      });
+
+  if (!consumer_) {
+    throw ConsumerFailedToInitializeException(info_.stream_name, error);
+  }
+
+  // Try fetching metadata first and check if topic exists.
+  RdKafka::ErrorCode err;
+  RdKafka::Metadata *raw_metadata = nullptr;
+  err = consumer_->metadata(true, nullptr, &raw_metadata, 1000);
+  std::unique_ptr<RdKafka::Metadata> metadata(raw_metadata);
+  if (err != RdKafka::ERR_NO_ERROR) {
+    throw ConsumerFailedToInitializeException(info_.stream_name,
+                                              RdKafka::err2str(err));
+  }
+
+  bool topic_found = false;
+  for (const auto &topic_metadata : *metadata->topics()) {
+    if (topic_metadata->topic() == info_.stream_topic) {
+      topic_found = true;
+      break;
+    }
+  }
+
+  if (!topic_found) {
+    throw TopicNotFoundException(info_.stream_name);
+  }
+
+  err = consumer_->subscribe({info_.stream_topic});
+  if (err != RdKafka::ERR_NO_ERROR) {
+    throw ConsumerFailedToInitializeException(info_.stream_name,
+                                              RdKafka::err2str(err));
+  }
+}
+
+void Consumer::StopConsuming() {
+  is_running_.store(false);
+  if (thread_.joinable()) thread_.join();
+}
+
+void Consumer::StartConsuming(
+    std::experimental::optional<int64_t> batch_limit) {
+  thread_ = std::thread([this, batch_limit]() {
+    int64_t batch_count = 0;
+    is_running_.store(true);
+
+    while (is_running_) {
+      int64_t remaining_timeout_in_ms =
+          info_.batch_interval_in_ms.value_or(kDefaultBatchIntervalMillis);
+      int64_t remaining_size = info_.batch_size.value_or(kDefaultBatchSize);
+
+      auto start = std::chrono::system_clock::now();
+
+      bool run_batch = true;
+      while (is_running_ && run_batch && remaining_size-- > 0) {
+        std::unique_ptr<RdKafka::Message> msg(
+            consumer_->consume(remaining_timeout_in_ms));
+        switch (msg->err()) {
+          case RdKafka::ERR__TIMED_OUT:
+            run_batch = false;
+            break;
+
+          case RdKafka::ERR_NO_ERROR:
+            // TODO (msantl): store message to current batch and pass the batch
+            // to transform
+            break;
+
+          default:
+            LOG(ERROR) << "Consumer error: " << msg->errstr();
+            is_running_.store(false);
+            break;
+        }
+
+        auto now = std::chrono::system_clock::now();
+        auto took =
+            std::chrono::duration_cast<std::chrono::milliseconds>(now - start);
+        if (took.count() >= remaining_timeout_in_ms) {
+          break;
+        }
+
+        remaining_timeout_in_ms = remaining_timeout_in_ms - took.count();
+        start = now;
+      }
+
+      if (batch_limit != std::experimental::nullopt) {
+        batch_count++;
+        if (batch_limit <= batch_count) {
+          is_running_.store(false);
+        }
+      }
+    }
+  });
+}
+
+void Consumer::Start(std::experimental::optional<int64_t> batch_limit) {
+  if (!consumer_) {
+    throw ConsumerNotAvailableException(info_.stream_name);
+  }
+
+  if (is_running_) {
+    throw ConsumerRunningException(info_.stream_name);
+  }
+
+  StartConsuming(batch_limit);
+}
+
+void Consumer::Stop() {
+  if (!consumer_) {
+    throw ConsumerNotAvailableException(info_.stream_name);
+  }
+
+  if (!is_running_) {
+    throw ConsumerStoppedException(info_.stream_name);
+  }
+
+  StopConsuming();
+}
+
+void Consumer::StartIfNotStopped() {
+  if (!consumer_) {
+    throw ConsumerNotAvailableException(info_.stream_name);
+  }
+
+  if (!is_running_) {
+    StartConsuming(std::experimental::nullopt);
+  }
+}
+
+void Consumer::StopIfNotRunning() {
+  if (!consumer_) {
+    throw ConsumerNotAvailableException(info_.stream_name);
+  }
+
+  if (is_running_) {
+    StopConsuming();
+  }
+}
+
+StreamInfo Consumer::info() {
+  info_.is_running = is_running_;
+  return info_;
+}
+
+}  // namespace kafka
+}  // namespace integrations
diff --git a/src/integrations/kafka/consumer.hpp b/src/integrations/kafka/consumer.hpp
new file mode 100644
index 000000000..806ce99f0
--- /dev/null
+++ b/src/integrations/kafka/consumer.hpp
@@ -0,0 +1,66 @@
+#pragma once
+
+#include <atomic>
+#include <experimental/optional>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "rdkafkacpp.h"
+
+namespace integrations {
+namespace kafka {
+
+struct StreamInfo {
+  std::string stream_name;
+  std::string stream_uri;
+  std::string stream_topic;
+  std::string transform_uri;
+  std::experimental::optional<int64_t> batch_interval_in_ms;
+  std::experimental::optional<int64_t> batch_size;
+
+  bool is_running = false;
+};
+
+class Consumer final : public RdKafka::EventCb {
+ public:
+  Consumer() = delete;
+
+  explicit Consumer(const StreamInfo &info);
+
+  Consumer(const Consumer &other) = delete;
+  Consumer(Consumer &&other) = delete;
+
+  Consumer &operator=(const Consumer &other) = delete;
+  Consumer &operator=(Consumer &&other) = delete;
+
+  void Start(std::experimental::optional<int64_t> batch_limit);
+
+  void Stop();
+
+  void StartIfNotStopped();
+
+  void StopIfNotRunning();
+
+  StreamInfo info();
+
+ private:
+  void event_cb(RdKafka::Event &event) override;
+
+  StreamInfo info_;
+
+  std::atomic<bool> is_running_{false};
+  std::thread thread_;
+
+  std::unique_ptr<RdKafka::KafkaConsumer,
+                  std::function<void(RdKafka::KafkaConsumer *)>>
+      consumer_;
+
+  void StopConsuming();
+
+  void StartConsuming(std::experimental::optional<int64_t> batch_limit);
+};
+
+}  // namespace kafka
+}  // namespace integrations
diff --git a/src/integrations/kafka/exceptions.hpp b/src/integrations/kafka/exceptions.hpp
new file mode 100644
index 000000000..25e74d024
--- /dev/null
+++ b/src/integrations/kafka/exceptions.hpp
@@ -0,0 +1,59 @@
+#pragma once
+
+#include "utils/exceptions.hpp"
+
+#include <fmt/format.h>
+
+class KafkaStreamException : public utils::BasicException {
+  using utils::BasicException::BasicException;
+};
+
+class StreamExistsException : public KafkaStreamException {
+ public:
+  explicit StreamExistsException(const std::string &stream_name)
+      : KafkaStreamException(
+            fmt::format("Kafka stream {} already exists.", stream_name)) {}
+};
+
+class StreamDoesntExistException : public KafkaStreamException {
+ public:
+  explicit StreamDoesntExistException(const std::string &stream_name)
+      : KafkaStreamException(
+            fmt::format("Kafka stream {} doesn't exist.", stream_name)) {}
+};
+
+class ConsumerFailedToInitializeException : public KafkaStreamException {
+ public:
+  ConsumerFailedToInitializeException(const std::string &stream_name,
+                                      const std::string &error)
+      : KafkaStreamException(fmt::format(
+            "Failed to initialize kafka stream {} : {}", stream_name, error)) {}
+};
+
+class ConsumerNotAvailableException : public KafkaStreamException {
+ public:
+  ConsumerNotAvailableException(const std::string &stream_name)
+      : KafkaStreamException(
+            fmt::format("Kafka stream {} not available", stream_name)) {}
+};
+
+class ConsumerRunningException : public KafkaStreamException {
+ public:
+  ConsumerRunningException(const std::string &stream_name)
+      : KafkaStreamException(
+            fmt::format("Kafka stream {} is already running", stream_name)) {}
+};
+
+class ConsumerStoppedException : public KafkaStreamException {
+ public:
+  ConsumerStoppedException(const std::string &stream_name)
+      : KafkaStreamException(
+            fmt::format("Kafka stream {} is already stopped", stream_name)) {}
+};
+
+class TopicNotFoundException : public KafkaStreamException {
+ public:
+  TopicNotFoundException(const std::string &stream_name)
+      : KafkaStreamException(
+            fmt::format("Kafka stream {}, topic not found", stream_name)) {}
+};
diff --git a/src/integrations/kafka/streams.cpp b/src/integrations/kafka/streams.cpp
new file mode 100644
index 000000000..7a440a32b
--- /dev/null
+++ b/src/integrations/kafka/streams.cpp
@@ -0,0 +1,68 @@
+#include "integrations/kafka/streams.hpp"
+#include "integrations/kafka/exceptions.hpp"
+
+namespace integrations {
+namespace kafka {
+
+void Streams::CreateStream(const StreamInfo &info) {
+  std::lock_guard<std::mutex> g(mutex_);
+  if (consumers_.find(info.stream_name) != consumers_.end())
+    throw StreamExistsException(info.stream_name);
+
+  consumers_.emplace(info.stream_name, info);
+}
+
+void Streams::DropStream(const std::string &stream_name) {
+  std::lock_guard<std::mutex> g(mutex_);
+  auto find_it = consumers_.find(stream_name);
+  if (find_it == consumers_.end())
+    throw StreamDoesntExistException(stream_name);
+
+  consumers_.erase(find_it);
+}
+
+void Streams::StartStream(const std::string &stream_name,
+                          std::experimental::optional<int64_t> batch_limit) {
+  std::lock_guard<std::mutex> g(mutex_);
+  auto find_it = consumers_.find(stream_name);
+  if (find_it == consumers_.end())
+    throw StreamDoesntExistException(stream_name);
+
+  find_it->second.Start(batch_limit);
+}
+
+void Streams::StopStream(const std::string &stream_name) {
+  std::lock_guard<std::mutex> g(mutex_);
+  auto find_it = consumers_.find(stream_name);
+  if (find_it == consumers_.end())
+    throw StreamDoesntExistException(stream_name);
+
+  find_it->second.Stop();
+}
+
+void Streams::StartAllStreams() {
+  std::lock_guard<std::mutex> g(mutex_);
+  for (auto &consumer_kv : consumers_) {
+    consumer_kv.second.StartIfNotStopped();
+  }
+}
+
+void Streams::StopAllStreams() {
+  std::lock_guard<std::mutex> g(mutex_);
+  for (auto &consumer_kv : consumers_) {
+    consumer_kv.second.StopIfNotRunning();
+  }
+}
+
+std::vector<StreamInfo> Streams::ShowStreams() {
+  std::vector<StreamInfo> streams;
+  std::lock_guard<std::mutex> g(mutex_);
+  for (auto &consumer_kv : consumers_) {
+    streams.emplace_back(consumer_kv.second.info());
+  }
+
+  return streams;
+}
+
+}  // namespace kafka
+}  // namespace integrations
diff --git a/src/integrations/kafka/streams.hpp b/src/integrations/kafka/streams.hpp
new file mode 100644
index 000000000..4d8023788
--- /dev/null
+++ b/src/integrations/kafka/streams.hpp
@@ -0,0 +1,38 @@
+#pragma once
+
+#include "integrations/kafka/consumer.hpp"
+
+#include <experimental/optional>
+#include <mutex>
+#include <unordered_map>
+
+namespace integrations {
+namespace kafka {
+
+class Streams final {
+ public:
+  void CreateStream(const StreamInfo &info);
+
+  void DropStream(const std::string &stream_name);
+
+  void StartStream(const std::string &stream_name,
+                   std::experimental::optional<int64_t> batch_limit =
+                       std::experimental::nullopt);
+
+  void StopStream(const std::string &stream_name);
+
+  void StartAllStreams();
+
+  void StopAllStreams();
+
+  std::vector<StreamInfo> ShowStreams();
+
+ private:
+  std::mutex mutex_;
+  std::unordered_map<std::string, Consumer> consumers_;
+
+  // TODO (msantl): persist stream storage
+};
+
+}  // namespace kafka
+}  // namespace integrations
diff --git a/src/query/frontend/ast/ast.capnp b/src/query/frontend/ast/ast.capnp
index c614bd601..118c13fcc 100644
--- a/src/query/frontend/ast/ast.capnp
+++ b/src/query/frontend/ast/ast.capnp
@@ -410,8 +410,10 @@ struct DropUser {
 struct CreateStream {
   streamName @0 :Text;
   streamUri @1 :Tree;
-  transformUri @2 :Tree;
-  batchInterval @3 :Tree;
+  streamTopic @2 :Tree;
+  transformUri @3 :Tree;
+  batchIntervalInMs @4 :Tree;
+  batchSize @5 :Tree;
 }
 
 struct DropStream {
diff --git a/src/query/frontend/ast/ast.cpp b/src/query/frontend/ast/ast.cpp
index 01014e66b..2762a056c 100644
--- a/src/query/frontend/ast/ast.cpp
+++ b/src/query/frontend/ast/ast.cpp
@@ -1390,12 +1390,20 @@ void CreateStream::Save(capnp::CreateStream::Builder *builder,
   auto stream_uri_builder = builder->getStreamUri();
   stream_uri_->Save(&stream_uri_builder, saved_uids);
 
+  auto stream_topic_builder = builder->getStreamTopic();
+  stream_topic_->Save(&stream_topic_builder, saved_uids);
+
   auto transform_uri_builder = builder->getTransformUri();
   transform_uri_->Save(&transform_uri_builder, saved_uids);
 
-  if (batch_interval_) {
-    auto batch_interval_builder = builder->getBatchInterval();
-    batch_interval_->Save(&batch_interval_builder, saved_uids);
+  if (batch_interval_in_ms_) {
+    auto batch_interval_builder = builder->getBatchIntervalInMs();
+    batch_interval_in_ms_->Save(&batch_interval_builder, saved_uids);
+  }
+
+  if (batch_size_) {
+    auto batch_size_builder = builder->getBatchSize();
+    batch_size_->Save(&batch_size_builder, saved_uids);
   }
 }
 
@@ -1408,16 +1416,28 @@ void CreateStream::Load(const capnp::Tree::Reader &base_reader,
   const auto stream_uri_reader = reader.getStreamUri();
   stream_uri_ =
       dynamic_cast<Expression *>(storage->Load(stream_uri_reader, loaded_uids));
+
+  const auto stream_topic_reader = reader.getStreamTopic();
+  stream_topic_ = dynamic_cast<Expression *>(
+      storage->Load(stream_topic_reader, loaded_uids));
+
   const auto transform_uri_reader = reader.getTransformUri();
   transform_uri_ = dynamic_cast<Expression *>(
       storage->Load(transform_uri_reader, loaded_uids));
 
-  batch_interval_ = nullptr;
-  if (reader.hasBatchInterval()) {
-    const auto batch_interval_reader = reader.getBatchInterval();
-    batch_interval_ = dynamic_cast<Expression *>(
+  batch_interval_in_ms_ = nullptr;
+  if (reader.hasBatchIntervalInMs()) {
+    const auto batch_interval_reader = reader.getBatchIntervalInMs();
+    batch_interval_in_ms_ = dynamic_cast<Expression *>(
         storage->Load(batch_interval_reader, loaded_uids));
   }
+
+  batch_size_ = nullptr;
+  if (reader.hasBatchSize()) {
+    const auto batch_size_reader = reader.getBatchSize();
+    batch_size_ = dynamic_cast<Expression *>(
+        storage->Load(batch_size_reader, loaded_uids));
+  }
 }
 
 CreateStream *CreateStream::Construct(const capnp::CreateStream::Reader &reader,
diff --git a/src/query/frontend/ast/ast.hpp b/src/query/frontend/ast/ast.hpp
index f42548056..fc9eed7ce 100644
--- a/src/query/frontend/ast/ast.hpp
+++ b/src/query/frontend/ast/ast.hpp
@@ -42,14 +42,14 @@ namespace query {
 
 #define CLONE_BINARY_EXPRESSION                                              \
   auto Clone(AstStorage &storage) const->std::remove_const<                  \
-      std::remove_pointer<decltype(this)>::type>::type *override {           \
+      std::remove_pointer<decltype(this)>::type>::type * override {          \
     return storage.Create<                                                   \
         std::remove_cv<std::remove_reference<decltype(*this)>::type>::type>( \
         expression1_->Clone(storage), expression2_->Clone(storage));         \
   }
 #define CLONE_UNARY_EXPRESSION                                               \
   auto Clone(AstStorage &storage) const->std::remove_const<                  \
-      std::remove_pointer<decltype(this)>::type>::type *override {           \
+      std::remove_pointer<decltype(this)>::type>::type * override {          \
     return storage.Create<                                                   \
         std::remove_cv<std::remove_reference<decltype(*this)>::type>::type>( \
         expression_->Clone(storage));                                        \
@@ -801,8 +801,8 @@ class SubscriptOperator : public BinaryOperator {
     return visitor.PostVisit(*this);
   }
   CLONE_BINARY_EXPRESSION;
-  static SubscriptOperator *Construct(
-      capnp::SubscriptOperator::Reader &reader, AstStorage *storage);
+  static SubscriptOperator *Construct(capnp::SubscriptOperator::Reader &reader,
+                                      AstStorage *storage);
 
  protected:
   using BinaryOperator::BinaryOperator;
@@ -813,8 +813,9 @@ class SubscriptOperator : public BinaryOperator {
   friend class boost::serialization::access;
   SERIALIZE_USING_BASE(BinaryOperator);
   template <class TArchive>
-  friend void boost::serialization::load_construct_data(
-      TArchive &, SubscriptOperator *, const unsigned int);
+  friend void boost::serialization::load_construct_data(TArchive &,
+                                                        SubscriptOperator *,
+                                                        const unsigned int);
 };
 
 class ListSlicingOperator : public Expression {
@@ -3673,8 +3674,9 @@ class CreateStream : public Clause {
   CreateStream *Clone(AstStorage &storage) const override {
     return storage.Create<CreateStream>(
         stream_name_, stream_uri_->Clone(storage),
-        transform_uri_->Clone(storage),
-        batch_interval_ ? batch_interval_->Clone(storage) : nullptr);
+        stream_topic_->Clone(storage), transform_uri_->Clone(storage),
+        batch_interval_in_ms_ ? batch_interval_in_ms_->Clone(storage) : nullptr,
+        batch_size_ ? batch_size_->Clone(storage) : nullptr);
   }
 
   static CreateStream *Construct(const capnp::CreateStream::Reader &reader,
@@ -3683,18 +3685,23 @@ class CreateStream : public Clause {
 
   std::string stream_name_;
   Expression *stream_uri_;
+  Expression *stream_topic_;
   Expression *transform_uri_;
-  Expression *batch_interval_;
+  Expression *batch_interval_in_ms_;
+  Expression *batch_size_;
 
  protected:
   explicit CreateStream(int uid) : Clause(uid) {}
   CreateStream(int uid, std::string stream_name, Expression *stream_uri,
-               Expression *transform_uri, Expression *batch_interval)
+               Expression *stream_topic, Expression *transform_uri,
+               Expression *batch_interval_in_ms, Expression *batch_size)
       : Clause(uid),
         stream_name_(std::move(stream_name)),
         stream_uri_(stream_uri),
+        stream_topic_(stream_topic),
         transform_uri_(transform_uri),
-        batch_interval_(batch_interval) {}
+        batch_interval_in_ms_(batch_interval_in_ms),
+        batch_size_(batch_size) {}
 
   void Save(capnp::Clause::Builder *builder,
             std::vector<int> *saved_uids) override;
@@ -3711,8 +3718,10 @@ class CreateStream : public Clause {
     ar &boost::serialization::base_object<Clause>(*this);
     ar &stream_name_;
     ar &stream_uri_;
+    ar &stream_topic_;
     ar &transform_uri_;
-    ar &batch_interval_;
+    ar &batch_interval_in_ms_;
+    ar &batch_size_;
   }
 
   template <class TArchive>
diff --git a/src/query/frontend/ast/cypher_main_visitor.cpp b/src/query/frontend/ast/cypher_main_visitor.cpp
index 28a5a8cfb..b9e2c3100 100644
--- a/src/query/frontend/ast/cypher_main_visitor.cpp
+++ b/src/query/frontend/ast/cypher_main_visitor.cpp
@@ -324,17 +324,30 @@ antlrcpp::Any CypherMainVisitor::visitCreateStream(
     throw SyntaxException("Stream URI should be a string literal!");
   }
   Expression *stream_uri = ctx->streamUri->accept(this);
+
+  if (!ctx->streamTopic->StringLiteral()) {
+    throw SyntaxException("Topic should be a string literal!");
+  }
+  Expression *stream_topic = ctx->streamTopic->accept(this);
+
   if (!ctx->transformUri->StringLiteral()) {
     throw SyntaxException("Transform URI should be a string literal!");
   }
   Expression *transform_uri = ctx->transformUri->accept(this);
-  Expression *batch_interval = nullptr;
 
+  Expression *batch_interval_in_ms = nullptr;
   if (ctx->batchIntervalOption()) {
-    batch_interval = ctx->batchIntervalOption()->accept(this);
+    batch_interval_in_ms = ctx->batchIntervalOption()->accept(this);
   }
-  return storage_.Create<CreateStream>(stream_name, stream_uri, transform_uri,
-                                       batch_interval);
+
+  Expression *batch_size = nullptr;
+  if (ctx->batchSizeOption()) {
+    batch_size = ctx->batchSizeOption()->accept(this);
+  }
+
+  return storage_.Create<CreateStream>(stream_name, stream_uri, stream_topic,
+                                       transform_uri, batch_interval_in_ms,
+                                       batch_size);
 }
 
 /**
@@ -349,6 +362,18 @@ antlrcpp::Any CypherMainVisitor::visitBatchIntervalOption(
   return ctx->literal()->accept(this);
 }
 
+/**
+ * @return Expression*
+ */
+antlrcpp::Any CypherMainVisitor::visitBatchSizeOption(
+    CypherParser::BatchSizeOptionContext *ctx) {
+  if (!ctx->literal()->numberLiteral() ||
+      !ctx->literal()->numberLiteral()->integerLiteral()) {
+    throw SyntaxException("Batch size should be an integer literal!");
+  }
+  return ctx->literal()->accept(this);
+}
+
 /**
  * @return DropStream*
  */
diff --git a/src/query/frontend/ast/cypher_main_visitor.hpp b/src/query/frontend/ast/cypher_main_visitor.hpp
index 0731fd1e8..b1ed8539f 100644
--- a/src/query/frontend/ast/cypher_main_visitor.hpp
+++ b/src/query/frontend/ast/cypher_main_visitor.hpp
@@ -195,6 +195,9 @@ class CypherMainVisitor : public antlropencypher::CypherBaseVisitor {
   antlrcpp::Any visitBatchIntervalOption(
       CypherParser::BatchIntervalOptionContext *ctx) override;
 
+  antlrcpp::Any visitBatchSizeOption(
+      CypherParser::BatchSizeOptionContext *ctx) override;
+
   /**
    * @return DropStream*
    */
diff --git a/src/query/frontend/opencypher/grammar/Cypher.g4 b/src/query/frontend/opencypher/grammar/Cypher.g4
index b80da9847..c5f7a5455 100644
--- a/src/query/frontend/opencypher/grammar/Cypher.g4
+++ b/src/query/frontend/opencypher/grammar/Cypher.g4
@@ -300,9 +300,13 @@ dropUser : DROP SP USER SP userName ( SP? ',' SP? userName )* ;
 
 streamName : UnescapedSymbolicName ;
 
-createStream : CREATE SP STREAM SP streamName SP AS SP LOAD SP DATA SP KAFKA SP streamUri=literal SP WITH SP TRANSFORM SP transformUri=literal ( SP batchIntervalOption )? ;
+createStream : CREATE SP STREAM SP streamName SP AS SP LOAD SP DATA SP KAFKA SP
+streamUri=literal SP WITH SP TOPIC SP streamTopic=literal SP WITH SP TRANSFORM
+SP transformUri=literal ( SP batchIntervalOption )? (SP batchSizeOption )? ;
 
-batchIntervalOption : BATCH SP INTERVAL SP literal ;
+batchIntervalOption : BATCH_INTERVAL SP literal ;
+
+batchSizeOption : BATCH_SIZE SP literal ;
 
 dropStream : DROP SP STREAM SP streamName ;
 
@@ -427,8 +431,8 @@ symbolicName : UnescapedSymbolicName
              | DATA
              | KAFKA
              | TRANSFORM
-             | BATCH
-             | INTERVAL
+             | BATCH_SIZE
+             | BATCH_INTERVAL
              | SHOW
              | START
              | STOP
@@ -559,9 +563,9 @@ KAFKA : ( 'K' | 'k' ) ( 'A' | 'a' ) ( 'F' | 'f' ) ( 'K' | 'k' ) ( 'A' | 'a' ) ;
 
 TRANSFORM : ( 'T' | 't' ) ( 'R' | 'r' ) ( 'A' | 'a' ) ( 'N' | 'n' ) ( 'S' | 's') ( 'F' | 'f' ) ( 'O' | 'o' ) ( 'R' | 'r' ) ( 'M' | 'm' ) ;
 
-BATCH : ( 'B' | 'b' ) ( 'A' | 'a' ) ( 'T' | 't' ) ( 'C' | 'c' ) ( 'H' | 'h' ) ;
+BATCH_SIZE : ( 'B' | 'b' ) ( 'A' | 'a' ) ( 'T' | 't' ) ( 'C' | 'c' ) ( 'H' | 'h') '_' ( 'S' | 's' ) ( 'I' | 'i' ) ( 'Z' | 'z' ) ( 'E' | 'e' ) ;
 
-INTERVAL : ( 'I' | 'i' ) ( 'N' | 'n' ) ( 'T' | 't' ) ( 'E' | 'e' ) ( 'R' | 'r' ) ( 'V' | 'v' ) ( 'A' | 'a' ) ( 'L' | 'l' ) ;
+BATCH_INTERVAL : ( 'B' | 'b' ) ( 'A' | 'a' ) ( 'T' | 't' ) ( 'C' | 'c' ) ( 'H' | 'h' ) '_' ( 'I' | 'i' ) ( 'N' | 'n' ) ( 'T' | 't' ) ( 'E' | 'e' ) ( 'R' | 'r' ) ( 'V' | 'v' ) ( 'A' | 'a' ) ( 'L' | 'l' ) ;
 
 SHOW : ( 'S' | 's' ) ( 'H' | 'h' ) ( 'O' | 'o' ) ( 'W' | 'w' ) ;
 
@@ -571,6 +575,8 @@ STOP : ( 'S' | 's' ) ( 'T' | 't' ) ( 'O' | 'o' ) ( 'P' | 'p' ) ;
 
 BATCHES : ( 'B' | 'b' )  ( 'A' | 'a' ) ( 'T' | 't' ) ( 'C' | 'c' ) ( 'H' | 'h' ) ( 'E' | 'e' ) ( 'S' | 's' ) ;
 
+TOPIC : ( 'T' | 't' ) ( 'O' | 'o' ) ( 'P' | 'p' ) ( 'I' | 'i' ) ( 'C' | 'c' ) ;
+
 UnescapedSymbolicName : IdentifierStart ( IdentifierPart )* ;
 
 /**
diff --git a/src/query/frontend/stripped_lexer_constants.hpp b/src/query/frontend/stripped_lexer_constants.hpp
index 6afa7417b..47189e53e 100644
--- a/src/query/frontend/stripped_lexer_constants.hpp
+++ b/src/query/frontend/stripped_lexer_constants.hpp
@@ -89,7 +89,7 @@ const trie::Trie kKeywords = {
     "extract",  "any",   "none",      "single", "true",       "false",
     "reduce",   "user",  "password",  "alter",  "drop",       "stream",
     "streams",  "load",  "data",      "kafka",  "transform",  "batch",
-    "interval", "show",  "start",     "stop"};
+    "interval", "show",  "start",     "stop",   "size",       "topic"};
 
 // Unicode codepoints that are allowed at the start of the unescaped name.
 const std::bitset<kBitsetSize> kUnescapedNameAllowedStarts(std::string(
diff --git a/src/query/plan/operator.cpp b/src/query/plan/operator.cpp
index 0d733f486..e227a8bfc 100644
--- a/src/query/plan/operator.cpp
+++ b/src/query/plan/operator.cpp
@@ -21,6 +21,7 @@
 #include "distributed/pull_rpc_clients.hpp"
 #include "distributed/updates_rpc_clients.hpp"
 #include "distributed/updates_rpc_server.hpp"
+#include "integrations/kafka/exceptions.hpp"
 #include "query/context.hpp"
 #include "query/exceptions.hpp"
 #include "query/frontend/ast/ast.hpp"
@@ -3871,28 +3872,67 @@ std::unique_ptr<Cursor> DropUser::MakeCursor(
 }
 
 CreateStream::CreateStream(std::string stream_name, Expression *stream_uri,
-                           Expression *transform_uri,
-                           Expression *batch_interval)
+                           Expression *stream_topic, Expression *transform_uri,
+                           Expression *batch_interval_in_ms,
+                           Expression *batch_size)
     : stream_name_(std::move(stream_name)),
       stream_uri_(stream_uri),
+      stream_topic_(stream_topic),
       transform_uri_(transform_uri),
-      batch_interval_(batch_interval) {}
+      batch_interval_in_ms_(batch_interval_in_ms),
+      batch_size_(batch_size) {}
 
 WITHOUT_SINGLE_INPUT(CreateStream)
 
 class CreateStreamCursor : public Cursor {
+  using StreamInfo = integrations::kafka::StreamInfo;
+
  public:
-  CreateStreamCursor(const CreateStream &, database::GraphDbAccessor &) {}
+  CreateStreamCursor(const CreateStream &self, database::GraphDbAccessor &db)
+      : self_(self), db_(db) {}
 
   bool Pull(Frame &frame, Context &ctx) override {
     if (ctx.in_explicit_transaction_) {
       throw StreamClauseInMulticommandTxException();
     }
+    ExpressionEvaluator evaluator(frame, &ctx, GraphView::OLD);
 
-    throw utils::NotYetImplemented("Create Stream");
+    TypedValue stream_uri = self_.stream_uri()->Accept(evaluator);
+    TypedValue stream_topic = self_.stream_topic()->Accept(evaluator);
+    TypedValue transform_uri = self_.transform_uri()->Accept(evaluator);
+
+    std::experimental::optional<int64_t> batch_interval_in_ms, batch_size;
+
+    if (self_.batch_interval_in_ms()) {
+      batch_interval_in_ms =
+          self_.batch_interval_in_ms()->Accept(evaluator).Value<int64_t>();
+    }
+    if (self_.batch_size()) {
+      batch_size = self_.batch_size()->Accept(evaluator).Value<int64_t>();
+    }
+
+    try {
+      StreamInfo info;
+      info.stream_name = self_.stream_name();
+      info.stream_uri = stream_uri.Value<std::string>();
+      info.stream_topic = stream_topic.Value<std::string>();
+      info.transform_uri = transform_uri.Value<std::string>();
+      info.batch_interval_in_ms = batch_interval_in_ms;
+      info.batch_size = batch_size;
+
+      db_.db().kafka_streams().CreateStream(info);
+    } catch (const KafkaStreamException &e) {
+      throw QueryRuntimeException(e.what());
+    }
+
+    return false;
   }
 
   void Reset() override { throw utils::NotYetImplemented("Create Stream"); }
+
+ private:
+  const CreateStream &self_;
+  database::GraphDbAccessor &db_;
 };
 
 std::unique_ptr<Cursor> CreateStream::MakeCursor(
@@ -3907,17 +3947,27 @@ WITHOUT_SINGLE_INPUT(DropStream)
 
 class DropStreamCursor : public Cursor {
  public:
-  DropStreamCursor(const DropStream &, database::GraphDbAccessor &) {}
+  DropStreamCursor(const DropStream &self, database::GraphDbAccessor &db)
+      : self_(self), db_(db) {}
 
   bool Pull(Frame &frame, Context &ctx) override {
     if (ctx.in_explicit_transaction_) {
       throw StreamClauseInMulticommandTxException();
     }
 
-    throw utils::NotYetImplemented("Drop Stream");
+    try {
+      db_.db().kafka_streams().DropStream(self_.stream_name());
+    } catch (const KafkaStreamException &e) {
+      throw QueryRuntimeException(e.what());
+    }
+    return false;
   }
 
   void Reset() override { throw utils::NotYetImplemented("Drop Stream"); }
+
+ private:
+  const DropStream &self_;
+  database::GraphDbAccessor &db_;
 };
 
 std::unique_ptr<Cursor> DropStream::MakeCursor(
@@ -3925,21 +3975,61 @@ std::unique_ptr<Cursor> DropStream::MakeCursor(
   return std::make_unique<DropStreamCursor>(*this, db);
 }
 
+ShowStreams::ShowStreams(Symbol name_symbol, Symbol uri_symbol,
+                         Symbol topic_symbol, Symbol transform_symbol,
+                         Symbol status_symbol)
+    : name_symbol_(name_symbol),
+      uri_symbol_(uri_symbol),
+      topic_symbol_(topic_symbol),
+      transform_symbol_(transform_symbol),
+      status_symbol_(status_symbol) {}
+
 WITHOUT_SINGLE_INPUT(ShowStreams)
 
+std::vector<Symbol> ShowStreams::OutputSymbols(const SymbolTable &) const {
+  return {name_symbol_, uri_symbol_, topic_symbol_, transform_symbol_,
+          status_symbol_};
+}
+
 class ShowStreamsCursor : public Cursor {
  public:
-  ShowStreamsCursor(const ShowStreams &, database::GraphDbAccessor &) {}
+  ShowStreamsCursor(const ShowStreams &self, database::GraphDbAccessor &db)
+      : self_(self), db_(db) {}
 
   bool Pull(Frame &frame, Context &ctx) override {
     if (ctx.in_explicit_transaction_) {
       throw StreamClauseInMulticommandTxException();
     }
 
-    throw utils::NotYetImplemented("Show Streams");
+    if (!is_initialized_) {
+      streams_ = db_.db().kafka_streams().ShowStreams();
+      streams_it_ = streams_.begin();
+      is_initialized_ = true;
+    }
+
+    if (streams_it_ == streams_.end()) return false;
+
+    frame[self_.name_symbol()] = streams_it_->stream_name;
+    frame[self_.uri_symbol()] = streams_it_->stream_uri;
+    frame[self_.topic_symbol()] = streams_it_->stream_topic;
+    frame[self_.transform_symbol()] = streams_it_->transform_uri;
+    frame[self_.status_symbol()] = streams_it_->is_running;
+
+    streams_it_++;
+
+    return true;
   }
 
   void Reset() override { throw utils::NotYetImplemented("Show Streams"); }
+
+ private:
+  const ShowStreams &self_;
+  database::GraphDbAccessor &db_;
+
+  bool is_initialized_ = false;
+  using StreamInfo = integrations::kafka::StreamInfo;
+  std::vector<StreamInfo> streams_;
+  std::vector<StreamInfo>::iterator streams_it_ = streams_.begin();
 };
 
 std::unique_ptr<Cursor> ShowStreams::MakeCursor(
@@ -3957,18 +4047,33 @@ WITHOUT_SINGLE_INPUT(StartStopStream)
 
 class StartStopStreamCursor : public Cursor {
  public:
-  StartStopStreamCursor(const StartStopStream &,
-                        database::GraphDbAccessor &db) {}
+  StartStopStreamCursor(const StartStopStream &self,
+                        database::GraphDbAccessor &db)
+      : self_(self), db_(db) {}
 
   bool Pull(Frame &frame, Context &ctx) override {
     if (ctx.in_explicit_transaction_) {
       throw StreamClauseInMulticommandTxException();
     }
 
-    throw utils::NotYetImplemented("Start/Stop Stream");
+    try {
+      if (self_.is_start()) {
+        db_.db().kafka_streams().StartStream(self_.stream_name());
+      } else {
+        db_.db().kafka_streams().StopStream(self_.stream_name());
+      }
+    } catch (const KafkaStreamException &e) {
+      throw QueryRuntimeException(e.what());
+    }
+
+    return false;
   }
 
   void Reset() override { throw utils::NotYetImplemented("Start/Stop Stream"); }
+
+ private:
+  const StartStopStream &self_;
+  database::GraphDbAccessor &db_;
 };
 
 std::unique_ptr<Cursor> StartStopStream::MakeCursor(
@@ -3982,20 +4087,35 @@ WITHOUT_SINGLE_INPUT(StartStopAllStreams)
 
 class StartStopAllStreamsCursor : public Cursor {
  public:
-  StartStopAllStreamsCursor(const StartStopAllStreams &,
-                            database::GraphDbAccessor &) {}
+  StartStopAllStreamsCursor(const StartStopAllStreams &self,
+                            database::GraphDbAccessor &db)
+      : self_(self), db_(db) {}
 
   bool Pull(Frame &frame, Context &ctx) override {
     if (ctx.in_explicit_transaction_) {
       throw StreamClauseInMulticommandTxException();
     }
 
-    throw utils::NotYetImplemented("Start/Stop All Streams");
+    try {
+      if (self_.is_start()) {
+        db_.db().kafka_streams().StartAllStreams();
+      } else {
+        db_.db().kafka_streams().StopAllStreams();
+      }
+    } catch (const KafkaStreamException &e) {
+      throw QueryRuntimeException(e.what());
+    }
+
+    return false;
   }
 
   void Reset() override {
     throw utils::NotYetImplemented("Start/Stop All Streams");
   }
+
+ private:
+  const StartStopAllStreams &self_;
+  database::GraphDbAccessor &db_;
 };
 
 std::unique_ptr<Cursor> StartStopAllStreams::MakeCursor(
diff --git a/src/query/plan/operator.lcp b/src/query/plan/operator.lcp
index f99264190..5e43a45fa 100644
--- a/src/query/plan/operator.lcp
+++ b/src/query/plan/operator.lcp
@@ -2527,6 +2527,14 @@ by having only one result from each worker.")
              :capnp-init nil
              :capnp-save #'save-ast-pointer
              :capnp-load (load-ast-pointer "Expression *"))
+    (stream-topic "Expression *"
+             :reader t
+             :save-fun #'save-pointer
+             :load-fun #'load-pointer
+             :capnp-type "Ast.Tree"
+             :capnp-init nil
+             :capnp-save #'save-ast-pointer
+             :capnp-load (load-ast-pointer "Expression *"))
     (transform-uri "Expression *"
              :reader t
              :save-fun #'save-pointer
@@ -2535,7 +2543,15 @@ by having only one result from each worker.")
              :capnp-init nil
              :capnp-save #'save-ast-pointer
              :capnp-load (load-ast-pointer "Expression *"))
-    (batch-interval "Expression *"
+    (batch-interval-in-ms "Expression *"
+             :reader t
+             :save-fun #'save-pointer
+             :load-fun #'load-pointer
+             :capnp-type "Ast.Tree"
+             :capnp-init nil
+             :capnp-save #'save-ast-pointer
+             :capnp-load (load-ast-pointer "Expression *"))
+    (batch-size "Expression *"
              :reader t
              :save-fun #'save-pointer
              :load-fun #'load-pointer
@@ -2547,8 +2563,9 @@ by having only one result from each worker.")
    "Creates a new stream with given parameters that imports data.")
   (:public
    #>cpp
-   CreateStream(std::string stream_name, Expression* stream_uri,
-             Expression* transform_uri, Expression* batch_interval);
+   CreateStream(std::string stream_name, Expression *stream_uri,
+             Expression *stream_topic, Expression *transform_uri,
+             Expression *batch_interval_in_ms, Expression *batch_size);
    DEFVISITABLE(HierarchicalLogicalOperatorVisitor);
    std::unique_ptr<Cursor> MakeCursor(
        database::GraphDbAccessor &db) const override;
@@ -2587,15 +2604,21 @@ by having only one result from each worker.")
   (:serialize :boost :capnp))
 
 (lcp:define-class show-streams (logical-operator)
-  ()
+  ((name-symbol "Symbol" :reader t)
+   (uri-symbol "Symbol" :reader t)
+   (topic-symbol "Symbol" :reader t)
+   (transform-symbol "Symbol" :reader t)
+   (status-symbol "Symbol" :reader t))
   (:documentation
    "Shows all streams, started and stopped, that were configured.")
   (:public
    #>cpp
-   ShowStreams() {}
+   ShowStreams(Symbol name_symbol, Symbol endpoint_symbol, Symbol topic_symbol,
+               Symbol transform_symbol, Symbol status_symbol);
    DEFVISITABLE(HierarchicalLogicalOperatorVisitor);
    std::unique_ptr<Cursor> MakeCursor(
        database::GraphDbAccessor &db) const override;
+   std::vector<Symbol> OutputSymbols(const SymbolTable &) const override;
    std::vector<Symbol> ModifiedSymbols(const SymbolTable &) const override {
      return {};
    }
@@ -2604,6 +2627,8 @@ by having only one result from each worker.")
    std::shared_ptr<LogicalOperator> input() const override;
    void set_input(std::shared_ptr<LogicalOperator>) override;
    cpp<#)
+  (:private
+   #>cpp ShowStreams() {} cpp<#)
   (:serialize :boost :capnp))
 
 (lcp:define-class start-stop-stream (logical-operator)
diff --git a/src/query/plan/rule_based_planner.hpp b/src/query/plan/rule_based_planner.hpp
index 82e537187..a510d741f 100644
--- a/src/query/plan/rule_based_planner.hpp
+++ b/src/query/plan/rule_based_planner.hpp
@@ -196,7 +196,8 @@ class RuleBasedPlanner {
           DCHECK(!input_op) << "Unexpected operator before CreateStream";
           input_op = std::make_unique<plan::CreateStream>(
               create_stream->stream_name_, create_stream->stream_uri_,
-              create_stream->transform_uri_, create_stream->batch_interval_);
+              create_stream->stream_topic_, create_stream->transform_uri_,
+              create_stream->batch_interval_in_ms_, create_stream->batch_size_);
         } else if (auto *drop_stream =
                        dynamic_cast<query::DropStream *>(clause)) {
           DCHECK(!input_op) << "Unexpected operator before DropStream";
@@ -204,7 +205,14 @@ class RuleBasedPlanner {
               std::make_unique<plan::DropStream>(drop_stream->stream_name_);
         } else if (dynamic_cast<query::ShowStreams *>(clause)) {
           DCHECK(!input_op) << "Unexpected operator before ShowStreams";
-          input_op = std::make_unique<plan::ShowStreams>();
+          // Create symbols for ShowStreams results
+          auto &symbol_table = context.symbol_table;
+          input_op = std::make_unique<plan::ShowStreams>(
+              symbol_table.CreateSymbol("name", false),
+              symbol_table.CreateSymbol("uri", false),
+              symbol_table.CreateSymbol("topic", false),
+              symbol_table.CreateSymbol("transform", false),
+              symbol_table.CreateSymbol("is running", false));
         } else if (auto *start_stop_stream =
                        dynamic_cast<query::StartStopStream *>(clause)) {
           DCHECK(!input_op) << "Unexpected operator before StartStopStream";
diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt
index 31b35f3e1..e5fe56cdd 100644
--- a/tests/unit/CMakeLists.txt
+++ b/tests/unit/CMakeLists.txt
@@ -134,7 +134,7 @@ add_unit_test(interpreter.cpp)
 target_link_libraries(${test_prefix}interpreter memgraph_lib kvstore_dummy_lib)
 
 add_unit_test(kvstore.cpp)
-target_link_libraries(${test_prefix}kvstore gtest gtest_main memgraph_lib kvstore_lib)
+target_link_libraries(${test_prefix}kvstore kvstore_lib memgraph_lib glog)
 
 add_unit_test(metrics.cpp)
 target_link_libraries(${test_prefix}metrics memgraph_lib kvstore_dummy_lib)
@@ -161,7 +161,7 @@ add_unit_test(pod_buffer.cpp)
 target_link_libraries(${test_prefix}pod_buffer memgraph_lib kvstore_dummy_lib)
 
 add_unit_test(property_value_store.cpp)
-target_link_libraries(${test_prefix}property_value_store memgraph_lib kvstore_lib)
+target_link_libraries(${test_prefix}property_value_store kvstore_lib memgraph_lib)
 
 add_unit_test(query_cost_estimator.cpp)
 target_link_libraries(${test_prefix}query_cost_estimator memgraph_lib kvstore_dummy_lib)
diff --git a/tests/unit/cypher_main_visitor.cpp b/tests/unit/cypher_main_visitor.cpp
index 3cadef81e..9d4eb68bf 100644
--- a/tests/unit/cypher_main_visitor.cpp
+++ b/tests/unit/cypher_main_visitor.cpp
@@ -1963,8 +1963,10 @@ TYPED_TEST(CypherMainVisitorTest, DropUser) {
 TYPED_TEST(CypherMainVisitorTest, CreateStream) {
   auto check_create_stream = [](
       std::string input, const std::string &stream_name,
-      const std::string &stream_uri, const std::string &transform_uri,
-      std::experimental::optional<int64_t> batch_interval) {
+      const std::string &stream_uri, const std::string &stream_topic,
+      const std::string &transform_uri,
+      std::experimental::optional<int64_t> batch_interval_in_ms,
+      std::experimental::optional<int64_t> batch_size) {
     TypeParam ast_generator(input);
     auto *query = ast_generator.query_;
     ASSERT_TRUE(query->single_query_);
@@ -1977,47 +1979,95 @@ TYPED_TEST(CypherMainVisitorTest, CreateStream) {
     ASSERT_TRUE(create_stream->stream_uri_);
     CheckLiteral(ast_generator.context_, create_stream->stream_uri_,
                  TypedValue(stream_uri));
+    ASSERT_TRUE(create_stream->stream_topic_);
+    CheckLiteral(ast_generator.context_, create_stream->stream_topic_,
+                 TypedValue(stream_topic));
     ASSERT_TRUE(create_stream->transform_uri_);
     CheckLiteral(ast_generator.context_, create_stream->transform_uri_,
                  TypedValue(transform_uri));
-    if (batch_interval) {
-      ASSERT_TRUE(create_stream->batch_interval_);
-      CheckLiteral(ast_generator.context_, create_stream->batch_interval_,
-                   TypedValue(*batch_interval));
+    if (batch_interval_in_ms) {
+      ASSERT_TRUE(create_stream->batch_interval_in_ms_);
+      CheckLiteral(ast_generator.context_, create_stream->batch_interval_in_ms_,
+                   TypedValue(*batch_interval_in_ms));
     } else {
-      EXPECT_EQ(create_stream->batch_interval_, nullptr);
+      EXPECT_EQ(create_stream->batch_interval_in_ms_, nullptr);
+    }
+    if (batch_size) {
+      ASSERT_TRUE(create_stream->batch_size_);
+      CheckLiteral(ast_generator.context_, create_stream->batch_size_,
+                   TypedValue(*batch_size));
+    } else {
+      EXPECT_EQ(create_stream->batch_size_, nullptr);
     }
   };
 
   check_create_stream(
       "CREATE STREAM strim AS LOAD DATA KAFKA 'localhost' "
+      "WITH TOPIC 'tropika' "
       "WITH TRANSFORM 'localhost/test.py'",
-      "strim", "localhost", "localhost/test.py", std::experimental::nullopt);
+      "strim", "localhost", "tropika", "localhost/test.py",
+      std::experimental::nullopt, std::experimental::nullopt);
+
   check_create_stream(
       "CreaTE StreaM strim AS LOad daTA KAFKA 'localhost' "
-      "WITH TRAnsFORM 'localhost/test.py' bAtCH inTErvAL 168",
-      "strim", "localhost", "localhost/test.py", 168);
+      "WitH TopIC 'tropika' "
+      "WITH TRAnsFORM 'localhost/test.py' bAtCH_inTErvAL 168",
+      "strim", "localhost", "tropika", "localhost/test.py", 168,
+      std::experimental::nullopt);
+
+  check_create_stream(
+      "CreaTE StreaM strim AS LOad daTA KAFKA 'localhost' "
+      "WITH TopIC 'tropika' "
+      "WITH TRAnsFORM 'localhost/test.py' bAtCH_SizE 17",
+      "strim", "localhost", "tropika", "localhost/test.py",
+      std::experimental::nullopt, 17);
+
+  check_create_stream(
+      "CreaTE StreaM strim AS LOad daTA KAFKA 'localhost' "
+      "WitH TOPic 'tropika' "
+      "WITH TRAnsFORM 'localhost/test.py' bAtCH_inTErvAL 168 Batch_SIze 17",
+      "strim", "localhost", "tropika", "localhost/test.py", 168, 17);
 
   EXPECT_THROW(check_create_stream(
                    "CREATE STREAM strim AS LOAD DATA KAFKA 'localhost' "
-                   "WITH TRANSFORM 'localhost/test.py' BATCH INTERVAL 'jedan' ",
-                   "strim", "localhost", "localhost/test.py", 168),
+                   "WITH TRANSFORM 'localhost/test.py' BATCH_INTERVAL 'jedan' ",
+                   "strim", "localhost", "tropika", "localhost/test.py", 168,
+                   std::experimental::nullopt),
+               SyntaxException);
+  EXPECT_THROW(check_create_stream(
+                   "CREATE STREAM strim AS LOAD DATA KAFKA 'localhost' "
+                   "WITH TOPIC 'tropika' "
+                   "WITH TRANSFORM 'localhost/test.py' BATCH_SIZE 'jedan' ",
+                   "strim", "localhost", "tropika", "localhost/test.py",
+                   std::experimental::nullopt, 17),
                SyntaxException);
   EXPECT_THROW(check_create_stream(
                    "CREATE STREAM 123 AS LOAD DATA KAFKA 'localhost' "
-                   "WITH TRANSFORM 'localhost/test.py' BATCH INTERVAL 168 ",
-                   "strim", "localhost", "localhost/test.py", 168),
+                   "WITH TOPIC 'tropika' "
+                   "WITH TRANSFORM 'localhost/test.py' BATCH_INTERVAL 168 ",
+                   "strim", "localhost", "tropika", "localhost/test.py", 168,
+                   std::experimental::nullopt),
+               SyntaxException);
+  EXPECT_THROW(check_create_stream(
+                   "CREATE STREAM strim AS LOAD DATA KAFKA localhost "
+                   "WITH TOPIC 'tropika' "
+                   "WITH TRANSFORM 'localhost/test.py'",
+                   "strim", "localhost", "tropika", "localhost/test.py",
+                   std::experimental::nullopt, std::experimental::nullopt),
                SyntaxException);
-  EXPECT_THROW(
-      check_create_stream("CREATE STREAM strim AS LOAD DATA KAFKA localhost "
-                          "WITH TRANSFORM 'localhost/test.py'",
-                          "strim", "localhost", "localhost/test.py",
-                          std::experimental::nullopt),
-      SyntaxException);
   EXPECT_THROW(check_create_stream(
                    "CREATE STREAM strim AS LOAD DATA KAFKA 'localhost' "
-                   "WITH TRANSFORM localhost/test.py BATCH INTERVAL 168 ",
-                   "strim", "localhost", "localhost/test.py", 168),
+                   "WITH TOPIC 2"
+                   "WITH TRANSFORM localhost/test.py BATCH_INTERVAL 168 ",
+                   "strim", "localhost", "tropika", "localhost/test.py", 168,
+                   std::experimental::nullopt),
+               SyntaxException);
+  EXPECT_THROW(check_create_stream(
+                   "CREATE STREAM strim AS LOAD DATA KAFKA 'localhost' "
+                   "WITH TOPIC 'tropika'"
+                   "WITH TRANSFORM localhost/test.py BATCH_INTERVAL 168 ",
+                   "strim", "localhost", "tropika", "localhost/test.py", 168,
+                   std::experimental::nullopt),
                SyntaxException);
 }
 
diff --git a/tests/unit/query_common.hpp b/tests/unit/query_common.hpp
index 4a5074d8f..46542bb47 100644
--- a/tests/unit/query_common.hpp
+++ b/tests/unit/query_common.hpp
@@ -578,9 +578,11 @@ auto GetMerge(AstStorage &storage, Pattern *pattern, OnMatch on_match,
 #define ALTER_USER(username, password) \
   storage.Create<query::ModifyUser>((username), LITERAL(password), false)
 #define DROP_USER(usernames) storage.Create<query::DropUser>((usernames))
-#define CREATE_STREAM(stream_name, stream_uri, transform_uri, batch_interval) \
-  storage.Create<query::CreateStream>((stream_name), LITERAL(stream_uri),     \
-                                      LITERAL(transform_uri), batch_interval)
+#define CREATE_STREAM(stream_name, stream_uri, stream_topic, transform_uri, \
+                      batch_interval, batch_size)                           \
+  storage.Create<query::CreateStream>(                                      \
+      (stream_name), LITERAL(stream_uri), LITERAL(stream_topic),            \
+      LITERAL(transform_uri), (batch_interval), (batch_size))
 #define DROP_STREAM(stream_name) \
   storage.Create<query::DropStream>((stream_name))
 #define SHOW_STREAMS storage.Create<query::ShowStreams>()
diff --git a/tests/unit/query_planner.cpp b/tests/unit/query_planner.cpp
index dcbc4b641..60c6283f9 100644
--- a/tests/unit/query_planner.cpp
+++ b/tests/unit/query_planner.cpp
@@ -486,39 +486,55 @@ class ExpectPullRemoteOrderBy : public OpChecker<PullRemoteOrderBy> {
 class ExpectCreateStream : public OpChecker<CreateStream> {
  public:
   ExpectCreateStream(std::string stream_name, query::Expression *stream_uri,
+                     query::Expression *stream_topic,
                      query::Expression *transform_uri,
-                     query::Expression *batch_interval)
+                     query::Expression *batch_interval_in_ms,
+                     query::Expression *batch_size)
       : stream_name_(stream_name),
         stream_uri_(stream_uri),
+        stream_topic_(stream_topic),
         transform_uri_(transform_uri),
-        batch_interval_(batch_interval) {}
+        batch_interval_in_ms_(batch_interval_in_ms),
+        batch_size_(batch_size) {}
 
   void ExpectOp(CreateStream &create_stream, const SymbolTable &) override {
     EXPECT_EQ(create_stream.stream_name(), stream_name_);
     // TODO: Proper expression equality
     EXPECT_EQ(typeid(create_stream.stream_uri()).hash_code(),
               typeid(stream_uri_).hash_code());
+    EXPECT_EQ(typeid(create_stream.stream_topic()).hash_code(),
+              typeid(stream_topic_).hash_code());
     EXPECT_EQ(typeid(create_stream.transform_uri()).hash_code(),
               typeid(transform_uri_).hash_code());
-    if (batch_interval_ && create_stream.batch_interval()) {
-      EXPECT_EQ(typeid(create_stream.batch_interval()).hash_code(),
-                typeid(batch_interval_).hash_code());
+    if (batch_interval_in_ms_ && create_stream.batch_interval_in_ms()) {
+      EXPECT_EQ(typeid(create_stream.batch_interval_in_ms()).hash_code(),
+                typeid(batch_interval_in_ms_).hash_code());
     } else {
-      EXPECT_TRUE(batch_interval_ == nullptr &&
-                  create_stream.batch_interval() == nullptr);
+      EXPECT_TRUE(batch_interval_in_ms_ == nullptr &&
+                  create_stream.batch_interval_in_ms() == nullptr);
+    }
+    if (batch_size_ && create_stream.batch_size()) {
+      EXPECT_EQ(typeid(create_stream.batch_size()).hash_code(),
+                typeid(batch_size_).hash_code());
+    } else {
+      EXPECT_TRUE(batch_size_ == nullptr &&
+                  create_stream.batch_size() == nullptr);
     }
   }
 
  private:
   std::string stream_name_;
   query::Expression *stream_uri_;
+  query::Expression *stream_topic_;
   query::Expression *transform_uri_;
-  query::Expression *batch_interval_;
+  query::Expression *batch_interval_in_ms_;
+  query::Expression *batch_size_;
 };
 
 class ExpectDropStream : public OpChecker<DropStream> {
  public:
-  ExpectDropStream(std::string stream_name) : stream_name_(stream_name) {}
+  explicit ExpectDropStream(std::string stream_name)
+      : stream_name_(stream_name) {}
 
   void ExpectOp(DropStream &drop_stream, const SymbolTable &) override {
     EXPECT_EQ(drop_stream.stream_name(), stream_name_);
@@ -558,7 +574,7 @@ class ExpectStartStopStream : public OpChecker<StartStopStream> {
 
 class ExpectStartStopAllStreams : public OpChecker<StartStopAllStreams> {
  public:
-  ExpectStartStopAllStreams(bool is_start) : is_start_(is_start) {}
+  explicit ExpectStartStopAllStreams(bool is_start) : is_start_(is_start) {}
 
   void ExpectOp(StartStopAllStreams &start_stop_all_streams,
                 const SymbolTable &) override {
@@ -2347,34 +2363,73 @@ TYPED_TEST(TestPlanner, DropUser) {
 
 TYPED_TEST(TestPlanner, CreateStream) {
   std::string stream_name("kafka"), stream_uri("localhost:1234"),
-      transform_uri("localhost:1234/file.py");
-  int64_t batch_interval = 100;
+      stream_topic("tropik"), transform_uri("localhost:1234/file.py");
+  int64_t batch_interval_in_ms = 100;
+  int64_t batch_size = 10;
 
   {
     FakeDbAccessor dba;
     AstStorage storage;
-    QUERY(SINGLE_QUERY(
-        CREATE_STREAM(stream_name, stream_uri, transform_uri, nullptr)));
-    auto expected = ExpectCreateStream(stream_name, LITERAL(stream_uri),
-                                       LITERAL(transform_uri), nullptr);
+    QUERY(SINGLE_QUERY(CREATE_STREAM(stream_name, stream_uri, stream_topic,
+                                     transform_uri, nullptr, nullptr)));
+    auto expected = ExpectCreateStream(
+        stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+        LITERAL(transform_uri), nullptr, nullptr);
     CheckPlan<TypeParam>(storage, expected);
-    auto expected_distributed = ExpectDistributed(
-        MakeCheckers(ExpectCreateStream(stream_name, LITERAL(stream_uri),
-                                        LITERAL(transform_uri), nullptr)));
+    auto expected_distributed =
+        ExpectDistributed(MakeCheckers(ExpectCreateStream(
+            stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+            LITERAL(transform_uri), nullptr, nullptr)));
     CheckDistributedPlan<TypeParam>(storage, expected_distributed);
   }
   {
     FakeDbAccessor dba;
     AstStorage storage;
-    QUERY(SINGLE_QUERY(CREATE_STREAM(stream_name, stream_uri, transform_uri,
-                                     LITERAL(batch_interval))));
+    QUERY(SINGLE_QUERY(CREATE_STREAM(stream_name, stream_uri, stream_topic,
+                                     transform_uri,
+                                     LITERAL(batch_interval_in_ms), nullptr)));
+    auto expected = ExpectCreateStream(
+        stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+        LITERAL(transform_uri), LITERAL(batch_interval_in_ms), nullptr);
+    CheckPlan<TypeParam>(storage, expected);
+    auto expected_distributed =
+        ExpectDistributed(MakeCheckers(ExpectCreateStream(
+            stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+            LITERAL(transform_uri), LITERAL(batch_interval_in_ms), nullptr)));
+    CheckDistributedPlan<TypeParam>(storage, expected_distributed);
+  }
+  {
+    FakeDbAccessor dba;
+    AstStorage storage;
+    QUERY(SINGLE_QUERY(CREATE_STREAM(stream_name, stream_uri, stream_topic,
+                                     transform_uri, nullptr,
+                                     LITERAL(batch_size))));
+    auto expected = ExpectCreateStream(
+        stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+        LITERAL(transform_uri), nullptr, LITERAL(batch_size));
+    CheckPlan<TypeParam>(storage, expected);
+    auto expected_distributed =
+        ExpectDistributed(MakeCheckers(ExpectCreateStream(
+            stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+            LITERAL(transform_uri), nullptr, LITERAL(batch_size))));
+    CheckDistributedPlan<TypeParam>(storage, expected_distributed);
+  }
+  {
+    FakeDbAccessor dba;
+    AstStorage storage;
+    QUERY(SINGLE_QUERY(
+        CREATE_STREAM(stream_name, stream_uri, stream_topic, transform_uri,
+                      LITERAL(batch_interval_in_ms), LITERAL(batch_size))));
     auto expected =
         ExpectCreateStream(stream_name, LITERAL(stream_uri),
-                           LITERAL(transform_uri), LITERAL(batch_interval));
+                           LITERAL(stream_topic), LITERAL(transform_uri),
+                           LITERAL(batch_interval_in_ms), LITERAL(batch_size));
     CheckPlan<TypeParam>(storage, expected);
-    auto expected_distributed = ExpectDistributed(MakeCheckers(
-        ExpectCreateStream(stream_name, LITERAL(stream_uri),
-                           LITERAL(transform_uri), LITERAL(batch_interval))));
+    auto expected_distributed =
+        ExpectDistributed(MakeCheckers(ExpectCreateStream(
+            stream_name, LITERAL(stream_uri), LITERAL(stream_topic),
+            LITERAL(transform_uri), LITERAL(batch_interval_in_ms),
+            LITERAL(batch_size))));
     CheckDistributedPlan<TypeParam>(storage, expected_distributed);
   }
 }