MINIFICPP-1962 Implement communication between process group through ports
Closes #1451
Signed-off-by: Marton Szasz <szaszm@apache.org>
diff --git a/extensions/standard-processors/tests/CMakeLists.txt b/extensions/standard-processors/tests/CMakeLists.txt
index 01e162d..784de18 100644
--- a/extensions/standard-processors/tests/CMakeLists.txt
+++ b/extensions/standard-processors/tests/CMakeLists.txt
@@ -97,8 +97,8 @@
endif()
add_test(NAME TailFileTest COMMAND TailFileTest "${TEST_RESOURCES}/TestTailFile.yml" "${TEST_RESOURCES}/")
-
add_test(NAME TailFileCronTest COMMAND TailFileTest "${TEST_RESOURCES}/TestTailFileCron.yml" "${TEST_RESOURCES}/")
+add_test(NAME ProcessGroupTest COMMAND ProcessGroupTest "${TEST_RESOURCES}/TestProcessGroup.yml")
FOREACH(resourcefile ${RESOURCE_APPS})
get_filename_component(resourcefilename "${resourcefile}" NAME_WE)
diff --git a/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp b/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp
new file mode 100644
index 0000000..5e83586
--- /dev/null
+++ b/extensions/standard-processors/tests/integration/ProcessGroupTest.cpp
@@ -0,0 +1,63 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#undef NDEBUG
+#include <cassert>
+#include <string>
+
+#include "core/logging/Logger.h"
+#include "FlowController.h"
+#include "TestBase.h"
+#include "Catch.h"
+#include "processors/GenerateFlowFile.h"
+#include "processors/LogAttribute.h"
+#include "processors/UpdateAttribute.h"
+#include "integration/IntegrationBase.h"
+#include "utils/IntegrationTestUtils.h"
+
+using namespace std::literals::chrono_literals;
+
+class ProcessGroupTestHarness : public IntegrationBase {
+ public:
+ ProcessGroupTestHarness() : IntegrationBase(2s) {
+ }
+
+ void testSetup() override {
+ LogTestController::getInstance().setInfo<minifi::processors::LogAttribute>();
+ LogTestController::getInstance().setTrace<minifi::processors::GenerateFlowFile>();
+ LogTestController::getInstance().setTrace<minifi::processors::UpdateAttribute>();
+ }
+
+ void runAssertions() override {
+ using org::apache::nifi::minifi::utils::verifyLogLinePresenceInPollTime;
+ assert(verifyLogLinePresenceInPollTime(std::chrono::milliseconds(wait_time_),
+ "key:test_attribute value:success"));
+ }
+};
+
+int main(int argc, char **argv) {
+ std::string test_file_location;
+ if (argc > 1) {
+ test_file_location = argv[1];
+ }
+
+ ProcessGroupTestHarness harness;
+ harness.run(test_file_location);
+
+ return 0;
+}
diff --git a/extensions/standard-processors/tests/unit/ProcessGroupTestUtils.h b/extensions/standard-processors/tests/unit/ProcessGroupTestUtils.h
index 58a0dac..bb9d167 100644
--- a/extensions/standard-processors/tests/unit/ProcessGroupTestUtils.h
+++ b/extensions/standard-processors/tests/unit/ProcessGroupTestUtils.h
@@ -60,9 +60,21 @@
}
};
+enum class ConnectionFailure {
+ UNRESOLVED_SOURCE,
+ UNRESOLVED_DESTINATION,
+ INPUT_CANNOT_BE_SOURCE,
+ OUTPUT_CANNOT_BE_DESTINATION,
+ INPUT_CANNOT_BE_DESTINATION,
+ OUTPUT_CANNOT_BE_SOURCE
+};
+
struct Proc {
+ Proc(std::string id, std::string name, const std::optional<ConnectionFailure>& failure = std::nullopt)
+ : id(std::move(id)), name(std::move(name)), failure(failure) {}
std::string id;
std::string name;
+ std::optional<ConnectionFailure> failure;
Lines serialize() const {
return {{
@@ -73,17 +85,33 @@
}
};
-struct UnresolvedProc {
- explicit UnresolvedProc(std::string id): id(std::move(id)) {}
+template<typename Tag>
+struct Port {
+ Port(std::string id, std::string name, const std::optional<ConnectionFailure>& failure = std::nullopt)
+ : id(std::move(id)), name(std::move(name)), failure(failure) {}
std::string id;
+ std::string name;
+ std::optional<ConnectionFailure> failure;
+
+ Lines serialize() const {
+ return {{
+ "- id: " + id,
+ " name: " + name
+ }};
+ }
};
+using InputPort = Port<struct InputTag>;
+using OutputPort = Port<struct OutputTag>;
+
struct MaybeProc {
- MaybeProc(const Proc& proc): id(proc.id), name(proc.name) {} // NOLINT
- MaybeProc(const UnresolvedProc& proc) : id(proc.id) {} // NOLINT
+ MaybeProc(const Proc& proc) : id(proc.id), name(proc.name), failure(proc.failure) {} // NOLINT(runtime/explicit)
+ MaybeProc(const InputPort& port) : id(port.id), name(port.name), failure(port.failure) {} // NOLINT(runtime/explicit)
+ MaybeProc(const OutputPort& port) : id(port.id), name(port.name), failure(port.failure) {} // NOLINT(runtime/explicit)
std::string id;
- std::optional<std::string> name;
+ std::string name;
+ std::optional<ConnectionFailure> failure;
};
struct Conn {
@@ -139,6 +167,14 @@
rpgs_ = std::move(rpgs);
return *this;
}
+ Group& With(std::vector<InputPort> input_ports) {
+ input_ports_ = std::move(input_ports);
+ return *this;
+ }
+ Group& With(std::vector<OutputPort> output_ports) {
+ output_ports_ = std::move(output_ports);
+ return *this;
+ }
Lines serialize(bool is_root = true) const {
Lines body;
if (processors_.empty()) {
@@ -169,6 +205,22 @@
body.append(subgroup.serialize(false).indentAll());
}
}
+ if (input_ports_.empty()) {
+ body.emplace_back("Input Ports: []");
+ } else {
+ body.emplace_back("Input Ports:");
+ for (const auto& port : input_ports_) {
+ body.append(port.serialize().indentAll());
+ }
+ }
+ if (output_ports_.empty()) {
+ body.emplace_back("Output Ports: []");
+ } else {
+ body.emplace_back("Output Ports:");
+ for (const auto& port : output_ports_) {
+ body.append(port.serialize().indentAll());
+ }
+ }
Lines lines;
if (is_root) {
lines.emplace_back("Flow Controller:");
@@ -186,12 +238,15 @@
std::vector<Proc> processors_;
std::vector<Group> subgroups_;
std::vector<RPG> rpgs_;
+ std::vector<InputPort> input_ports_;
+ std::vector<OutputPort> output_ports_;
};
struct ProcessGroupTestAccessor {
FIELD_ACCESSOR(processors_)
FIELD_ACCESSOR(connections_)
FIELD_ACCESSOR(child_process_groups_)
+ FIELD_ACCESSOR(ports_)
};
template<typename T, typename = void>
@@ -222,6 +277,54 @@
return nullptr;
}
+void assertFailure(const Conn& expected, ConnectionFailure failure) {
+ auto assertMessage = [](const std::string& message) {
+ REQUIRE(utils::verifyLogLinePresenceInPollTime(std::chrono::seconds{1}, message));
+ };
+
+ switch (failure) {
+ case ConnectionFailure::UNRESOLVED_DESTINATION: {
+ assertMessage("Cannot find the destination processor with id '" + expected.destination.id + "' for the connection [name = '" + expected.name + "'");
+ break;
+ }
+ case ConnectionFailure::UNRESOLVED_SOURCE: {
+ assertMessage("Cannot find the source processor with id '" + expected.source.id + "' for the connection [name = '" + expected.name + "'");
+ break;
+ }
+ case ConnectionFailure::INPUT_CANNOT_BE_SOURCE: {
+ assertMessage("Input port [id = '" + expected.source.id + "'] cannot be a source outside the process group in the connection [name = '" + expected.name + "'");
+ break;
+ }
+ case ConnectionFailure::OUTPUT_CANNOT_BE_DESTINATION: {
+ assertMessage("Output port [id = '" + expected.destination.id + "'] cannot be a destination outside the process group in the connection [name = '" + expected.name + "'");
+ break;
+ }
+ case ConnectionFailure::INPUT_CANNOT_BE_DESTINATION: {
+ assertMessage("Input port [id = '" + expected.destination.id + "'] cannot be a destination inside the process group in the connection [name = '" + expected.name + "'");
+ break;
+ }
+ case ConnectionFailure::OUTPUT_CANNOT_BE_SOURCE: {
+ assertMessage("Output port [id = '" + expected.source.id + "'] cannot be a source inside the process group in the connection [name = '" + expected.name + "'");
+ break;
+ }
+ }
+}
+
+void verifyConnectionNode(minifi::Connection* conn, const Conn& expected) {
+ if (expected.source.failure) {
+ REQUIRE(conn->getSource() == nullptr);
+ assertFailure(expected, *expected.source.failure);
+ } else {
+ REQUIRE(conn->getSource()->getName() == expected.source.name);
+ }
+ if (expected.destination.failure) {
+ REQUIRE(conn->getDestination() == nullptr);
+ assertFailure(expected, *expected.destination.failure);
+ } else {
+ REQUIRE(conn->getDestination()->getName() == expected.destination.name);
+ }
+}
+
void verifyProcessGroup(core::ProcessGroup& group, const Group& pattern) {
// verify name
REQUIRE(group.getName() == pattern.name_);
@@ -231,33 +334,34 @@
for (auto& expected : pattern.connections_) {
auto conn = findByName(connections, expected.name);
REQUIRE(conn);
- if (!expected.source.name) {
- REQUIRE(conn->getSource() == nullptr);
- REQUIRE(utils::verifyLogLinePresenceInPollTime(
- std::chrono::seconds{1},
- "Cannot find the source processor with id '" + expected.source.id
- + "' for the connection [name = '" + expected.name + "'"));
- } else {
- REQUIRE(conn->getSource()->getName() == expected.source.name);
- }
- if (!expected.destination.name) {
- REQUIRE(conn->getDestination() == nullptr);
- REQUIRE(utils::verifyLogLinePresenceInPollTime(
- std::chrono::seconds{1},
- "Cannot find the destination processor with id '" + expected.destination.id
- + "' for the connection [name = '" + expected.name + "'"));
- } else {
- REQUIRE(conn->getDestination()->getName() == expected.destination.name);
- }
+ verifyConnectionNode(conn, expected);
}
- // verify processors
+ // verify processors and ports
const auto& processors = ProcessGroupTestAccessor::get_processors_(group);
- REQUIRE(processors.size() == pattern.processors_.size());
+ REQUIRE(processors.size() == pattern.processors_.size() + pattern.input_ports_.size() + pattern.output_ports_.size());
for (auto& expected : pattern.processors_) {
REQUIRE(findByName(processors, expected.name));
}
+ for (auto& expected : pattern.input_ports_) {
+ REQUIRE(findByName(processors, expected.name));
+ }
+
+ for (auto& expected : pattern.output_ports_) {
+ REQUIRE(findByName(processors, expected.name));
+ }
+
+ const auto& ports = ProcessGroupTestAccessor::get_ports_(group);
+ REQUIRE(ports.size() == pattern.input_ports_.size() + pattern.output_ports_.size());
+ for (auto& expected : pattern.input_ports_) {
+ REQUIRE(findByName(ports, expected.name));
+ }
+
+ for (auto& expected : pattern.output_ports_) {
+ REQUIRE(findByName(ports, expected.name));
+ }
+
std::set<core::ProcessGroup*> simple_subgroups;
std::set<core::ProcessGroup*> rpg_subgroups;
for (auto& subgroup : ProcessGroupTestAccessor::get_child_process_groups_(group)) {
diff --git a/extensions/standard-processors/tests/unit/YamlProcessGroupParserTests.cpp b/extensions/standard-processors/tests/unit/YamlProcessGroupParserTests.cpp
index 52e264d..0f58795 100644
--- a/extensions/standard-processors/tests/unit/YamlProcessGroupParserTests.cpp
+++ b/extensions/standard-processors/tests/unit/YamlProcessGroupParserTests.cpp
@@ -24,7 +24,7 @@
static core::YamlConfiguration config(nullptr, nullptr, nullptr, nullptr, std::make_shared<minifi::Configure>());
-TEST_CASE("Root process group is correctly parsed", "[YamlProcessGroupParser1]") {
+TEST_CASE("Root process group is correctly parsed", "[YamlProcessGroupParser]") {
auto pattern = Group("root")
.With({
Conn{"Conn1",
@@ -48,7 +48,7 @@
verifyProcessGroup(*root, pattern);
}
-TEST_CASE("Nested process group is correctly parsed", "[YamlProcessGroupParser2]") {
+TEST_CASE("Nested process group is correctly parsed", "[YamlProcessGroupParser]") {
auto pattern = Group("root")
.With({Conn{"Conn1",
Proc{"00000000-0000-0000-0000-000000000001", "Proc1"},
@@ -71,7 +71,7 @@
verifyProcessGroup(*root, pattern);
}
-TEST_CASE("Cannot connect processors from different groups", "[YamlProcessGroupParser3]") {
+TEST_CASE("Cannot connect processors from different groups", "[YamlProcessGroupParser]") {
TestController controller;
LogTestController::getInstance().setTrace<core::YamlConfiguration>();
Proc Proc1{"00000000-0000-0000-0000-000000000001", "Proc1"};
@@ -102,18 +102,18 @@
}
SECTION("Connecting processors in their child/parent group") {
- Conn1.source = UnresolvedProc{Child1_Proc1.id};
- Conn1.destination = UnresolvedProc{Child1_Port1.id};
+ Conn1.source = Proc{Child1_Proc1.id, Child1_Proc1.name, ConnectionFailure::UNRESOLVED_SOURCE};
+ Conn1.destination = Proc{Child1_Port1.id, Child1_Port1.name, ConnectionFailure::UNRESOLVED_DESTINATION};
- Child1_Conn1.source = UnresolvedProc{Proc1.id};
- Child1_Conn1.destination = UnresolvedProc{Port1.id};
+ Child1_Conn1.source = Proc{Proc1.id, Proc1.name, ConnectionFailure::UNRESOLVED_SOURCE};
+ Child1_Conn1.destination = Proc{Port1.id, Proc1.name, ConnectionFailure::UNRESOLVED_DESTINATION};
}
SECTION("Connecting processors between their own and their child/parent group") {
Conn1.source = Proc1;
- Conn1.destination = UnresolvedProc{Child1_Port1.id};
+ Conn1.destination = Proc{Child1_Port1.id, Child1_Port1.name, ConnectionFailure::UNRESOLVED_DESTINATION};
- Child1_Conn1.source = UnresolvedProc{Port1.id};
+ Child1_Conn1.source = Proc{Port1.id, Port1.name, ConnectionFailure::UNRESOLVED_SOURCE};
Child1_Conn1.destination = Child1_Proc1;
}
@@ -121,11 +121,171 @@
Conn1.source = Proc1;
Conn1.destination = Port1;
- Child1_Conn1.source = UnresolvedProc{Child2_Proc1.id};
- Child1_Conn1.destination = UnresolvedProc{Child2_Port1.id};
+ Child1_Conn1.source = Proc{Child2_Proc1.id, Child2_Proc1.name, ConnectionFailure::UNRESOLVED_SOURCE};
+ Child1_Conn1.destination = Proc{Child2_Port1.id, Child2_Port1.name, ConnectionFailure::UNRESOLVED_DESTINATION};
}
auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
verifyProcessGroup(*root, pattern);
}
+
+TEST_CASE("Processor can communicate with child process group's input port", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ Proc{"00000000-0000-0000-0000-000000000001", "Proc1"},
+ InputPort{"00000000-0000-0000-0000-000000000002", "Port1"}}})
+ .With({Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}})
+ .With({
+ Group("Child1")
+ .With({InputPort{"00000000-0000-0000-0000-000000000002", "Port1"}})
+ });
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Child process group can provide input for root processor through output port", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ OutputPort{"00000000-0000-0000-0000-000000000002", "Port1"},
+ Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}}})
+ .With({Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}})
+ .With({
+ Group("Child1")
+ .With({OutputPort{"00000000-0000-0000-0000-000000000002", "Port1"}})
+ });
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Child process groups can communicate through ports", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ OutputPort{"00000000-0000-0000-0000-000000000002", "Port1"},
+ InputPort{"00000000-0000-0000-0000-000000000003", "Port2"}}})
+ .With({Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}})
+ .With({
+ Group("Child1")
+ .With({OutputPort{"00000000-0000-0000-0000-000000000002", "Port1"}}),
+ Group("Child2")
+ .With({InputPort{"00000000-0000-0000-0000-000000000003", "Port2"}})
+ });
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Processor cannot communicate with child's nested process group", "[YamlProcessGroupParser]") {
+ Proc Proc1{"00000000-0000-0000-0000-000000000001", "Proc1"};
+ OutputPort Port1{"00000000-0000-0000-0000-000000000002", "Port1"};
+ InputPort Port2{"00000000-0000-0000-0000-000000000003", "Port2", ConnectionFailure::UNRESOLVED_DESTINATION};
+
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ Proc1,
+ Port2}})
+ .With({Proc1})
+ .With({
+ Group("Child1")
+ .With({Port1})
+ .With({Group("Child2")
+ .With({Port2})})
+ });
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Input port can be a connection's source and the output port can be a destination inside the process group", "[YamlProcessGroupParser7]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ InputPort{"00000000-0000-0000-0000-000000000001", "Port1"},
+ OutputPort{"00000000-0000-0000-0000-000000000002", "Port2"}}})
+ .With({InputPort{"00000000-0000-0000-0000-000000000001", "Port1"}})
+ .With({OutputPort{"00000000-0000-0000-0000-000000000002", "Port2"}});
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Input port cannot be a connection's destination inside the process group", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ Proc{"00000000-0000-0000-0000-000000000002", "Proc1"},
+ InputPort{"00000000-0000-0000-0000-000000000001", "Port1", ConnectionFailure::INPUT_CANNOT_BE_DESTINATION}}})
+ .With({InputPort{"00000000-0000-0000-0000-000000000001", "Port1"}})
+ .With({Proc{"00000000-0000-0000-0000-000000000002", "Proc1"}});
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Output port cannot be a connection's source inside the process group", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ OutputPort{"00000000-0000-0000-0000-000000000001", "Port1", ConnectionFailure::OUTPUT_CANNOT_BE_SOURCE},
+ Proc{"00000000-0000-0000-0000-000000000002", "Proc1"}}})
+ .With({OutputPort{"00000000-0000-0000-0000-000000000001", "Port1"}})
+ .With({Proc{"00000000-0000-0000-0000-000000000002", "Proc1"}});
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Input port can be a connection's source and the output port can be a destination inside the process group through processor", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ InputPort{"00000000-0000-0000-0000-000000000001", "Port1"},
+ Proc{"00000000-0000-0000-0000-000000000003", "Proc1"}},
+ Conn{"Conn2",
+ Proc{"00000000-0000-0000-0000-000000000003", "Proc1"},
+ OutputPort{"00000000-0000-0000-0000-000000000002", "Port2"}}})
+ .With({Proc{"00000000-0000-0000-0000-000000000003", "Proc1"}})
+ .With({InputPort{"00000000-0000-0000-0000-000000000001", "Port1"}})
+ .With({OutputPort{"00000000-0000-0000-0000-000000000002", "Port2"}});
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Processor cannot set connection's destination to child process group's output port", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ Proc{"00000000-0000-0000-0000-000000000001", "Proc1"},
+ OutputPort{"00000000-0000-0000-0000-000000000002", "Port1", ConnectionFailure::OUTPUT_CANNOT_BE_DESTINATION}}})
+ .With({Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}})
+ .With({
+ Group("Child1")
+ .With({OutputPort{"00000000-0000-0000-0000-000000000002", "Port1"}})
+ });
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
+
+TEST_CASE("Processor cannot set connection's source to child process group's input port", "[YamlProcessGroupParser]") {
+ auto pattern = Group("root")
+ .With({Conn{"Conn1",
+ InputPort{"00000000-0000-0000-0000-000000000002", "Port1", ConnectionFailure::INPUT_CANNOT_BE_SOURCE},
+ Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}}})
+ .With({Proc{"00000000-0000-0000-0000-000000000001", "Proc1"}})
+ .With({
+ Group("Child1")
+ .With({InputPort{"00000000-0000-0000-0000-000000000002", "Port1"}})
+ });
+
+ auto root = config.getRootFromPayload(pattern.serialize().join("\n"));
+
+ verifyProcessGroup(*root, pattern);
+}
diff --git a/libminifi/include/core/Funnel.h b/libminifi/include/ForwardingNode.h
similarity index 68%
rename from libminifi/include/core/Funnel.h
rename to libminifi/include/ForwardingNode.h
index 8c8e08c..d492d34 100644
--- a/libminifi/include/core/Funnel.h
+++ b/libminifi/include/ForwardingNode.h
@@ -17,35 +17,35 @@
*/
#pragma once
-#include <memory>
#include <string>
+#include <memory>
#include <utility>
-#include "logging/LoggerFactory.h"
-#include "Processor.h"
+#include "core/logging/LoggerFactory.h"
+#include "core/Processor.h"
-namespace org::apache::nifi::minifi::core {
+namespace org::apache::nifi::minifi {
-class Funnel final : public Processor {
+class ForwardingNode : public core::Processor {
public:
- Funnel(std::string name, const utils::Identifier& uuid) : Processor(std::move(name), uuid), logger_(logging::LoggerFactory<Funnel>::getLogger()) {}
- explicit Funnel(std::string name) : Processor(std::move(name)), logger_(logging::LoggerFactory<Funnel>::getLogger()) {}
+ ForwardingNode(std::string name, const utils::Identifier& uuid, std::shared_ptr<core::logging::Logger> logger) : Processor(std::move(name), uuid), logger_(std::move(logger)) {
+ strategy_ = core::SchedulingStrategy::EVENT_DRIVEN;
+ }
+ ForwardingNode(std::string name, std::shared_ptr<core::logging::Logger> logger) : Processor(std::move(name)), logger_(std::move(logger)) {}
static auto properties() { return std::array<core::Property, 0>{}; }
MINIFIAPI static const core::Relationship Success;
static auto relationships() { return std::array{Success}; }
MINIFIAPI static constexpr bool SupportsDynamicProperties = false;
MINIFIAPI static constexpr bool SupportsDynamicRelationships = false;
- MINIFIAPI static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_REQUIRED;
MINIFIAPI static constexpr bool IsSingleThreaded = false;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
void initialize() override;
void onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) override;
private:
- std::shared_ptr<logging::Logger> logger_;
+ std::shared_ptr<core::logging::Logger> logger_;
};
-} // namespace org::apache::nifi::minifi::core
+} // namespace org::apache::nifi::minifi
diff --git a/libminifi/include/Funnel.h b/libminifi/include/Funnel.h
new file mode 100644
index 0000000..e17689f
--- /dev/null
+++ b/libminifi/include/Funnel.h
@@ -0,0 +1,36 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <string>
+#include <utility>
+
+#include "ForwardingNode.h"
+
+namespace org::apache::nifi::minifi {
+
+class Funnel final : public ForwardingNode {
+ public:
+ Funnel(std::string name, const utils::Identifier& uuid) : ForwardingNode(std::move(name), uuid, core::logging::LoggerFactory<Funnel>::getLogger()) {}
+ explicit Funnel(std::string name) : ForwardingNode(std::move(name), core::logging::LoggerFactory<Funnel>::getLogger()) {}
+
+ MINIFIAPI static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_REQUIRED;
+ ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
+};
+
+} // namespace org::apache::nifi::minifi
diff --git a/libminifi/include/Port.h b/libminifi/include/Port.h
new file mode 100644
index 0000000..65c358b
--- /dev/null
+++ b/libminifi/include/Port.h
@@ -0,0 +1,47 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <string>
+#include <utility>
+
+#include "ForwardingNode.h"
+
+namespace org::apache::nifi::minifi {
+
+enum class PortType {
+ INPUT,
+ OUTPUT
+};
+
+class Port final : public ForwardingNode {
+ public:
+ Port(std::string name, const utils::Identifier& uuid, PortType port_type) : ForwardingNode(std::move(name), uuid, core::logging::LoggerFactory<Port>::getLogger()), port_type_(port_type) {}
+ Port(std::string name, PortType port_type) : ForwardingNode(std::move(name), core::logging::LoggerFactory<Port>::getLogger()), port_type_(port_type) {}
+ PortType getPortType() const {
+ return port_type_;
+ }
+
+ MINIFIAPI static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_ALLOWED;
+ ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
+
+ private:
+ PortType port_type_;
+};
+
+} // namespace org::apache::nifi::minifi
diff --git a/libminifi/include/core/ProcessGroup.h b/libminifi/include/core/ProcessGroup.h
index 76a0c92..96dd684 100644
--- a/libminifi/include/core/ProcessGroup.h
+++ b/libminifi/include/core/ProcessGroup.h
@@ -27,19 +27,21 @@
#include <algorithm>
#include <set>
#include <utility>
+#include <tuple>
#include "Processor.h"
-#include "Funnel.h"
#include "Exception.h"
#include "TimerDrivenSchedulingAgent.h"
#include "EventDrivenSchedulingAgent.h"
#include "CronDrivenSchedulingAgent.h"
+#include "Port.h"
#include "core/logging/Logger.h"
#include "controller/ControllerServiceNode.h"
#include "controller/ControllerServiceMap.h"
#include "utils/Id.h"
#include "utils/BaseHTTPClient.h"
#include "utils/CallBackTimer.h"
+#include "range/v3/algorithm/find_if.hpp"
struct ProcessGroupTestAccessor;
@@ -164,18 +166,22 @@
std::lock_guard<std::recursive_mutex> lock(mutex_);
return parent_process_group_;
}
- // Add processor
- void addProcessor(std::unique_ptr<Processor> processor);
- // Add child processor group
+ [[maybe_unused]] std::tuple<Processor*, bool> addProcessor(std::unique_ptr<Processor> processor);
+ void addPort(std::unique_ptr<Port> port);
void addProcessGroup(std::unique_ptr<ProcessGroup> child);
- // ! Add connections
void addConnection(std::unique_ptr<Connection> connection);
- // Generic find
+ const std::set<Port*>& getPorts() const {
+ return ports_;
+ }
+
+ Port* findPortById(const utils::Identifier& uuid) const;
+ Port* findChildPortById(const utils::Identifier& uuid) const;
+
template <typename Fun>
Processor* findProcessor(Fun condition, Traverse traverse) const {
std::lock_guard<std::recursive_mutex> lock(mutex_);
- const auto found = std::find_if(processors_.cbegin(), processors_.cend(), condition);
- if (found != processors_.cend()) {
+ const auto found = ranges::find_if(processors_, condition);
+ if (found != ranges::end(processors_)) {
return found->get();
}
for (const auto& processGroup : child_process_groups_) {
@@ -231,9 +237,10 @@
int config_version_;
// Process Group Type
const ProcessGroupType type_;
- // Processors (ProcessNode) inside this process group which include Input/Output Port, Remote Process Group input/Output port
+ // Processors (ProcessNode) inside this process group which include Remote Process Group input/Output port
std::set<std::unique_ptr<Processor>> processors_;
std::set<Processor*> failed_processors_;
+ std::set<Port*> ports_;
std::set<std::unique_ptr<ProcessGroup>> child_process_groups_;
// Connections between the processor inside the group;
std::set<std::unique_ptr<Connection>> connections_;
@@ -259,6 +266,8 @@
core::controller::ControllerServiceMap controller_service_map_;
private:
+ static Port* findPortById(const std::set<Port*>& ports, const utils::Identifier& uuid);
+
// Mutex for protection
mutable std::recursive_mutex mutex_;
// Logger
diff --git a/libminifi/include/core/yaml/YamlConfiguration.h b/libminifi/include/core/yaml/YamlConfiguration.h
index 18ff1ee..bf7f280 100644
--- a/libminifi/include/core/yaml/YamlConfiguration.h
+++ b/libminifi/include/core/yaml/YamlConfiguration.h
@@ -15,8 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_CORE_YAML_YAMLCONFIGURATION_H_
-#define LIBMINIFI_INCLUDE_CORE_YAML_YAMLCONFIGURATION_H_
+#pragma once
#include <memory>
#include <optional>
@@ -37,11 +36,7 @@
class YamlConfigurationTestAccessor;
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace core {
+namespace org::apache::nifi::minifi::core {
static constexpr char const* CONFIG_YAML_FLOW_CONTROLLER_KEY = "Flow Controller";
static constexpr char const* CONFIG_YAML_PROCESSORS_KEY = "Processors";
@@ -50,6 +45,8 @@
static constexpr char const* CONFIG_YAML_REMOTE_PROCESS_GROUP_KEY_V3 = "Remote Process Groups";
static constexpr char const* CONFIG_YAML_PROVENANCE_REPORT_KEY = "Provenance Reporting";
static constexpr char const* CONFIG_YAML_FUNNELS_KEY = "Funnels";
+static constexpr char const* CONFIG_YAML_INPUT_PORTS_KEY = "Input Ports";
+static constexpr char const* CONFIG_YAML_OUTPUT_PORTS_KEY = "Output Ports";
#define YAML_CONFIGURATION_USE_REGEX
@@ -262,6 +259,17 @@
void parseFunnelsYaml(const YAML::Node& node, core::ProcessGroup* parent);
/**
+ * Parses the Input/Output Ports section of a configuration YAML.
+ * The resulting ports are added to the parent ProcessGroup.
+ *
+ * @param node the YAML::Node containing the Input/Output Ports section
+ * of the configuration YAML
+ * @param parent the root node of flow configuration to which
+ * to add the funnels that are parsed
+ */
+ void parsePorts(const YAML::Node& node, core::ProcessGroup* parent, PortType port_type);
+
+ /**
* A helper function for parsing or generating optional id fields.
*
* In parsing YAML flow configurations for config schema v1, the
@@ -320,10 +328,4 @@
void raiseComponentError(const std::string &component_name, const std::string &yaml_section, const std::string &reason) const;
};
-} // namespace core
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
-
-#endif // LIBMINIFI_INCLUDE_CORE_YAML_YAMLCONFIGURATION_H_
+} // namespace org::apache::nifi::minifi::core
diff --git a/libminifi/src/core/Funnel.cpp b/libminifi/src/ForwardingNode.cpp
similarity index 72%
rename from libminifi/src/core/Funnel.cpp
rename to libminifi/src/ForwardingNode.cpp
index 5d31194..c1bf0a1 100644
--- a/libminifi/src/core/Funnel.cpp
+++ b/libminifi/src/ForwardingNode.cpp
@@ -15,18 +15,18 @@
* limitations under the License.
*/
-#include "core/Funnel.h"
+#include "ForwardingNode.h"
#include "core/ProcessSession.h"
-namespace org::apache::nifi::minifi::core {
+namespace org::apache::nifi::minifi {
-const Relationship Funnel::Success("success", "FlowFiles are routed to success relationship");
+const core::Relationship ForwardingNode::Success("success", "FlowFiles are routed to success relationship");
-void Funnel::initialize() {
+void ForwardingNode::initialize() {
setSupportedRelationships(relationships());
}
-void Funnel::onTrigger(const std::shared_ptr<core::ProcessContext>& /*context*/, const std::shared_ptr<core::ProcessSession>& session) {
+void ForwardingNode::onTrigger(const std::shared_ptr<core::ProcessContext>& /*context*/, const std::shared_ptr<core::ProcessSession>& session) {
logger_->log_trace("On trigger %s", getUUIDStr());
std::shared_ptr<core::FlowFile> flow_file = session->get();
if (!flow_file) {
@@ -35,4 +35,4 @@
session->transfer(flow_file, Success);
}
-} // namespace org::apache::nifi::minifi::core
+} // namespace org::apache::nifi::minifi
diff --git a/libminifi/src/core/ProcessGroup.cpp b/libminifi/src/core/ProcessGroup.cpp
index b7444ec..7922717 100644
--- a/libminifi/src/core/ProcessGroup.cpp
+++ b/libminifi/src/core/ProcessGroup.cpp
@@ -91,7 +91,7 @@
}
-void ProcessGroup::addProcessor(std::unique_ptr<Processor> processor) {
+std::tuple<Processor*, bool> ProcessGroup::addProcessor(std::unique_ptr<Processor> processor) {
gsl_Expects(processor);
const auto name = processor->getName();
std::lock_guard<std::recursive_mutex> lock(mutex_);
@@ -101,6 +101,15 @@
} else {
logger_->log_debug("Not adding processor %s into process group %s, as it is already there", name, name_);
}
+ return std::make_tuple(iter->get(), inserted);
+}
+
+void ProcessGroup::addPort(std::unique_ptr<Port> port) {
+ auto [processor, inserted] = addProcessor(std::move(port));
+ if (inserted) {
+ std::lock_guard<std::recursive_mutex> lock(mutex_);
+ ports_.insert(static_cast<Port*>(processor));
+ }
}
void ProcessGroup::addProcessGroup(std::unique_ptr<ProcessGroup> child) {
@@ -323,6 +332,33 @@
}
}
+Port* ProcessGroup::findPortById(const std::set<Port*>& ports, const utils::Identifier& uuid) {
+ const auto found = ranges::find_if(ports, [&](auto port) {
+ utils::Identifier port_uuid = port->getUUID();
+ return port_uuid && uuid == port_uuid;
+ });
+ if (found != ranges::cend(ports)) {
+ return *found;
+ }
+ return nullptr;
+}
+
+Port* ProcessGroup::findPortById(const utils::Identifier& uuid) const {
+ std::lock_guard<std::recursive_mutex> lock(mutex_);
+ return findPortById(ports_, uuid);
+}
+
+Port* ProcessGroup::findChildPortById(const utils::Identifier& uuid) const {
+ std::lock_guard<std::recursive_mutex> lock(mutex_);
+ for (const auto& processGroup : child_process_groups_) {
+ const auto& ports = processGroup->getPorts();
+ if (auto port = findPortById(ports, uuid)) {
+ return port;
+ }
+ }
+ return nullptr;
+}
+
void ProcessGroup::addConnection(std::unique_ptr<Connection> connection) {
std::lock_guard<std::recursive_mutex> lock(mutex_);
@@ -334,19 +370,52 @@
auto& insertedConnection = *insertPos;
logger_->log_debug("Add connection %s into process group %s", insertedConnection->getName(), name_);
- // only allow connections between processors of the same process group
- auto source = this->findProcessorById(insertedConnection->getSourceUUID(), Traverse::ExcludeChildren);
+ // only allow connections between processors of the same process group or in/output ports of child process groups
+ // check input and output ports connection restrictions inside and outside a process group
+ Processor* source = findPortById(insertedConnection->getSourceUUID());
+ if (source && static_cast<Port*>(source)->getPortType() == PortType::OUTPUT) {
+ logger_->log_error("Output port [id = '%s'] cannot be a source inside the process group in the connection [name = '%s', id = '%s']",
+ insertedConnection->getSourceUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
+ source = nullptr;
+ } else if (!source) {
+ source = findChildPortById(insertedConnection->getSourceUUID());
+ if (source && static_cast<Port*>(source)->getPortType() == PortType::INPUT) {
+ logger_->log_error("Input port [id = '%s'] cannot be a source outside the process group in the connection [name = '%s', id = '%s']",
+ insertedConnection->getSourceUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
+ source = nullptr;
+ } else if (!source) {
+ source = findProcessorById(insertedConnection->getSourceUUID(), Traverse::ExcludeChildren);
+ if (!source) {
+ logger_->log_error("Cannot find the source processor with id '%s' for the connection [name = '%s', id = '%s']",
+ insertedConnection->getSourceUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
+ }
+ }
+ }
+
if (source) {
source->addConnection(insertedConnection.get());
- } else {
- logger_->log_error("Cannot find the source processor with id '%s' for the connection [name = '%s', id = '%s']",
- insertedConnection->getSourceUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
}
- auto destination = this->findProcessorById(insertedConnection->getDestinationUUID(), Traverse::ExcludeChildren);
- if (!destination) {
- logger_->log_error("Cannot find the destination processor with id '%s' for the connection [name = '%s', id = '%s']",
+
+ Processor* destination = findPortById(insertedConnection->getDestinationUUID());
+ if (destination && static_cast<Port*>(destination)->getPortType() == PortType::INPUT) {
+ logger_->log_error("Input port [id = '%s'] cannot be a destination inside the process group in the connection [name = '%s', id = '%s']",
insertedConnection->getDestinationUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
+ destination = nullptr;
+ } else if (!destination) {
+ destination = findChildPortById(insertedConnection->getDestinationUUID());
+ if (destination && static_cast<Port*>(destination)->getPortType() == PortType::OUTPUT) {
+ logger_->log_error("Output port [id = '%s'] cannot be a destination outside the process group in the connection [name = '%s', id = '%s']",
+ insertedConnection->getDestinationUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
+ destination = nullptr;
+ } else if (!destination) {
+ destination = findProcessorById(insertedConnection->getDestinationUUID(), Traverse::ExcludeChildren);
+ if (!destination) {
+ logger_->log_error("Cannot find the destination processor with id '%s' for the connection [name = '%s', id = '%s']",
+ insertedConnection->getDestinationUUID().to_string(), insertedConnection->getName(), insertedConnection->getUUIDStr());
+ }
+ }
}
+
if (destination && destination != source) {
destination->addConnection(insertedConnection.get());
}
diff --git a/libminifi/src/core/yaml/YamlConfiguration.cpp b/libminifi/src/core/yaml/YamlConfiguration.cpp
index e1f23f3..0560d85 100644
--- a/libminifi/src/core/yaml/YamlConfiguration.cpp
+++ b/libminifi/src/core/yaml/YamlConfiguration.cpp
@@ -27,6 +27,7 @@
#include "core/state/Value.h"
#include "Defaults.h"
#include "utils/TimeUtil.h"
+#include "Funnel.h"
#ifdef YAML_CONFIGURATION_USE_REGEX
#include "utils/RegexUtils.h"
@@ -93,6 +94,8 @@
YAML::Node processorsNode = yamlNode[CONFIG_YAML_PROCESSORS_KEY];
YAML::Node connectionsNode = yamlNode[yaml::YamlConnectionParser::CONFIG_YAML_CONNECTIONS_KEY];
YAML::Node funnelsNode = yamlNode[CONFIG_YAML_FUNNELS_KEY];
+ YAML::Node inputPortsNode = yamlNode[CONFIG_YAML_INPUT_PORTS_KEY];
+ YAML::Node outputPortsNode = yamlNode[CONFIG_YAML_OUTPUT_PORTS_KEY];
YAML::Node remoteProcessingGroupsNode = [&] {
// assignment is not supported on invalid Yaml nodes
YAML::Node candidate = yamlNode[CONFIG_YAML_REMOTE_PROCESS_GROUP_KEY];
@@ -106,9 +109,8 @@
parseProcessorNodeYaml(processorsNode, group.get());
parseRemoteProcessGroupYaml(remoteProcessingGroupsNode, group.get());
parseFunnelsYaml(funnelsNode, group.get());
- // parse connections last to give feedback if the source and/or destination
- // is not in the same process group
- parseConnectionYaml(connectionsNode, group.get());
+ parsePorts(inputPortsNode, group.get(), PortType::INPUT);
+ parsePorts(outputPortsNode, group.get(), PortType::OUTPUT);
if (childProcessGroupNodeSeq && childProcessGroupNodeSeq.IsSequence()) {
for (YAML::const_iterator it = childProcessGroupNodeSeq.begin(); it != childProcessGroupNodeSeq.end(); ++it) {
@@ -116,6 +118,10 @@
group->addProcessGroup(parseProcessGroupYaml(childProcessGroupNode, childProcessGroupNode));
}
}
+
+ // parse connections last to give feedback if the source and/or destination processors
+ // is not in the same process group or input/output port connections are not allowed
+ parseConnectionYaml(connectionsNode, group.get());
return group;
}
@@ -773,7 +779,7 @@
throw Exception(ExceptionType::GENERAL_EXCEPTION, "Incorrect funnel UUID format.");
});
- auto funnel = std::make_unique<core::Funnel>(name, uuid.value());
+ auto funnel = std::make_unique<Funnel>(name, uuid.value());
logger_->log_debug("Created funnel with UUID %s and name %s", id, name);
funnel->setScheduledState(core::RUNNING);
funnel->setSchedulingStrategy(core::EVENT_DRIVEN);
@@ -781,6 +787,36 @@
}
}
+void YamlConfiguration::parsePorts(const YAML::Node& node, core::ProcessGroup* parent, PortType port_type) {
+ if (!parent) {
+ logger_->log_error("parsePorts: no parent group was provided");
+ return;
+ }
+ if (!node || !node.IsSequence()) {
+ return;
+ }
+
+ for (const auto& element : node) {
+ const auto port_node = element.as<YAML::Node>();
+
+ std::string id = getOrGenerateId(port_node);
+
+ // Default name to be same as ID
+ const auto name = port_node["name"].as<std::string>(id);
+
+ const auto uuid = utils::Identifier::parse(id) | utils::orElse([this] {
+ logger_->log_debug("Incorrect port UUID format.");
+ throw Exception(ExceptionType::GENERAL_EXCEPTION, "Incorrect port UUID format.");
+ });
+
+ auto port = std::make_unique<Port>(name, uuid.value(), port_type);
+ logger_->log_debug("Created port UUID %s and name %s", id, name);
+ port->setScheduledState(core::RUNNING);
+ port->setSchedulingStrategy(core::EVENT_DRIVEN);
+ parent->addPort(std::move(port));
+ }
+}
+
void YamlConfiguration::validateComponentProperties(ConfigurableComponent& component, const std::string &component_name, const std::string &yaml_section) const {
const auto &component_properties = component.getProperties();
diff --git a/libminifi/src/core/yaml/YamlConnectionParser.cpp b/libminifi/src/core/yaml/YamlConnectionParser.cpp
index 5e246d8..9435be3 100644
--- a/libminifi/src/core/yaml/YamlConnectionParser.cpp
+++ b/libminifi/src/core/yaml/YamlConnectionParser.cpp
@@ -18,6 +18,7 @@
#include "core/yaml/YamlConnectionParser.h"
#include "core/yaml/CheckRequiredField.h"
+#include "Funnel.h"
namespace org::apache::nifi::minifi::core::yaml {
@@ -41,8 +42,8 @@
}
auto& processor_ref = *processor;
- if (typeid(minifi::core::Funnel) == typeid(processor_ref)) {
- addNewRelationshipToConnection(minifi::core::Funnel::Success.getName(), connection);
+ if (typeid(minifi::Funnel) == typeid(processor_ref)) {
+ addNewRelationshipToConnection(minifi::Funnel::Success.getName(), connection);
}
}
diff --git a/libminifi/test/resources/TestProcessGroup.yml b/libminifi/test/resources/TestProcessGroup.yml
new file mode 100644
index 0000000..0b1ee75
--- /dev/null
+++ b/libminifi/test/resources/TestProcessGroup.yml
@@ -0,0 +1,69 @@
+MiNiFi Config Version: 3
+Flow Controller:
+ name: MiNiFi Flow
+Processors:
+- name: GenerateFlowFile
+ id: 4812b638-2f79-4dc2-9693-847a90399cbd
+ class: org.apache.nifi.minifi.processors.GenerateFlowFile
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 100 ms
+ penalization period: 1000 ms
+ Properties:
+ Batch Size: '1'
+ Data Format: Binary
+ File Size: 10 B
+ Unique FlowFiles: 'true'
+- name: Log attributes
+ id: 2ecd4bb4-b103-43fe-a45d-6a79b12da79b
+ class: org.apache.nifi.minifi.processors.LogAttribute
+ scheduling strategy: EVENT_DRIVEN
+ auto-terminated relationships list:
+ - success
+ Properties:
+ FlowFiles To Log: '0'
+Connections:
+- name: GenerateFlowFile/success/ProcessGroup
+ id: 492bc370-5d4c-4657-952f-3d6093147ad8
+ source id: 4812b638-2f79-4dc2-9693-847a90399cbd
+ source relationship names:
+ - success
+ destination id: 012fc536-3137-4360-be65-3e3b47e05941
+- name: ProcessGroup/success/LogAttribute
+ id: 12656e8e-0b91-4694-a2b7-3aa147574cd2
+ source id: 46dd8c65-8255-4980-8b7e-4381da00867a
+ source relationship names:
+ - success
+ destination id: 2ecd4bb4-b103-43fe-a45d-6a79b12da79b
+Controller Services: []
+Remote Process Groups: []
+Process Groups:
+ - id: 0a3aaf32-8574-4fa7-b720-84001f8dd71a
+ name: Update the attributes
+ Processors:
+ - id: 11624e01-baca-4590-bb9d-512ae2616615
+ name: UpdateAttribute
+ class: org.apache.nifi.minifi.processors.UpdateAttribute
+ scheduling strategy: EVENT_DRIVEN
+ auto-terminated relationships list:
+ - failure
+ Properties:
+ test_attribute: success
+ Input Ports:
+ - id: 012fc536-3137-4360-be65-3e3b47e05941
+ name: in
+ Output Ports:
+ - id: 46dd8c65-8255-4980-8b7e-4381da00867a
+ name: out
+ Connections:
+ - name: Input/success/UpdateAttribute
+ id: 2d33779c-2305-4e1a-88b8-1d2b6a9b134c
+ source id: 012fc536-3137-4360-be65-3e3b47e05941
+ source relationship names:
+ - success
+ destination id: 11624e01-baca-4590-bb9d-512ae2616615
+ - name: UpdateAttribute/success/Output
+ id: 5af95cc8-455a-4d5e-afbb-e16699407ed2
+ source id: 11624e01-baca-4590-bb9d-512ae2616615
+ source relationship names:
+ - success
+ destination id: 46dd8c65-8255-4980-8b7e-4381da00867a
