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

Fix stream creation failure handling

Browse Source 
Summary: Stream was created although transform script couldn't be downloaded.

Reviewers: mferencevic, msantl, teon.banek

Reviewed By: mferencevic, msantl, teon.banek

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D1683
This commit is contained in:
Marin Tomic 2018-10-19 13:29:43 +02:00
parent 022a7887b4
commit 8c2e87b8a2
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/integrations/kafka/streams.cpp
View File

@ -134,11 +134,6 @@ void Streams::Create(const StreamInfo &info, bool download_transform_script) {
if (consumers_.find(info.stream_name) != consumers_.end()) if (consumers_.find(info.stream_name) != consumers_.end())
throw StreamExistsException(info.stream_name); throw StreamExistsException(info.stream_name);
// Store stream_info in metadata_store_.
if (!metadata_store_.Put(info.stream_name, Serialize(info).dump())) {
throw StreamMetadataCouldNotBeStored(info.stream_name);
}
// Make sure transform directory exists or we can create it. // Make sure transform directory exists or we can create it.
if (!utils::EnsureDir(GetTransformScriptDir())) { if (!utils::EnsureDir(GetTransformScriptDir())) {
throw TransformScriptCouldNotBeCreatedException(info.stream_name); throw TransformScriptCouldNotBeCreatedException(info.stream_name);
@ -152,6 +147,11 @@ void Streams::Create(const StreamInfo &info, bool download_transform_script) {
throw TransformScriptDownloadException(info.transform_uri); throw TransformScriptDownloadException(info.transform_uri);
} }
// Store stream_info in metadata_store_.
if (!metadata_store_.Put(info.stream_name, Serialize(info).dump())) {
throw StreamMetadataCouldNotBeStored(info.stream_name);
}
try { try {
consumers_.emplace( consumers_.emplace(
std::piecewise_construct, std::forward_as_tuple(info.stream_name), std::piecewise_construct, std::forward_as_tuple(info.stream_name),