This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch enrichment_processors
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/enrichment_processors by this
push:
new 390299a8b review changes
390299a8b is described below
commit 390299a8bed3734f09c87acb72364e6aea3cc361
Author: Martin Zink <[email protected]>
AuthorDate: Mon Oct 19 08:00:33 2026 +0200
review changes
---
.../ubuntu_22_04_clang_arm_manifest.json | 2 +-
PROCESSORS.md | 2 +-
extension-framework/include/core/FlowFileStore.h | 4 +-
.../processors/ForkEnrichment.cpp | 7 ++-
.../processors/ForkEnrichment.h | 13 +++---
.../processors/JoinEnrichmentAttributes.cpp | 53 ++++++++++------------
.../processors/JoinEnrichmentAttributes.h | 16 +++----
.../tests/features/enrichment.feature | 15 ++++--
.../tests/unit/ForkEnrichmentTests.cpp | 8 ++--
.../tests/unit/JoinEnrichmentAttributesTests.cpp | 51 ++++++++++++++++++++-
.../standard-processors/utils/EnrichmentUtils.h | 11 +++++
11 files changed, 125 insertions(+), 57 deletions(-)
diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json
b/.github/references/ubuntu_22_04_clang_arm_manifest.json
index 01199f116..fbbb8a95b 100644
--- a/.github/references/ubuntu_22_04_clang_arm_manifest.json
+++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json
@@ -9962,7 +9962,7 @@
},
{
"name": "joined",
- "description": "The resultant FlowFile with
Records joined together from both the original and enrichment FlowFiles will be
routed to this relationship"
+ "description": "The resultant FlowFile with
attributes joined together from both the original and enrichment FlowFiles will
be routed to this relationship"
},
{
"name": "original",
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 68487069c..3803226fe 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -1328,7 +1328,7 @@ In the list below, the names of required properties
appear in bold. Any other pr
| Name | Description
|
|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| invalid | Any FlowFiles without the requisite attributes will be routed
here
|
-| joined | The resultant FlowFile with Records joined together from both the
original and enrichment FlowFiles will be routed to this relationship
|
+| joined | The resultant FlowFile with attributes joined together from both
the original and enrichment FlowFiles will be routed to this relationship
|
| original | Both of the incoming FlowFiles ('original' and 'enrichment') will
be routed to this Relationship. I.e., this is the 'original' version of both of
these FlowFiles.
|
| timeout | If one of the incoming FlowFiles (i.e., the 'original' FlowFile
or the 'enrichment' FlowFile) arrives to this Processor but the other does not
arrive within the configured Timeout period, the FlowFile that did arrive is
routed to this relationship. |
diff --git a/extension-framework/include/core/FlowFileStore.h
b/extension-framework/include/core/FlowFileStore.h
index 94df5722f..b5d88148b 100644
--- a/extension-framework/include/core/FlowFileStore.h
+++ b/extension-framework/include/core/FlowFileStore.h
@@ -17,7 +17,9 @@
#pragma once
+#include <atomic>
#include <memory>
+#include <mutex>
#include <unordered_set>
#include <utility>
@@ -32,7 +34,7 @@ class FlowFileStore {
return {};
}
std::lock_guard<std::mutex> guard(flow_file_mutex_);
- return std::move(incoming_files_);
+ return std::exchange(incoming_files_, {});
}
void put(const std::shared_ptr<FlowFile>& flowFile) {
diff --git a/extensions/standard-processors/processors/ForkEnrichment.cpp
b/extensions/standard-processors/processors/ForkEnrichment.cpp
index 9cc3b34de..36e00b704 100644
--- a/extensions/standard-processors/processors/ForkEnrichment.cpp
+++ b/extensions/standard-processors/processors/ForkEnrichment.cpp
@@ -17,7 +17,10 @@
#include "ForkEnrichment.h"
+#include <string>
+
#include "core/Resource.h"
+#include "magic_enum/magic_enum.hpp"
#include "minifi-cpp/core/ProcessSession.h"
#include "utils/ProcessorConfigUtils.h"
@@ -42,8 +45,8 @@ void ForkEnrichment::onTrigger(core::ProcessContext&,
core::ProcessSession& sess
while (const auto original = session.get()) {
const auto enrichment = session.clone(*original);
- original->setAttribute(ENRICHMENT_ROLE, "ORIGINAL");
- enrichment->setAttribute(ENRICHMENT_ROLE, "ENRICHMENT");
+ original->setAttribute(ENRICHMENT_ROLE,
std::string{magic_enum::enum_name(EnrichmentRole::ORIGINAL)});
+ enrichment->setAttribute(ENRICHMENT_ROLE,
std::string{magic_enum::enum_name(EnrichmentRole::ENRICHMENT)});
const std::string group_id =
utils::IdGenerator::getIdGenerator()->generate().to_string();
original->setAttribute(ENRICHMENT_GROUP_ID, group_id);
diff --git a/extensions/standard-processors/processors/ForkEnrichment.h
b/extensions/standard-processors/processors/ForkEnrichment.h
index a8f103bfc..b14299df2 100644
--- a/extensions/standard-processors/processors/ForkEnrichment.h
+++ b/extensions/standard-processors/processors/ForkEnrichment.h
@@ -17,14 +17,14 @@
#pragma once
-#include <string>
+#include <array>
+#include <cstdint>
+#include <optional>
#include "core/ProcessorImpl.h"
#include "core/PropertyDefinitionBuilder.h"
#include "minifi-cpp/core/PropertyDefinition.h"
#include "utils/EnrichmentUtils.h"
-#include "utils/Enum.h"
-#include "utils/RegexUtils.h"
namespace org::apache::nifi::minifi::standard {
@@ -56,13 +56,14 @@ class ForkEnrichment : public core::ProcessorImpl {
EXTENSIONAPI static constexpr core::annotation::Input InputRequirement =
core::annotation::Input::INPUT_REQUIRED;
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
- EXTENSIONAPI static constexpr auto EnrichmentRole =
core::OutputAttributeDefinition<2>{
+ EXTENSIONAPI static constexpr auto EnrichmentRoleOutputAttribute =
core::OutputAttributeDefinition<2>{
ENRICHMENT_ROLE, {Enrichment, Original}, "The role to use for
enrichment. This will either be ORIGINAL or ENRICHMENT."};
- EXTENSIONAPI static constexpr auto EnrichmentGroupId =
core::OutputAttributeDefinition<2>{ENRICHMENT_GROUP_ID,
+ EXTENSIONAPI static constexpr auto EnrichmentGroupIdOutputAttribute =
core::OutputAttributeDefinition<2>{ENRICHMENT_GROUP_ID,
{Enrichment, Original},
"The Group ID to use in order to correlate the 'original' FlowFile with
the 'enrichment' FlowFile."};
- EXTENSIONAPI static constexpr auto OutputAttributes =
std::array<core::OutputAttributeReference, 2>{EnrichmentRole,
EnrichmentGroupId};
+ EXTENSIONAPI static constexpr auto OutputAttributes =
+ std::array<core::OutputAttributeReference,
2>{EnrichmentRoleOutputAttribute, EnrichmentGroupIdOutputAttribute};
ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
diff --git
a/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp
b/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp
index 9a616e1c8..84d56d644 100644
--- a/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp
+++ b/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp
@@ -16,9 +16,12 @@
*/
#include "JoinEnrichmentAttributes.h"
+#include <string>
+#include <utility>
+
#include "core/Resource.h"
+#include "magic_enum/magic_enum.hpp"
#include "minifi-cpp/core/ProcessSession.h"
-#include "utils/AttributeErrors.h"
#include "utils/EnrichmentUtils.h"
#include "utils/ProcessorConfigUtils.h"
@@ -33,8 +36,10 @@ void JoinEnrichmentAttributes::initialize() {
void JoinEnrichmentAttributes::onSchedule(core::ProcessContext& context,
core::ProcessSessionFactory& session_factory) {
using namespace std::literals::chrono_literals;
+ time_out_tracker_.reset();
if (const auto timeout = utils::parseOptionalDurationProperty(context,
TimeoutProperty); timeout && *timeout > 0ms) {
time_out_tracker_.emplace(*timeout);
+ context.setTriggerWhenEmpty(true);
}
max_batch_size_ = utils::parseOptionalU64Property(context, MaxBatchSize);
if (max_batch_size_ && *max_batch_size_ == 0) {
@@ -43,27 +48,23 @@ void
JoinEnrichmentAttributes::onSchedule(core::ProcessContext& context, core::P
ProcessorImpl::onSchedule(context, session_factory);
}
-namespace {
-bool checkRequiredAttributes(const core::FlowFile& flow_file) {
- return flow_file.getAttribute(ENRICHMENT_ROLE).has_value() &&
flow_file.getAttribute(ENRICHMENT_GROUP_ID).has_value();
+void JoinEnrichmentAttributes::addToSessionIfNeeded(const
std::shared_ptr<core::FlowFile>& flow_file, core::ProcessSession& session) {
+ if (session_flow_files_.insert(flow_file->getUUID()).second) {
+ session.add(flow_file);
+ }
}
-} // namespace
void JoinEnrichmentAttributes::join(const std::shared_ptr<core::FlowFile>&
original, const std::shared_ptr<core::FlowFile>& enrichment,
- core::ProcessSession& session) const {
+ core::ProcessSession& session) {
const auto cloned = session.clone(*original);
for (const auto& [k, v] : enrichment->getAttributes()) {
if (k != ENRICHMENT_ROLE) {
cloned->setAttribute(k, v);
}
}
- cloned->setAttribute(ENRICHMENT_ROLE, "JOINED");
- if (!std::ranges::contains(session_flow_files_, original->getUUID())) {
- session.add(original);
- }
- if (!std::ranges::contains(session_flow_files_, enrichment->getUUID())) {
- session.add(enrichment);
- }
+ cloned->setAttribute(ENRICHMENT_ROLE, std::string{JOINED_ROLE});
+ addToSessionIfNeeded(original, session);
+ addToSessionIfNeeded(enrichment, session);
session.transfer(original, Original);
session.transfer(enrichment, Original);
session.transfer(cloned, Joined);
@@ -71,32 +72,32 @@ void JoinEnrichmentAttributes::join(const
std::shared_ptr<core::FlowFile>& origi
void JoinEnrichmentAttributes::handleFlowFile(std::shared_ptr<core::FlowFile>
flow_file, core::ProcessSession& session,
const std::chrono::steady_clock::time_point current_time) {
- if (!checkRequiredAttributes(*flow_file)) {
- logger_->log_warn("{} is missing enrichment.group.id and/or
enrichment.role, routing it to Invalid", flow_file->getId());
+ addToSessionIfNeeded(flow_file, session);
+ const auto role_attribute = flow_file->getAttribute(ENRICHMENT_ROLE);
+ const auto group_id_attribute = flow_file->getAttribute(ENRICHMENT_GROUP_ID);
+ if (!role_attribute || !group_id_attribute) {
+ logger_->log_warn("{} is missing {} and/or {}, routing it to Invalid",
flow_file->getId(), ENRICHMENT_GROUP_ID, ENRICHMENT_ROLE);
session.transfer(flow_file, Invalid);
return;
}
- const auto role = flow_file->getAttribute(ENRICHMENT_ROLE) |
utils::toExpected(make_error_code(core::AttributeErrorCode::MissingAttribute)) |
- utils::andThen(parsing::parseEnum<EnrichmentRole>);
+ const auto role = parsing::parseEnum<EnrichmentRole>(*role_attribute);
if (!role) {
- logger_->log_warn("{} has invalid role due to {}", flow_file->getId(),
role.error());
+ logger_->log_warn("{} has invalid {} due to {}", flow_file->getId(),
ENRICHMENT_ROLE, role.error());
session.transfer(flow_file, Invalid);
return;
}
- std::string group_id = *(flow_file->getAttribute(ENRICHMENT_GROUP_ID));
+ std::string group_id = *group_id_attribute;
auto& my_map = role == EnrichmentRole::ENRICHMENT ? enrichments_ :
originals_;
auto& pair_map = role == EnrichmentRole::ENRICHMENT ? originals_ :
enrichments_;
if (const auto previous_node = my_map.extract(group_id)) {
logger_->log_warn("Encountered duplicate {} for {}, routing both to
Invalid", magic_enum::enum_name(*role), group_id);
+ addToSessionIfNeeded(previous_node.mapped(), session);
session.transfer(flow_file, Invalid);
session.transfer(previous_node.mapped(), Invalid);
- if (!std::ranges::contains(session_flow_files_,
previous_node.mapped()->getUUID())) {
- session.add(previous_node.mapped());
- }
return;
}
@@ -116,6 +117,7 @@ void
JoinEnrichmentAttributes::handleFlowFile(std::shared_ptr<core::FlowFile> fl
}
void JoinEnrichmentAttributes::onTrigger(core::ProcessContext&,
core::ProcessSession& session) {
+ session_flow_files_.clear();
const auto current_time = std::chrono::steady_clock::now();
for (auto flow_file : flow_file_store_.getNewFlowFiles()) {
handleFlowFile(std::move(flow_file), session, current_time);
@@ -133,10 +135,7 @@ void
JoinEnrichmentAttributes::onTrigger(core::ProcessContext&, core::ProcessSes
for (auto timed_out_group :
time_out_tracker_->getTimedOutFlowFiles(current_time)) {
const auto removeFromMap = [&](StoredFlowFileMap& map) {
if (auto timed_out_node = map.extract(timed_out_group)) {
- // Fresh FlowFiles shouldn't time out (we only use a single
time_point per session)
- gsl_AssertAudit(!std::ranges::contains(session_flow_files_,
timed_out_node.mapped()->getUUID()));
-
- session.add(timed_out_node.mapped());
+ addToSessionIfNeeded(timed_out_node.mapped(), session);
session.transfer(timed_out_node.mapped(), TimeoutRelationship);
}
};
@@ -144,8 +143,6 @@ void
JoinEnrichmentAttributes::onTrigger(core::ProcessContext&, core::ProcessSes
removeFromMap(enrichments_);
}
}
-
- session_flow_files_.clear();
}
void JoinEnrichmentAttributes::restore(const std::shared_ptr<core::FlowFile>&
flowFile) {
diff --git
a/extensions/standard-processors/processors/JoinEnrichmentAttributes.h
b/extensions/standard-processors/processors/JoinEnrichmentAttributes.h
index 7a1d18bca..f4f40bebb 100644
--- a/extensions/standard-processors/processors/JoinEnrichmentAttributes.h
+++ b/extensions/standard-processors/processors/JoinEnrichmentAttributes.h
@@ -25,6 +25,7 @@
#include <optional>
#include <string>
#include <unordered_map>
+#include <unordered_set>
#include <utility>
#include <vector>
@@ -32,8 +33,9 @@
#include "core/ProcessorImpl.h"
#include "core/PropertyDefinitionBuilder.h"
#include "minifi-cpp/core/PropertyDefinition.h"
-#include "utils/Enum.h"
-#include "utils/RegexUtils.h"
+#include "utils/EnrichmentUtils.h"
+#include "utils/Id.h"
+#include "utils/StringUtils.h"
namespace org::apache::nifi::minifi::standard {
@@ -86,7 +88,7 @@ class JoinEnrichmentAttributes : public core::ProcessorImpl {
EXTENSIONAPI static constexpr auto Invalid =
core::RelationshipDefinition{"invalid",
"Any FlowFiles without the requisite attributes will be routed here"};
EXTENSIONAPI static constexpr auto Joined =
core::RelationshipDefinition{"joined",
- "The resultant FlowFile with Records joined together from both the
original and enrichment FlowFiles will be routed to this relationship"};
+ "The resultant FlowFile with attributes joined together from both the
original and enrichment FlowFiles will be routed to this relationship"};
EXTENSIONAPI static constexpr auto Original =
core::RelationshipDefinition{"original",
"Both of the incoming FlowFiles ('original' and 'enrichment') will be
routed to this Relationship. I.e., this is the 'original' version of "
"both of these FlowFiles."};
@@ -127,13 +129,9 @@ class JoinEnrichmentAttributes : public
core::ProcessorImpl {
void restore(const std::shared_ptr<core::FlowFile>& flowFile) override;
private:
- enum class EnrichmentRole {
- ORIGINAL,
- ENRICHMENT,
- };
-
void handleFlowFile(std::shared_ptr<core::FlowFile> flow_file,
core::ProcessSession& session, std::chrono::steady_clock::time_point
current_time);
- void join(const std::shared_ptr<core::FlowFile>& original, const
std::shared_ptr<core::FlowFile>& enrichment, core::ProcessSession& session)
const;
+ void join(const std::shared_ptr<core::FlowFile>& original, const
std::shared_ptr<core::FlowFile>& enrichment, core::ProcessSession& session);
+ void addToSessionIfNeeded(const std::shared_ptr<core::FlowFile>& flow_file,
core::ProcessSession& session);
core::FlowFileStore flow_file_store_;
// We need to track current session's FlowFiles (we cant add those)
diff --git a/extensions/standard-processors/tests/features/enrichment.feature
b/extensions/standard-processors/tests/features/enrichment.feature
index 429b535a8..e45fa2d9c 100644
--- a/extensions/standard-processors/tests/features/enrichment.feature
+++ b/extensions/standard-processors/tests/features/enrichment.feature
@@ -22,6 +22,9 @@ Feature: ForkEnrichment and JoinEnrichmentAttributes
And the "Data Format" property of the GenerateFlowFile processor is set to
"Text"
And the "Unique FlowFiles" property of the GenerateFlowFile processor is
set to "false"
+ And an UpdateAttribute processor with the name "SetOriginalAttributes" and
the "kept_prop" property set to "original_value"
+ And the "overwritten_prop" property of the SetOriginalAttributes processor
is set to "original_value"
+
And a ForkEnrichment processor
And a JoinEnrichmentAttributes processor
@@ -29,19 +32,23 @@ Feature: ForkEnrichment and JoinEnrichmentAttributes
And the "Replacement Strategy" property of the ReplaceText processor is
set to "Always Replace"
And the "Replacement Value" property of the ReplaceText processor is set
to "replaced_content"
- And an UpdateAttribute processor with the "extra_prop" property set to
"foo"
+ And an UpdateAttribute processor with the name "EnrichAttributes" and the
"extra_prop" property set to "foo"
+ And the "overwritten_prop" property of the EnrichAttributes processor is
set to "enriched_value"
And a LogAttribute processor with the "Log Payload" property set to "true"
- And the "success" relationship of the GenerateFlowFile processor is
connected to the ForkEnrichment
+ And the "success" relationship of the GenerateFlowFile processor is
connected to the SetOriginalAttributes
+ And the "success" relationship of the SetOriginalAttributes processor is
connected to the ForkEnrichment
And the "original" relationship of the ForkEnrichment processor is
connected to the JoinEnrichmentAttributes
And the "enrichment" relationship of the ForkEnrichment processor is
connected to the ReplaceText
- And the "success" relationship of the ReplaceText processor is connected
to the UpdateAttribute
- And the "success" relationship of the UpdateAttribute processor is
connected to the JoinEnrichmentAttributes
+ And the "success" relationship of the ReplaceText processor is connected
to the EnrichAttributes
+ And the "success" relationship of the EnrichAttributes processor is
connected to the JoinEnrichmentAttributes
And the "joined" relationship of the JoinEnrichmentAttributes processor is
connected to the LogAttribute
And JoinEnrichmentAttributes's original relationship is auto-terminated
And LogAttribute's success relationship is auto-terminated
When the MiNiFi instance starts up
Then the Minifi logs contain the following message: "key:enrichment.role
value:JOINED" in less than 10 seconds
And the Minifi logs contain the following message: "key:extra_prop
value:foo" in less than 1 second
+ And the Minifi logs contain the following message: "key:kept_prop
value:original_value" in less than 1 second
+ And the Minifi logs contain the following message: "key:overwritten_prop
value:enriched_value" in less than 1 second
And the Minifi logs contain the following message: "original_content" in
less than 1 second
diff --git a/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp
b/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp
index a81ae1408..a1dd69231 100644
--- a/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp
+++ b/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp
@@ -37,11 +37,11 @@ TEST_CASE("Fork Enrichment processor works") {
CHECK(original_content == enrichment_content);
CHECK(original_content == "test_content");
-
CHECK(original_results.at(0)->getAttribute(ForkEnrichment::EnrichmentRole.name)
== "ORIGINAL");
-
CHECK(enrichment_results.at(0)->getAttribute(ForkEnrichment::EnrichmentRole.name)
== "ENRICHMENT");
+
CHECK(original_results.at(0)->getAttribute(ForkEnrichment::EnrichmentRoleOutputAttribute.name)
== "ORIGINAL");
+
CHECK(enrichment_results.at(0)->getAttribute(ForkEnrichment::EnrichmentRoleOutputAttribute.name)
== "ENRICHMENT");
-
CHECK(original_results.at(0)->getAttribute(ForkEnrichment::EnrichmentGroupId.name)
==
-
enrichment_results.at(0)->getAttribute(ForkEnrichment::EnrichmentGroupId.name));
+
CHECK(original_results.at(0)->getAttribute(ForkEnrichment::EnrichmentGroupIdOutputAttribute.name)
==
+
enrichment_results.at(0)->getAttribute(ForkEnrichment::EnrichmentGroupIdOutputAttribute.name));
}
TEST_CASE("ForkEnrichment no max batch size") {
diff --git
a/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp
b/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp
index beeb2db8e..7e9c31d06 100644
---
a/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp
+++
b/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp
@@ -18,6 +18,7 @@
#include <thread>
+#include "FlowFileRecord.h"
#include "JoinEnrichmentAttributes.h"
#include "unit/Catch.h"
#include "unit/ProcessorUtils.h"
@@ -112,7 +113,7 @@ TEST_CASE("JoinEnrichmentAttributes test timeout") {
// First trigger no output (it holds the original waiting for its pair)
CHECK(std::ranges::all_of(first_trigger, [](const auto& res) -> bool {
return res.second.empty(); }));
- std::this_thread::sleep_for(1ms);
+ std::this_thread::sleep_for(10ms);
const auto second_trigger =
test_controller.trigger(minifi::test::InputFlowFileData{.content = "second",
.attributes = {{std::string{ENRICHMENT_GROUP_ID}, "bar"},
{std::string{ENRICHMENT_ROLE}, "ENRICHMENT"}, {"second_attr", "2"}}});
CHECK(second_trigger.at(JoinEnrichmentAttributes::Original).empty());
@@ -121,6 +122,54 @@ TEST_CASE("JoinEnrichmentAttributes test timeout") {
REQUIRE(second_trigger.at(JoinEnrichmentAttributes::Joined).empty());
}
+TEST_CASE("JoinEnrichmentAttributes times out without further input") {
+ minifi::test::SingleProcessorTestController
+
test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes"));
+ const auto proc = test_controller.getProcessor();
+ CHECK(test_controller.plan->setProperty(proc,
JoinEnrichmentAttributes::TimeoutProperty.name, "1 ms"));
+
+ const auto first_trigger =
test_controller.trigger(minifi::test::InputFlowFileData{.content = "first",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"},
{std::string{ENRICHMENT_ROLE}, "ORIGINAL"}}});
+ CHECK(std::ranges::all_of(first_trigger, [](const auto& res) -> bool {
return res.second.empty(); }));
+ // the scheduler only triggers us again without input if the processor asked
for it
+ CHECK(proc->getTriggerWhenEmpty());
+
+ std::this_thread::sleep_for(10ms);
+ const auto second_trigger = test_controller.trigger();
+
CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).size()
== 1);
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Joined).empty());
+}
+
+TEST_CASE("JoinEnrichmentAttributes restored FlowFile without the enrichment
attributes is routed to Invalid") {
+ minifi::test::SingleProcessorTestController
+
test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes"));
+
test_controller.getProcessor()->restore(std::make_shared<minifi::FlowFileRecordImpl>());
+
+ const auto trigger_result = test_controller.trigger();
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Invalid).size() == 1);
+}
+
+TEST_CASE("JoinEnrichmentAttributes joins a restored FlowFile with its pair") {
+ minifi::test::SingleProcessorTestController
+
test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes"));
+ const auto restored_original =
std::make_shared<minifi::FlowFileRecordImpl>();
+ restored_original->setAttribute(ENRICHMENT_GROUP_ID, "foo");
+ restored_original->setAttribute(ENRICHMENT_ROLE, "ORIGINAL");
+ restored_original->setAttribute("first_attr", "1");
+ test_controller.getProcessor()->restore(restored_original);
+
+ const auto trigger_result =
test_controller.trigger(minifi::test::InputFlowFileData{.content = "second",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"},
{std::string{ENRICHMENT_ROLE}, "ENRICHMENT"}, {"second_attr", "2"}}});
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Invalid).empty());
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Original).size() == 2);
+ REQUIRE(trigger_result.at(JoinEnrichmentAttributes::Joined).size() == 1);
+
+ const auto joined_attrs =
trigger_result.at(JoinEnrichmentAttributes::Joined).at(0)->getAttributes();
+ CHECK(joined_attrs.at(std::string{ENRICHMENT_ROLE}) == "JOINED");
+ CHECK(joined_attrs.at("first_attr") == "1");
+ CHECK(joined_attrs.at("second_attr") == "2");
+}
+
TEST_CASE("JoinEnrichmentAttributes no max batch size") {
minifi::test::SingleProcessorTestController
test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes"));
diff --git a/extensions/standard-processors/utils/EnrichmentUtils.h
b/extensions/standard-processors/utils/EnrichmentUtils.h
index 76dae0052..e56c59b59 100644
--- a/extensions/standard-processors/utils/EnrichmentUtils.h
+++ b/extensions/standard-processors/utils/EnrichmentUtils.h
@@ -19,5 +19,16 @@
#include <string_view>
+namespace org::apache::nifi::minifi::standard {
+
constexpr std::string_view ENRICHMENT_ROLE = "enrichment.role";
constexpr std::string_view ENRICHMENT_GROUP_ID = "enrichment.group.id";
+
+enum class EnrichmentRole {
+ ORIGINAL,
+ ENRICHMENT,
+};
+
+constexpr std::string_view JOINED_ROLE = "JOINED";
+
+} // namespace org::apache::nifi::minifi::standard
