tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add configs for kafka streams (#328)

Browse Source
This commit is contained in:
János Benjamin Antal
2022-01-31 17:26:53 +01:00
committed by GitHub
parent ced84e17b6
commit 6c00d146f2
21 changed files with 471 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/integrations/constants.hpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -10,11 +10,14 @@
// licenses/APL.txt.
#pragma once
#include <chrono>
#include <string>
namespace integrations {
constexpr int64_t kDefaultCheckBatchLimit{1};
constexpr std::chrono::milliseconds kDefaultCheckTimeout{30000};
constexpr std::chrono::milliseconds kMinimumInterval{1};
constexpr int64_t kMinimumSize{1};
const std::string kReducted{"<REDUCTED>"};
} // namespace integrations

14
src/integrations/kafka/consumer.cpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -126,6 +126,18 @@ Consumer::Consumer(ConsumerInfo info, ConsumerFunction consumer_function)
std::string error;
for (const auto &[key, value] : info_.public_configs) {
if (conf->set(key, value, error) != RdKafka::Conf::CONF_OK) {
throw SettingCustomConfigFailed(info_.consumer_name, error, key, value);
}
}
for (const auto &[key, value] : info_.private_configs) {
if (conf->set(key, value, error) != RdKafka::Conf::CONF_OK) {
throw SettingCustomConfigFailed(info_.consumer_name, error, key, kReducted);
}
}
if (conf->set("event_cb", this, error) != RdKafka::Conf::CONF_OK) {
throw ConsumerFailedToInitializeException(info_.consumer_name, error);
}

4
src/integrations/kafka/consumer.hpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -86,6 +86,8 @@ struct ConsumerInfo {
std::string bootstrap_servers;
std::chrono::milliseconds batch_interval;
int64_t batch_size;
std::unordered_map<std::string, std::string> public_configs;
std::unordered_map<std::string, std::string> private_configs;
};
/// Memgraphs Kafka consumer wrapper.

25
src/integrations/kafka/exceptions.hpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -11,7 +11,7 @@
#pragma once
#include <string>
#include <string_view>
#include "utils/exceptions.hpp"
@@ -22,37 +22,46 @@ class KafkaStreamException : public utils::BasicException {
class ConsumerFailedToInitializeException : public KafkaStreamException {
public:
ConsumerFailedToInitializeException(const std::string &consumer_name, const std::string &error)
ConsumerFailedToInitializeException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Failed to initialize Kafka consumer {} : {}", consumer_name, error) {}
};
class SettingCustomConfigFailed : public ConsumerFailedToInitializeException {
public:
SettingCustomConfigFailed(const std::string_view consumer_name, const std::string_view error,
const std::string_view key, const std::string_view value)
: ConsumerFailedToInitializeException(
consumer_name,
fmt::format(R"(failed to set custom config ("{}": "{}"), because of error {})", key, value, error)) {}
};
class ConsumerRunningException : public KafkaStreamException {
public:
explicit ConsumerRunningException(const std::string &consumer_name)
explicit ConsumerRunningException(const std::string_view consumer_name)
: KafkaStreamException("Kafka consumer {} is already running", consumer_name) {}
};
class ConsumerStoppedException : public KafkaStreamException {
public:
explicit ConsumerStoppedException(const std::string &consumer_name)
explicit ConsumerStoppedException(const std::string_view consumer_name)
: KafkaStreamException("Kafka consumer {} is already stopped", consumer_name) {}
};
class ConsumerCheckFailedException : public KafkaStreamException {
public:
explicit ConsumerCheckFailedException(const std::string &consumer_name, const std::string &error)
explicit ConsumerCheckFailedException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Kafka consumer {} check failed: {}", consumer_name, error) {}
};
class ConsumerStartFailedException : public KafkaStreamException {
public:
explicit ConsumerStartFailedException(const std::string &consumer_name, const std::string &error)
explicit ConsumerStartFailedException(const std::string_view consumer_name, const std::string_view error)
: KafkaStreamException("Starting Kafka consumer {} failed: {}", consumer_name, error) {}
};
class TopicNotFoundException : public KafkaStreamException {
public:
TopicNotFoundException(const std::string &consumer_name, const std::string &topic_name)
TopicNotFoundException(const std::string_view consumer_name, const std::string_view topic_name)
: KafkaStreamException("Kafka consumer {} cannot find topic {}", consumer_name, topic_name) {}
};
} // namespace integrations::kafka

54
src/query/frontend/ast/ast.lcp
View File

@@ -54,9 +54,7 @@ cpp<#
size_t size = 0;
slk::Load(&size, reader);
self->${member}.resize(size);
for (size_t i = 0;
i < size;
++i) {
for (size_t i = 0; i < size; ++i) {
self->${member}[i] = query::LoadAstPointer<query::${type}>(storage, reader);
}
cpp<#))
@@ -75,9 +73,7 @@ cpp<#
#>cpp
size_t size = 0;
slk::Load(&size, reader);
for (size_t i = 0;
i < size;
++i) {
for (size_t i = 0; i < size; ++i) {
query::PropertyIx key;
slk::Load(&key, reader, storage);
auto *value = query::LoadAstPointer<query::Expression>(storage, reader);
@@ -93,6 +89,34 @@ cpp<#
}
cpp<#)
(defun slk-save-expression-map (member)
#>cpp
size_t size = self.${member}.size();
slk::Save(size, builder);
for (const auto &entry : self.${member}) {
query::SaveAstPointer(entry.first, builder);
query::SaveAstPointer(entry.second, builder);
}
cpp<#)
(defun slk-load-expression-map (member)
#>cpp
size_t size = 0;
slk::Load(&size, reader);
for (size_t i = 0; i < size; ++i) {
auto *key = query::LoadAstPointer<query::Expression>(storage, reader);
auto *value = query::LoadAstPointer<query::Expression>(storage, reader);
self->${member}.emplace(key, value);
}
cpp<#)
(defun clone-expression-map (source dest)
#>cpp
for (const auto &[key, value] : ${source}) {
${dest}[key->Clone(storage)] = value->Clone(storage);
}
cpp<#)
(defun slk-load-name-ix (name-type)
(lambda (member)
#>cpp
@@ -1819,9 +1843,7 @@ cpp<#
size_t size = 0;
slk::Load(&size, reader);
self->${member}.resize(size);
for (size_t i = 0;
i < size;
++i) {
for (size_t i = 0; i < size; ++i) {
slk::Load(&self->${member}[i], reader, storage);
}
cpp<#)
@@ -2531,7 +2553,7 @@ cpp<#
:slk-save #'slk-save-ast-pointer
:slk-load (slk-load-ast-pointer "Expression"))
(topic_names "std::variant<Expression*, std::vector<std::string>>" :initval "nullptr"
(topic_names "std::variant<Expression*, std::vector<std::string>>" :initval "nullptr"
:clone #'clone-variant-topic-names
:scope :public)
(consumer_group "std::string" :scope :public)
@@ -2541,7 +2563,17 @@ cpp<#
(service_url "Expression *" :initval "nullptr" :scope :public
:slk-save #'slk-save-ast-pointer
:slk-load (slk-load-ast-pointer "Expression")))
:slk-load (slk-load-ast-pointer "Expression"))
(configs "std::unordered_map<Expression *, Expression *>" :scope :public
:slk-save #'slk-save-expression-map
:slk-load #'slk-load-expression-map
:clone #'clone-expression-map)
(credentials "std::unordered_map<Expression *, Expression *>" :scope :public
:slk-save #'slk-save-expression-map
:slk-load #'slk-load-expression-map
:clone #'clone-expression-map))
(:public
(lcp:define-enum action

44
src/query/frontend/ast/cypher_main_visitor.cpp
View File

@@ -554,7 +554,7 @@ std::string_view ToString(const CommonStreamConfigKey key) {
__VA_ARGS__ \
};
GENERATE_STREAM_CONFIG_KEY_ENUM(Kafka, TOPICS, CONSUMER_GROUP, BOOTSTRAP_SERVERS);
GENERATE_STREAM_CONFIG_KEY_ENUM(Kafka, TOPICS, CONSUMER_GROUP, BOOTSTRAP_SERVERS, CONFIGS, CREDENTIALS);
std::string_view ToString(const KafkaConfigKey key) {
switch (key) {
@@ -564,6 +564,10 @@ std::string_view ToString(const KafkaConfigKey key) {
return "CONSUMER_GROUP";
case KafkaConfigKey::BOOTSTRAP_SERVERS:
return "BOOTSTRAP_SERVERS";
case KafkaConfigKey::CONFIGS:
return "CONFIGS";
case KafkaConfigKey::CREDENTIALS:
return "CREDENTIALS";
}
}
@@ -574,6 +578,21 @@ void MapCommonStreamConfigs(auto &memory, StreamQuery &stream_query) {
}
} // namespace
antlrcpp::Any CypherMainVisitor::visitConfigKeyValuePair(MemgraphCypher::ConfigKeyValuePairContext *ctx) {
MG_ASSERT(ctx->literal().size() == 2);
return std::pair{ctx->literal(0)->accept(this).as<Expression *>(), ctx->literal(1)->accept(this).as<Expression *>()};
}
antlrcpp::Any CypherMainVisitor::visitConfigMap(MemgraphCypher::ConfigMapContext *ctx) {
std::unordered_map<Expression *, Expression *> map;
for (auto *key_value_pair : ctx->configKeyValuePair()) {
// If the queries are cached, then only the stripped query is parsed, so the actual keys cannot be determined
// here. That means duplicates cannot be checked.
map.insert(key_value_pair->accept(this).as<std::pair<Expression *, Expression *>>());
}
return map;
}
antlrcpp::Any CypherMainVisitor::visitKafkaCreateStream(MemgraphCypher::KafkaCreateStreamContext *ctx) {
auto *stream_query = storage_->Create<StreamQuery>();
stream_query->action_ = StreamQuery::Action::CREATE_STREAM;
@@ -587,6 +606,10 @@ antlrcpp::Any CypherMainVisitor::visitKafkaCreateStream(MemgraphCypher::KafkaCre
MapConfig<true, std::vector<std::string>, Expression *>(memory_, KafkaConfigKey::TOPICS, stream_query->topic_names_);
MapConfig<false, std::string>(memory_, KafkaConfigKey::CONSUMER_GROUP, stream_query->consumer_group_);
MapConfig<false, Expression *>(memory_, KafkaConfigKey::BOOTSTRAP_SERVERS, stream_query->bootstrap_servers_);
MapConfig<false, std::unordered_map<Expression *, Expression *>>(memory_, KafkaConfigKey::CONFIGS,
stream_query->configs_);
MapConfig<false, std::unordered_map<Expression *, Expression *>>(memory_, KafkaConfigKey::CREDENTIALS,
stream_query->credentials_);
MapCommonStreamConfigs(memory_, *stream_query);
@@ -622,18 +645,33 @@ antlrcpp::Any CypherMainVisitor::visitKafkaCreateStreamConfig(MemgraphCypher::Ka
if (ctx->TOPICS()) {
ThrowIfExists(memory_, KafkaConfigKey::TOPICS);
const auto topics_key = static_cast<uint8_t>(KafkaConfigKey::TOPICS);
constexpr auto topics_key = static_cast<uint8_t>(KafkaConfigKey::TOPICS);
GetTopicNames(memory_[topics_key], ctx->topicNames(), *this);
return {};
}
if (ctx->CONSUMER_GROUP()) {
ThrowIfExists(memory_, KafkaConfigKey::CONSUMER_GROUP);
const auto consumer_group_key = static_cast<uint8_t>(KafkaConfigKey::CONSUMER_GROUP);
constexpr auto consumer_group_key = static_cast<uint8_t>(KafkaConfigKey::CONSUMER_GROUP);
memory_[consumer_group_key] = JoinSymbolicNamesWithDotsAndMinus(*this, *ctx->consumerGroup);
return {};
}
if (ctx->CONFIGS()) {
ThrowIfExists(memory_, KafkaConfigKey::CONFIGS);
constexpr auto configs_key = static_cast<uint8_t>(KafkaConfigKey::CONFIGS);
memory_.emplace(configs_key, ctx->configsMap->accept(this).as<std::unordered_map<Expression *, Expression *>>());
return {};
}
if (ctx->CREDENTIALS()) {
ThrowIfExists(memory_, KafkaConfigKey::CREDENTIALS);
constexpr auto credentials_key = static_cast<uint8_t>(KafkaConfigKey::CREDENTIALS);
memory_.emplace(credentials_key,
ctx->credentialsMap->accept(this).as<std::unordered_map<Expression *, Expression *>>());
return {};
}
MG_ASSERT(ctx->BOOTSTRAP_SERVERS());
ThrowIfExists(memory_, KafkaConfigKey::BOOTSTRAP_SERVERS);
if (!ctx->bootstrapServers->StringLiteral()) {

16
src/query/frontend/ast/cypher_main_visitor.hpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -269,6 +269,16 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
*/
antlrcpp::Any visitCreateStream(MemgraphCypher::CreateStreamContext *ctx) override;
/**
* @return StreamQuery*
*/
antlrcpp::Any visitConfigKeyValuePair(MemgraphCypher::ConfigKeyValuePairContext *ctx) override;
/**
* @return StreamQuery*
*/
antlrcpp::Any visitConfigMap(MemgraphCypher::ConfigMapContext *ctx) override;
/**
* @return StreamQuery*
*/
@@ -849,7 +859,9 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
ParsingContext context_;
AstStorage *storage_;
std::unordered_map<uint8_t, std::variant<Expression *, std::string, std::vector<std::string>>> memory_;
std::unordered_map<uint8_t, std::variant<Expression *, std::string, std::vector<std::string>,
std::unordered_map<Expression *, Expression *>>>
memory_;
// Set of identifiers from queries.
std::unordered_set<std::string> users_identifiers;
// Identifiers that user didn't name.

8
src/query/frontend/opencypher/grammar/MemgraphCypher.g4
View File

@@ -35,7 +35,9 @@ memgraphCypherKeyword : cypherKeyword
| COMMIT
| COMMITTED
| CONFIG
| CONFIGS
| CONSUMER_GROUP
| CREDENTIALS
| CSV
| DATA
| DELIMITER
@@ -306,9 +308,15 @@ commonCreateStreamConfig : TRANSFORM transformationName=procedureName
createStream : kafkaCreateStream | pulsarCreateStream ;
configKeyValuePair : literal ':' literal ;
configMap : '{' ( configKeyValuePair ( ',' configKeyValuePair )* )? '}' ;
kafkaCreateStreamConfig : TOPICS topicNames
| CONSUMER_GROUP consumerGroup=symbolicNameWithDotsAndMinus
| BOOTSTRAP_SERVERS bootstrapServers=literal
| CONFIGS configsMap=configMap
| CREDENTIALS credentialsMap=configMap
| commonCreateStreamConfig
;

2
src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4
View File

@@ -40,7 +40,9 @@ CLEAR : C L E A R ;
COMMIT : C O M M I T ;
COMMITTED : C O M M I T T E D ;
CONFIG : C O N F I G ;
CONFIGS : C O N F I G S;
CONSUMER_GROUP : C O N S U M E R UNDERSCORE G R O U P ;
CREDENTIALS : C R E D E N T I A L S ;
CSV : C S V ;
DATA : D A T A ;
DELIMITER : D E L I M I T E R ;

24
src/query/interpreter.cpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -564,10 +564,26 @@ Callback::CallbackFunction GetKafkaCreateCallback(StreamQuery *stream_query, Exp
}
auto common_stream_info = GetCommonStreamInfo(stream_query, evaluator);
const auto get_config_map = [&evaluator](std::unordered_map<Expression *, Expression *> map,
std::string_view map_name) -> std::unordered_map<std::string, std::string> {
std::unordered_map<std::string, std::string> config_map;
for (const auto [key_expr, value_expr] : map) {
const auto key = key_expr->Accept(evaluator);
const auto value = value_expr->Accept(evaluator);
if (!key.IsString() || !value.IsString()) {
throw SemanticException("{} must contain only string keys and values!", map_name);
}
config_map.emplace(key.ValueString(), value.ValueString());
}
return config_map;
};
return [interpreter_context, stream_name = stream_query->stream_name_,
topic_names = EvaluateTopicNames(evaluator, stream_query->topic_names_),
consumer_group = std::move(consumer_group), common_stream_info = std::move(common_stream_info),
bootstrap_servers = std::move(bootstrap), owner = StringPointerToOptional(username)]() mutable {
bootstrap_servers = std::move(bootstrap), owner = StringPointerToOptional(username),
configs = get_config_map(stream_query->configs_, "Configs"),
credentials = get_config_map(stream_query->credentials_, "Credentials")]() mutable {
std::string bootstrap = bootstrap_servers
? std::move(*bootstrap_servers)
: std::string{interpreter_context->config.default_kafka_bootstrap_servers};
@@ -575,7 +591,9 @@ Callback::CallbackFunction GetKafkaCreateCallback(StreamQuery *stream_query, Exp
{.common_info = std::move(common_stream_info),
.topics = std::move(topic_names),
.consumer_group = std::move(consumer_group),
.bootstrap_servers = std::move(bootstrap)},
.bootstrap_servers = std::move(bootstrap),
.configs = std::move(configs),
.credentials = std::move(credentials)},
std::move(owner));
return std::vector<std::vector<TypedValue>>{};

19
src/query/stream/sources.cpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -13,6 +13,8 @@
#include <json/json.hpp>
#include "integrations/constants.hpp"
namespace query::stream {
KafkaStream::KafkaStream(std::string stream_name, StreamInfo stream_info,
ConsumerFunction<integrations::kafka::Message> consumer_function) {
@@ -23,6 +25,8 @@ KafkaStream::KafkaStream(std::string stream_name, StreamInfo stream_info,
.bootstrap_servers = std::move(stream_info.bootstrap_servers),
.batch_interval = stream_info.common_info.batch_interval,
.batch_size = stream_info.common_info.batch_size,
.public_configs = std::move(stream_info.configs),
.private_configs = std::move(stream_info.credentials),
};
consumer_.emplace(std::move(consumer_info), std::move(consumer_function));
};
@@ -34,7 +38,9 @@ KafkaStream::StreamInfo KafkaStream::Info(std::string transformation_name) const
.transformation_name = std::move(transformation_name)},
.topics = info.topics,
.consumer_group = info.consumer_group,
.bootstrap_servers = info.bootstrap_servers};
.bootstrap_servers = info.bootstrap_servers,
.configs = info.public_configs,
.credentials = info.private_configs};
}
void KafkaStream::Start() { consumer_->Start(); }
@@ -54,6 +60,10 @@ namespace {
const std::string kTopicsKey{"topics"};
const std::string kConsumerGroupKey{"consumer_group"};
const std::string kBoostrapServers{"bootstrap_servers"};
const std::string kConfigs{"configs"};
const std::string kCredentials{"credentials"};
const std::unordered_map<std::string, std::string> kDefaultConfigsMap;
} // namespace
void to_json(nlohmann::json &data, KafkaStream::StreamInfo &&info) {
@@ -61,6 +71,8 @@ void to_json(nlohmann::json &data, KafkaStream::StreamInfo &&info) {
data[kTopicsKey] = std::move(info.topics);
data[kConsumerGroupKey] = info.consumer_group;
data[kBoostrapServers] = std::move(info.bootstrap_servers);
data[kConfigs] = std::move(info.configs);
data[kCredentials] = std::move(info.credentials);
}
void from_json(const nlohmann::json &data, KafkaStream::StreamInfo &info) {
@@ -68,6 +80,9 @@ void from_json(const nlohmann::json &data, KafkaStream::StreamInfo &info) {
data.at(kTopicsKey).get_to(info.topics);
data.at(kConsumerGroupKey).get_to(info.consumer_group);
data.at(kBoostrapServers).get_to(info.bootstrap_servers);
// These values might not be present in the persisted JSON object
info.configs = data.value(kConfigs, kDefaultConfigsMap);
info.credentials = data.value(kCredentials, kDefaultConfigsMap);
}
PulsarStream::PulsarStream(std::string stream_name, StreamInfo stream_info,

4
src/query/stream/sources.hpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -24,6 +24,8 @@ struct KafkaStream {
std::vector<std::string> topics;
std::string consumer_group;
std::string bootstrap_servers;
std::unordered_map<std::string, std::string> configs;
std::unordered_map<std::string, std::string> credentials;
};
using Message = integrations::kafka::Message;

80
src/query/stream/streams.cpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -18,6 +18,7 @@
#include <spdlog/spdlog.h>
#include <json/json.hpp>
#include "integrations/constants.hpp"
#include "mg_procedure.h"
#include "query/db_accessor.hpp"
#include "query/discard_value_stream.hpp"
@@ -29,6 +30,7 @@
#include "query/stream/sources.hpp"
#include "query/typed_value.hpp"
#include "utils/event_counter.hpp"
#include "utils/logging.hpp"
#include "utils/memory.hpp"
#include "utils/on_scope_exit.hpp"
#include "utils/pmr/string.hpp"
@@ -208,10 +210,12 @@ void Streams::RegisterKafkaProcedures() {
constexpr std::string_view consumer_group_result_name = "consumer_group";
constexpr std::string_view topics_result_name = "topics";
constexpr std::string_view bootstrap_servers_result_name = "bootstrap_servers";
constexpr std::string_view configs_result_name = "configs";
constexpr std::string_view credentials_result_name = "credentials";
auto get_stream_info = [this, proc_name, consumer_group_result_name, topics_result_name,
bootstrap_servers_result_name](mgp_list *args, mgp_graph * /*graph*/, mgp_result *result,
mgp_memory *memory) {
bootstrap_servers_result_name, configs_result_name, credentials_result_name](
mgp_list *args, mgp_graph * /*graph*/, mgp_result *result, mgp_memory *memory) {
auto *arg_stream_name = procedure::Call<mgp_value *>(mgp_list_at, args, 0);
const auto *stream_name = procedure::Call<const char *>(mgp_value_get_string, arg_stream_name);
auto lock_ptr = streams_.Lock();
@@ -259,13 +263,12 @@ void Streams::RegisterKafkaProcedures() {
procedure::MgpUniquePtr<mgp_value> topics_value{nullptr, mgp_value_destroy};
{
const auto success = procedure::TryOrSetError(
[&] {
return procedure::CreateMgpObject(topics_value, mgp_value_make_list, topic_names.release());
},
[&] { return procedure::CreateMgpObject(topics_value, mgp_value_make_list, topic_names.get()); },
result);
if (!success) {
return;
}
static_cast<void>(topic_names.release());
}
const auto bootstrap_servers_value =
@@ -274,6 +277,57 @@ void Streams::RegisterKafkaProcedures() {
return;
}
const auto convert_config_map =
[result, memory](const std::unordered_map<std::string, std::string> &configs_to_convert)
-> procedure::MgpUniquePtr<mgp_value> {
procedure::MgpUniquePtr<mgp_value> configs_value{nullptr, mgp_value_destroy};
procedure::MgpUniquePtr<mgp_map> configs{nullptr, mgp_map_destroy};
{
const auto success = procedure::TryOrSetError(
[&] { return procedure::CreateMgpObject(configs, mgp_map_make_empty, memory); }, result);
if (!success) {
return configs_value;
}
}
for (const auto &[key, value] : configs_to_convert) {
auto value_value = procedure::GetStringValueOrSetError(value.c_str(), memory, result);
if (!value_value) {
return configs_value;
}
configs->items.emplace(key, std::move(*value_value));
}
{
const auto success = procedure::TryOrSetError(
[&] { return procedure::CreateMgpObject(configs_value, mgp_value_make_map, configs.get()); },
result);
if (!success) {
return configs_value;
}
static_cast<void>(configs.release());
}
return configs_value;
};
const auto configs_value = convert_config_map(info.configs);
if (configs_value == nullptr) {
return;
}
using CredentialsType = decltype(KafkaStream::StreamInfo::credentials);
CredentialsType reducted_credentials;
std::transform(info.credentials.begin(), info.credentials.end(),
std::inserter(reducted_credentials, reducted_credentials.end()),
[](const auto &pair) -> CredentialsType::value_type {
return {pair.first, integrations::kReducted};
});
const auto credentials_value = convert_config_map(reducted_credentials);
if (credentials_value == nullptr) {
return;
}
if (!procedure::InsertResultOrSetError(result, record, consumer_group_result_name.data(),
consumer_group_value.get())) {
return;
@@ -287,6 +341,16 @@ void Streams::RegisterKafkaProcedures() {
bootstrap_servers_value.get())) {
return;
}
if (!procedure::InsertResultOrSetError(result, record, configs_result_name.data(),
configs_value.get())) {
return;
}
if (!procedure::InsertResultOrSetError(result, record, credentials_result_name.data(),
credentials_value.get())) {
return;
}
},
[proc_name](auto && /*other*/) {
throw QueryRuntimeException("'{}' can be only used for Kafka stream sources", proc_name);
@@ -305,6 +369,10 @@ void Streams::RegisterKafkaProcedures() {
MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, bootstrap_servers_result_name.data(),
procedure::Call<mgp_type *>(mgp_type_string)) == MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, configs_result_name.data(), procedure::Call<mgp_type *>(mgp_type_map)) ==
MGP_ERROR_NO_ERROR);
MG_ASSERT(mgp_proc_add_result(&proc, credentials_result_name.data(), procedure::Call<mgp_type *>(mgp_type_map)) ==
MGP_ERROR_NO_ERROR);
procedure::gModuleRegistry.RegisterMgProcedure(proc_name, std::move(proc));
}

5
src/query/stream/streams.hpp
View File

@@ -1,4 +1,4 @@
// Copyright 2021 Memgraph Ltd.
// Copyright 2022 Memgraph Ltd.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -31,6 +31,7 @@
#include "utils/rw_lock.hpp"
#include "utils/synchronized.hpp"
class StreamsTest;
namespace query {
struct InterpreterContext;
@@ -73,6 +74,8 @@ using TransformationResult = std::vector<std::vector<TypedValue>>;
///
/// This class is responsible for all query supported actions to happen.
class Streams final {
friend StreamsTest;
public:
/// Initializes the streams.
///