You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2026/09/18 13:38:53 UTC
(nifi-minifi-cpp) 02/04: cpp stable api enrichment
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch api_1_1_enrichment
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit ea70eaf1158af277d2775c9d3809c9425216fac6
Author: Martin Zink <ma...@apache.org>
AuthorDate: Mon Aug 24 16:01:56 2026 +0200
cpp stable api enrichment
---
extensions/enrichment/CMakeLists.txt | 35 ++++++
extensions/enrichment/EnrichmentUtils.h | 23 ++++
extensions/enrichment/ExtensionInitializer.cpp | 41 ++++++
extensions/enrichment/ForkEnrichment.cpp | 56 +++++++++
extensions/enrichment/ForkEnrichment.h | 78 ++++++++++++
extensions/enrichment/JoinEnrichmentAttributes.cpp | 124 ++++++++++++++++++
extensions/enrichment/JoinEnrichmentAttributes.h | 138 ++++++++++++++++++++
extensions/enrichment/tests/CMakeLists.txt | 40 ++++++
.../enrichment/tests/ForkEnrichmentTests.cpp | 70 +++++++++++
.../tests/JoinEnrichmentAttributesTests.cpp | 140 +++++++++++++++++++++
.../enrichment/tests/features/enrichment.feature | 47 +++++++
.../tests/features/enrichment_restart.feature | 71 +++++++++++
.../enrichment/tests/features/environment.py | 28 +++++
.../enrichment/tests/features/steps/steps.py | 21 ++++
14 files changed, 912 insertions(+)
diff --git a/extensions/enrichment/CMakeLists.txt b/extensions/enrichment/CMakeLists.txt
new file mode 100644
index 000000000..3588011c0
--- /dev/null
+++ b/extensions/enrichment/CMakeLists.txt
@@ -0,0 +1,35 @@
+#
+# 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.
+#
+
+if (NOT (ENABLE_ALL OR MINIFI_EXTENSION_ENRICHMENT))
+ return()
+endif()
+
+include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
+
+file(GLOB SOURCES "*.cpp")
+
+add_minifi_library(minifi-enrichment SHARED ${SOURCES})
+target_include_directories(minifi-enrichment PUBLIC "${CMAKE_SOURCE_DIR}/extensions/enrichment")
+target_include_directories(minifi-enrichment PUBLIC "${ENRICHMENT_INCLUDE_DIRS}")
+
+include(Fetchstduuid)
+target_link_libraries(minifi-enrichment minifi-cpp-extension-lib stduuid::stduuid)
+
+register_c_api_extension(minifi-enrichment "ENRICHMENT EXTENSION" ENRICHMENT-EXTENSION "Provides Fork/Join enrichment processors" "extensions/enrichment/tests")
diff --git a/extensions/enrichment/EnrichmentUtils.h b/extensions/enrichment/EnrichmentUtils.h
new file mode 100644
index 000000000..76dae0052
--- /dev/null
+++ b/extensions/enrichment/EnrichmentUtils.h
@@ -0,0 +1,23 @@
+/**
+ * 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_view>
+
+constexpr std::string_view ENRICHMENT_ROLE = "enrichment.role";
+constexpr std::string_view ENRICHMENT_GROUP_ID = "enrichment.group.id";
diff --git a/extensions/enrichment/ExtensionInitializer.cpp b/extensions/enrichment/ExtensionInitializer.cpp
new file mode 100644
index 000000000..19c26c071
--- /dev/null
+++ b/extensions/enrichment/ExtensionInitializer.cpp
@@ -0,0 +1,41 @@
+/**
+* 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.
+ */
+
+#include "api/core/Resource.h"
+#include "api/utils/minifi-c-utils.h"
+#include "JoinEnrichmentAttributes.h"
+#include "ForkEnrichment.h"
+
+#define MKSOC(x) #x
+#define MAKESTRING(x) MKSOC(x) // NOLINT(cppcoreguidelines-macro-usage)
+
+namespace minifi = org::apache::nifi::minifi;
+
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_API_VERSION;
+
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_context) {
+ minifi_extension_definition extension_definition{
+ .name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+ .version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
+ .group_name = minifi::api::utils::minifiStringView(MAKESTRING(MINIFI_EXTENSION_GROUP_NAME)),
+ .deinit = nullptr,
+ .user_data = nullptr
+ };
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
+ minifi::api::core::registerProcessors<minifi::enrichment::ForkEnrichment>(extension);
+ minifi::api::core::registerProcessors<minifi::enrichment::JoinEnrichmentAttributes>(extension);
+}
diff --git a/extensions/enrichment/ForkEnrichment.cpp b/extensions/enrichment/ForkEnrichment.cpp
new file mode 100644
index 000000000..52e35f460
--- /dev/null
+++ b/extensions/enrichment/ForkEnrichment.cpp
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#include "ForkEnrichment.h"
+
+#include "api/core/ProcessSession.h"
+#include "api/utils/ProcessorConfigUtils.h"
+#include "stduuid/uuid.hpp"
+
+namespace org::apache::nifi::minifi::enrichment {
+
+minifi_status ForkEnrichment::onScheduleImpl(api::core::ProcessContext& context) {
+ max_batch_size_ = api::utils::parseOptionalU64Property(context, MaxBatchSize);
+ if (max_batch_size_ && *max_batch_size_ == 0) {
+ return MINIFI_STATUS_VALIDATION_FAILED;
+ }
+
+ return MINIFI_STATUS_SUCCESS;
+}
+
+minifi_status ForkEnrichment::onTriggerImpl(api::core::ProcessContext&, api::core::ProcessSession& session) {
+ uint64_t processed = 0;
+ while (api::core::FlowFile original = session.get()) {
+ api::core::FlowFile enrichment = session.clone(original);
+
+ session.setAttribute(original, ENRICHMENT_ROLE, "ORIGINAL");
+ session.setAttribute(enrichment, ENRICHMENT_ROLE, "ENRICHMENT");
+
+ const std::string group_id = uuids::to_string(uuid_gen());
+ session.setAttribute(original, ENRICHMENT_GROUP_ID, group_id);
+ session.setAttribute(enrichment, ENRICHMENT_GROUP_ID, group_id);
+
+ session.transfer(std::move(original), Original);
+ session.transfer(std::move(enrichment), Enrichment);
+ if (max_batch_size_ && ++processed >= max_batch_size_) {
+ break;
+ }
+ }
+ return MINIFI_STATUS_SUCCESS;
+}
+
+} // namespace org::apache::nifi::minifi::enrichment
diff --git a/extensions/enrichment/ForkEnrichment.h b/extensions/enrichment/ForkEnrichment.h
new file mode 100644
index 000000000..312fb1a5d
--- /dev/null
+++ b/extensions/enrichment/ForkEnrichment.h
@@ -0,0 +1,78 @@
+/**
+ * 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 <random>
+
+#include "EnrichmentUtils.h"
+#include "api/core/ProcessorImpl.h"
+#include "api/utils/Export.h"
+#include "core/PropertyDefinitionBuilder.h"
+#include "minifi-cpp/core/Annotation.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+#include "stduuid/uuid.hpp"
+
+namespace org::apache::nifi::minifi::enrichment {
+
+class ForkEnrichment : public api::core::ProcessorImpl {
+ public:
+ using ProcessorImpl::ProcessorImpl;
+
+ EXTENSIONAPI static constexpr const char* Description =
+ "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary "
+ "for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have "
+ "appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and "
+ "then be routed to the 'enrichment' relationship.";
+
+ EXTENSIONAPI static constexpr auto MaxBatchSize =
+ core::PropertyDefinitionBuilder<>::createProperty("Max Batch Size")
+ .withDescription("The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.")
+ .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR)
+ .build();
+
+ EXTENSIONAPI static constexpr auto Enrichment = core::RelationshipDefinition{"enrichment",
+ "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes."};
+ EXTENSIONAPI static constexpr auto Original = core::RelationshipDefinition{"original",
+ "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes."};
+ EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 1>{MaxBatchSize};
+ EXTENSIONAPI static constexpr auto Relationships = std::array{Enrichment, Original};
+
+ EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
+ EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
+ 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>{
+ 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,
+ {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};
+
+ protected:
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+
+ private:
+ std::optional<uint64_t> max_batch_size_;
+ std::random_device rd;
+ std::mt19937 rng{rd()};
+ uuids::uuid_random_generator uuid_gen{rng};
+};
+} // namespace org::apache::nifi::minifi::enrichment
diff --git a/extensions/enrichment/JoinEnrichmentAttributes.cpp b/extensions/enrichment/JoinEnrichmentAttributes.cpp
new file mode 100644
index 000000000..bed6e0237
--- /dev/null
+++ b/extensions/enrichment/JoinEnrichmentAttributes.cpp
@@ -0,0 +1,124 @@
+/**
+ * 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.
+ */
+#include "JoinEnrichmentAttributes.h"
+
+#include "EnrichmentUtils.h"
+#include "api/core/Resource.h"
+#include "api/utils/ProcessorConfigUtils.h"
+#include "utils/AttributeErrors.h"
+
+namespace org::apache::nifi::minifi::enrichment {
+
+minifi_status JoinEnrichmentAttributes::onScheduleImpl(api::core::ProcessContext& context) {
+ using namespace std::literals::chrono_literals;
+ if (const auto timeout = api::utils::parseOptionalDurationProperty(context, TimeoutProperty); timeout && *timeout > 0ms) {
+ time_out_tracker_.emplace(*timeout);
+ }
+ max_batch_size_ = api::utils::parseOptionalU64Property(context, MaxBatchSize);
+ if (max_batch_size_ && *max_batch_size_ == 0) {
+ return MINIFI_STATUS_VALIDATION_FAILED;
+ }
+ return MINIFI_STATUS_SUCCESS;
+}
+
+namespace {
+bool checkRequiredAttributes(api::core::ProcessSession& session, api::core::FlowFile& flow_file) {
+ return session.getAttribute(flow_file, ENRICHMENT_ROLE).has_value() && session.getAttribute(flow_file, ENRICHMENT_GROUP_ID).has_value();
+}
+} // namespace
+
+void JoinEnrichmentAttributes::join(api::core::FlowFile& original, api::core::FlowFile& enrichment, api::core::ProcessSession& session) {
+ api::core::FlowFile joined = session.clone(original);
+ for (const auto& [k, v] : session.getAttributes(enrichment)) {
+ if (k != ENRICHMENT_ROLE) {
+ session.setAttribute(joined, k, v);
+ }
+ }
+ session.setAttribute(joined, ENRICHMENT_ROLE, "JOINED");
+ session.transfer(std::move(original), Original);
+ session.transfer(std::move(enrichment), Original);
+ session.transfer(std::move(joined), Joined);
+}
+
+void JoinEnrichmentAttributes::handleFlowFile(api::core::FlowFile flow_file, api::core::ProcessSession& session,
+ const std::chrono::steady_clock::time_point current_time) {
+ if (!checkRequiredAttributes(session, flow_file)) {
+ logger_->log_warn("{} is missing enrichment.group.id and/or enrichment.role, routing it to Invalid", session.getFlowFileId(flow_file));
+ session.transfer(std::move(flow_file), Invalid);
+ return;
+ }
+
+ // SAFETY: checkRequiredAttributes already checks for ENRICHMENT_ROLE
+ const auto role = parsing::parseEnum<EnrichmentRole>(*session.getAttribute(flow_file, ENRICHMENT_ROLE));
+ if (!role) {
+ logger_->log_warn("{} has invalid role due to {}", session.getFlowFileId(flow_file), role.error());
+ session.transfer(std::move(flow_file), Invalid);
+ return;
+ }
+
+ // SAFETY: checkRequiredAttributes already checks for ENRICHMENT_GROUP_ID
+ const std::string group_id = *session.getAttribute(flow_file, ENRICHMENT_GROUP_ID);
+
+ 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);
+ session.transfer(std::move(flow_file), Invalid);
+ session.transfer(session.unstash(std::move(previous_node.mapped())), Invalid);
+ return;
+ }
+
+ if (const auto pair_node = pair_map.extract(group_id)) {
+ logger_->log_trace("Match found for {}", group_id);
+ api::core::FlowFile pair = session.unstash(std::move(pair_node.mapped()));
+ auto [original, enrichment] =
+ (*role == EnrichmentRole::ORIGINAL) ? std::tie(flow_file, pair) : std::tie(pair, flow_file);
+ join(original, enrichment, session);
+ } else {
+ my_map.insert({group_id, session.stash(std::move(flow_file))});
+ if (time_out_tracker_) {
+ time_out_tracker_->track(std::move(group_id), current_time);
+ }
+ }
+}
+
+minifi_status JoinEnrichmentAttributes::onTriggerImpl(api::core::ProcessContext&, api::core::ProcessSession& session) {
+ const auto current_time = std::chrono::steady_clock::now();
+ uint64_t processed = 0;
+ while (auto flow_file = session.get()) {
+ handleFlowFile(std::move(flow_file), session, current_time);
+ if (max_batch_size_ && ++processed >= max_batch_size_) {
+ break;
+ }
+ }
+
+ if (time_out_tracker_) {
+ 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)) {
+ session.transfer(session.unstash(std::move(timed_out_node.mapped())), TimeoutRelationship);
+ }
+ };
+ removeFromMap(originals_);
+ removeFromMap(enrichments_);
+ }
+ }
+ return MINIFI_STATUS_SUCCESS;
+}
+
+} // namespace org::apache::nifi::minifi::enrichment
diff --git a/extensions/enrichment/JoinEnrichmentAttributes.h b/extensions/enrichment/JoinEnrichmentAttributes.h
new file mode 100644
index 000000000..e2c457407
--- /dev/null
+++ b/extensions/enrichment/JoinEnrichmentAttributes.h
@@ -0,0 +1,138 @@
+/**
+ * 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 <array>
+#include <chrono>
+#include <deque>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "EnrichmentUtils.h"
+#include "api/core/ProcessorImpl.h"
+#include "api/utils/Export.h"
+#include "core/PropertyDefinitionBuilder.h"
+#include "minifi-cpp/core/Annotation.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+
+namespace org::apache::nifi::minifi::enrichment {
+
+namespace join_enrichment_attributes {
+class TimeOutTracker {
+ public:
+ explicit TimeOutTracker(std::chrono::steady_clock::duration timeout) : time_out_(timeout) {
+ }
+ TimeOutTracker(const TimeOutTracker&) = delete;
+ TimeOutTracker& operator=(const TimeOutTracker&) = delete;
+ TimeOutTracker(TimeOutTracker&&) = delete;
+ TimeOutTracker& operator=(TimeOutTracker&&) = delete;
+ ~TimeOutTracker() = default;
+
+ void track(std::string id, std::chrono::steady_clock::time_point timestamp) {
+ queue_.emplace_back(timestamp, std::move(id));
+ }
+
+ std::vector<std::string> getTimedOutFlowFiles(std::chrono::steady_clock::time_point current_time) {
+ std::vector<std::string> result;
+ // Even with 0 time_out_, we won't return just added FlowFiles
+ while (!queue_.empty() && queue_.front().timestamp + time_out_ < current_time) {
+ result.push_back(std::move(queue_.front().group_name));
+ queue_.pop_front();
+ }
+ return result;
+ }
+
+ private:
+ struct TimeStampedGroup {
+ std::chrono::steady_clock::time_point timestamp;
+ std::string group_name;
+ };
+
+ std::chrono::steady_clock::duration time_out_;
+ std::deque<TimeStampedGroup> queue_;
+};
+} // namespace join_enrichment_attributes
+
+using StoredFlowFileMap = std::unordered_map<std::string, api::core::StashedFlowFile, utils::string::transparent_string_hash, std::equal_to<>>;
+
+class JoinEnrichmentAttributes : public api::core::ProcessorImpl {
+ public:
+ using ProcessorImpl::ProcessorImpl;
+
+ EXTENSIONAPI static constexpr const char* Description =
+ "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes "
+ "from both of them (prioritizing Enrichment's).";
+
+ 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"};
+ 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."};
+ EXTENSIONAPI static constexpr auto TimeoutRelationship = core::RelationshipDefinition{"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."};
+
+ EXTENSIONAPI static constexpr auto MaxBatchSize =
+ core::PropertyDefinitionBuilder<>::createProperty("Max Batch Size")
+ .withDescription("The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.")
+ .withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR)
+ .build();
+
+ EXTENSIONAPI static constexpr auto TimeoutProperty =
+ core::PropertyDefinitionBuilder<>::createProperty("Timeout")
+ .withDescription(
+ "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the "
+ "first FlowFile will be routed to the 'timeout' relationship.")
+ .withValidator(core::StandardPropertyValidators::TIME_PERIOD_VALIDATOR)
+ .isRequired(false)
+ .build();
+
+ EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 2>{TimeoutProperty, MaxBatchSize};
+ EXTENSIONAPI static constexpr auto Relationships = std::array{Invalid, Joined, Original, TimeoutRelationship};
+
+ EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
+ EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
+ EXTENSIONAPI static constexpr auto InputRequirement = core::annotation::Input::INPUT_REQUIRED;
+ EXTENSIONAPI static constexpr bool IsSingleThreaded = true;
+
+ protected:
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+
+ private:
+ enum class EnrichmentRole {
+ ORIGINAL,
+ ENRICHMENT,
+ };
+
+ void handleFlowFile(api::core::FlowFile flow_file, api::core::ProcessSession& session, std::chrono::steady_clock::time_point current_time);
+ static void join(api::core::FlowFile& original, api::core::FlowFile& enrichment, api::core::ProcessSession& session);
+
+ std::optional<join_enrichment_attributes::TimeOutTracker> time_out_tracker_;
+ StoredFlowFileMap originals_;
+ StoredFlowFileMap enrichments_;
+ std::optional<uint64_t> max_batch_size_;
+};
+} // namespace org::apache::nifi::minifi::enrichment
diff --git a/extensions/enrichment/tests/CMakeLists.txt b/extensions/enrichment/tests/CMakeLists.txt
new file mode 100644
index 000000000..4629067d5
--- /dev/null
+++ b/extensions/enrichment/tests/CMakeLists.txt
@@ -0,0 +1,40 @@
+#
+# 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.
+#
+
+file(GLOB ENRICHMENT_TESTS "*.cpp")
+
+SET(EXTENSIONS_TEST_COUNT 0)
+FOREACH(testfile ${ENRICHMENT_TESTS})
+ get_filename_component(testfilename "${testfile}" NAME_WE)
+ add_minifi_executable(${testfilename} "${testfile}")
+ target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/libminifi/include")
+ target_include_directories(${testfilename} BEFORE PRIVATE "${CMAKE_SOURCE_DIR}/extensions/enrichment/processors")
+ createTests(${testfilename})
+ target_link_libraries(${testfilename} Catch2WithMain)
+ target_link_libraries(${testfilename} minifi-enrichment)
+ target_link_libraries(${testfilename} minifi-standard-processors)
+ target_link_libraries(${testfilename} libminifi-c-unittest)
+ target_compile_definitions("${testfilename}" PRIVATE TZ_DATA_DIR="${CMAKE_BINARY_DIR}/tzdata")
+ target_compile_definitions(${testfilename} PRIVATE "MINIFI_EXTENSION_GROUP_NAME=org.apache.nifi.minifi.test")
+
+ MATH(EXPR EXTENSIONS_TEST_COUNT "${EXTENSIONS_TEST_COUNT}+1")
+ add_test(NAME ${testfilename} COMMAND ${testfilename} WORKING_DIRECTORY ${TEST_DIR})
+ set_tests_properties("${testfilename}" PROPERTIES LABELS "enrichment;memchecked")
+ENDFOREACH()
+message("-- Finished building ${EXTENSIONS_TEST_COUNT} enrichment related test file(s)...")
diff --git a/extensions/enrichment/tests/ForkEnrichmentTests.cpp b/extensions/enrichment/tests/ForkEnrichmentTests.cpp
new file mode 100644
index 000000000..320b15ec1
--- /dev/null
+++ b/extensions/enrichment/tests/ForkEnrichmentTests.cpp
@@ -0,0 +1,70 @@
+/**
+ *
+ * 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.
+ */
+#include "CProcessorTestUtils.h"
+#include "ForkEnrichment.h"
+#include "unit/Catch.h"
+#include "unit/ProcessorUtils.h"
+#include "unit/SingleProcessorTestController.h"
+
+namespace org::apache::nifi::minifi::enrichment::test {
+TEST_CASE("Fork Enrichment processor works") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<ForkEnrichment>(
+ core::ProcessorMetadata{utils::Identifier{}, "ForkEnrichment", logging::LoggerFactory<ForkEnrichment>::getLogger()}));
+ const auto trigger_result = test_controller.trigger("test_content");
+ REQUIRE(trigger_result.contains(ForkEnrichment::Original));
+ REQUIRE(trigger_result.contains(ForkEnrichment::Enrichment));
+ const auto original_results = trigger_result.at(ForkEnrichment::Original);
+ const auto enrichment_results = trigger_result.at(ForkEnrichment::Enrichment);
+ REQUIRE(original_results.size() == 1);
+ REQUIRE(enrichment_results.size() == 1);
+
+ const auto original_content = test_controller.plan->getContent(original_results.at(0));
+ const auto enrichment_content = test_controller.plan->getContent(enrichment_results.at(0));
+
+ 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::EnrichmentGroupId.name) ==
+ enrichment_results.at(0)->getAttribute(ForkEnrichment::EnrichmentGroupId.name));
+}
+
+TEST_CASE("ForkEnrichment no max batch size") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<ForkEnrichment>(
+ core::ProcessorMetadata{utils::Identifier{}, "ForkEnrichment", logging::LoggerFactory<ForkEnrichment>::getLogger()}));
+ const auto trigger_result = test_controller.trigger({{.content = "one"}, {.content = "two"}, {.content = "three"}});
+ const auto original_results = trigger_result.at(ForkEnrichment::Original);
+ const auto enrichment_results = trigger_result.at(ForkEnrichment::Enrichment);
+ REQUIRE(original_results.size() == 3);
+ REQUIRE(enrichment_results.size() == 3);
+}
+
+TEST_CASE("ForkEnrichment max batch size 2") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<ForkEnrichment>(
+ core::ProcessorMetadata{utils::Identifier{}, "ForkEnrichment", logging::LoggerFactory<ForkEnrichment>::getLogger()}));
+ const auto proc = test_controller.getProcessor();
+ CHECK(test_controller.plan->setProperty(proc, ForkEnrichment::MaxBatchSize.name, "2"));
+ const auto trigger_result = test_controller.trigger({{.content = "one"}, {.content = "two"}, {.content = "three"}});
+ const auto original_results = trigger_result.at(ForkEnrichment::Original);
+ const auto enrichment_results = trigger_result.at(ForkEnrichment::Enrichment);
+ REQUIRE(original_results.size() == 2);
+ REQUIRE(enrichment_results.size() == 2);
+}
+} // namespace org::apache::nifi::minifi::enrichment::test
diff --git a/extensions/enrichment/tests/JoinEnrichmentAttributesTests.cpp b/extensions/enrichment/tests/JoinEnrichmentAttributesTests.cpp
new file mode 100644
index 000000000..3ec640abd
--- /dev/null
+++ b/extensions/enrichment/tests/JoinEnrichmentAttributesTests.cpp
@@ -0,0 +1,140 @@
+/**
+ *
+ * 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.
+ */
+
+#include <thread>
+
+#include "CProcessorTestUtils.h"
+#include "EnrichmentUtils.h"
+#include "JoinEnrichmentAttributes.h"
+#include "unit/Catch.h"
+#include "unit/ProcessorUtils.h"
+#include "unit/SingleProcessorTestController.h"
+
+namespace org::apache::nifi::minifi::enrichment::test {
+TEST_CASE("JoinEnrichmentAttributes input without appropriate attributes") {
+ minifi::test::SingleProcessorTestController controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto trigger_result = controller.trigger("test_content");
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Invalid).size() == 1);
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Original).empty());
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::TimeoutRelationship).empty());
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Joined).empty());
+}
+
+TEST_CASE("JoinEnrichmentAttributes invalid role") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto trigger_result = test_controller.trigger(minifi::test::InputFlowFileData{.content = "first",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "FIREBIRD"}}});
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Invalid).size() == 1);
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Original).empty());
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::TimeoutRelationship).empty());
+ CHECK(trigger_result.at(JoinEnrichmentAttributes::Joined).empty());
+}
+
+TEST_CASE("JoinEnrichmentAttributes same id same role same session") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto trigger = test_controller.trigger(
+ {minifi::test::InputFlowFileData{.content = "first",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "ORIGINAL"}}},
+ minifi::test::InputFlowFileData{.content = "second",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "ORIGINAL"}}}});
+ CHECK(trigger.at(JoinEnrichmentAttributes::Invalid).size() == 2);
+ CHECK(trigger.at(JoinEnrichmentAttributes::Original).empty());
+ CHECK(trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).empty());
+ CHECK(trigger.at(JoinEnrichmentAttributes::Joined).empty());
+}
+
+TEST_CASE("JoinEnrichmentAttributes same id same role different session") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto first_trigger = test_controller.trigger(minifi::test::InputFlowFileData{.content = "first",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "ORIGINAL"}}});
+ // 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(); }));
+
+ const auto second_trigger = test_controller.trigger(minifi::test::InputFlowFileData{.content = "second",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "ORIGINAL"}}});
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Invalid).size() == 2);
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Original).empty());
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).empty());
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Joined).empty());
+}
+
+TEST_CASE("JoinEnrichmentAttributes same id diff role different session") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto first_trigger = test_controller.trigger(minifi::test::InputFlowFileData{.content = "first",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "ORIGINAL"}, {"first_attr", "1"}}});
+ // 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(); }));
+
+ const auto second_trigger = test_controller.trigger(minifi::test::InputFlowFileData{.content = "second",
+ .attributes = {{std::string{ENRICHMENT_GROUP_ID}, "foo"}, {std::string{ENRICHMENT_ROLE}, "ENRICHMENT"}, {"second_attr", "2"}}});
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Original).size() == 2);
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Invalid).empty());
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).empty());
+ REQUIRE(second_trigger.at(JoinEnrichmentAttributes::Joined).size() == 1);
+
+ const auto joined_content = test_controller.plan->getContent(second_trigger.at(JoinEnrichmentAttributes::Joined).at(0));
+ const auto joined_attrs = second_trigger.at(JoinEnrichmentAttributes::Joined).at(0)->getAttributes();
+
+ CHECK(joined_content == "first");
+ CHECK(joined_attrs.at(std::string{ENRICHMENT_GROUP_ID}) == "foo");
+ 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 test timeout") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ 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"}, {"first_attr", "1"}}});
+ // 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);
+ 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());
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::Invalid).empty());
+ CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 1);
+ REQUIRE(second_trigger.at(JoinEnrichmentAttributes::Joined).empty());
+}
+
+TEST_CASE("JoinEnrichmentAttributes no max batch size") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto trigger_result = test_controller.trigger({{.content = "one"}, {.content = "two"}, {.content = "three"}});
+ REQUIRE(trigger_result.at(JoinEnrichmentAttributes::Invalid).size() == 3);
+}
+
+TEST_CASE("JoinEnrichmentAttributes max batch size 2") {
+ minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_custom_c_processor<JoinEnrichmentAttributes>(
+ core::ProcessorMetadata{utils::Identifier{}, "JoinEnrichmentAttributes", logging::LoggerFactory<JoinEnrichmentAttributes>::getLogger()}));
+ const auto proc = test_controller.getProcessor();
+ CHECK(test_controller.plan->setProperty(proc, JoinEnrichmentAttributes::MaxBatchSize.name, "2"));
+ const auto trigger_result = test_controller.trigger({{.content = "one"}, {.content = "two"}, {.content = "three"}});
+ REQUIRE(trigger_result.at(JoinEnrichmentAttributes::Invalid).size() == 2);
+}
+} // namespace org::apache::nifi::minifi::enrichment::test
diff --git a/extensions/enrichment/tests/features/enrichment.feature b/extensions/enrichment/tests/features/enrichment.feature
new file mode 100644
index 000000000..1401abc46
--- /dev/null
+++ b/extensions/enrichment/tests/features/enrichment.feature
@@ -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.
+
+@MINIFI_EXTENSION_ENRICHMENT @SUPPORTS_WINDOWS
+Feature: ForkEnrichment and JoinEnrichmentAttributes
+
+ Scenario: Merges correctly
+ Given a GenerateFlowFile processor with the "Custom Text" property set to "original_${literal("content")}"
+ And the scheduling period of the GenerateFlowFile processor is set to "1 hour"
+ 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 a ForkEnrichment processor
+ And a JoinEnrichmentAttributes processor
+
+ And a ReplaceText processor with the "Evaluation Mode" property set to "Entire text"
+ 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 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 "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 "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: "original_content" in less than 1 second
diff --git a/extensions/enrichment/tests/features/enrichment_restart.feature b/extensions/enrichment/tests/features/enrichment_restart.feature
new file mode 100644
index 000000000..36d329101
--- /dev/null
+++ b/extensions/enrichment/tests/features/enrichment_restart.feature
@@ -0,0 +1,71 @@
+# 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.
+
+@MINIFI_EXTENSION_ENRICHMENT
+Feature: JoinEnrichmentAttributes recovers released FlowFiles after an agent restart
+
+ # JoinEnrichmentAttributes releases the first half of a pair and holds it in memory until the
+ # other half arrives. The released FlowFile is not routed, persisted-as-moved, nor deleted on
+ # commit, so its record stays in the (default, RocksDB-backed) flow file repository. When the
+ # agent is restarted the held FlowFile must be recovered from disk and re-enqueued, so that the
+ # join still completes once its pair finally arrives.
+ #
+ # The two halves are produced independently with GetFile + UpdateAttribute (instead of
+ # ForkEnrichment) so their arrival can be separated across the restart: the ORIGINAL is present
+ # at startup, and its ENRICHMENT pair is only delivered after the restart.
+
+ Scenario: A released FlowFile held across an agent restart is recovered and joined with its late-arriving pair
+ Given a GetFile processor with the name "GetOriginal" and the "Input Directory" property set to "/tmp/original_input"
+ And the scheduling period of the GetOriginal processor is set to "100 ms"
+ And a UpdateAttribute processor with the name "TagOriginal" and the "enrichment.role" property set to "ORIGINAL"
+ And the "enrichment.group.id" property of the TagOriginal processor is set to "group-1"
+
+ And a GetFile processor with the name "GetEnrichment" and the "Input Directory" property set to "/tmp/enrichment_input"
+ And the scheduling period of the GetEnrichment processor is set to "100 ms"
+ And a UpdateAttribute processor with the name "TagEnrichment" and the "enrichment.role" property set to "ENRICHMENT"
+ And the "enrichment.group.id" property of the TagEnrichment processor is set to "group-1"
+
+ And a JoinEnrichmentAttributes processor
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+
+ And the "success" relationship of the GetOriginal processor is connected to the TagOriginal
+ And the "success" relationship of the TagOriginal processor is connected to the JoinEnrichmentAttributes
+ And the "success" relationship of the GetEnrichment processor is connected to the TagEnrichment
+ And the "success" relationship of the TagEnrichment processor is connected to the JoinEnrichmentAttributes
+ And the "joined" relationship of the JoinEnrichmentAttributes processor is connected to the PutFile
+
+ And JoinEnrichmentAttributes's original relationship is auto-terminated
+ And JoinEnrichmentAttributes's invalid relationship is auto-terminated
+ And JoinEnrichmentAttributes's timeout relationship is auto-terminated
+ And PutFile's success relationship is auto-terminated
+ And PutFile's failure relationship is auto-terminated
+
+ # Only the ORIGINAL half exists at startup; the ENRICHMENT half arrives after the restart.
+ And a directory at "/tmp/original_input" has a file with the content "original_content"
+
+ When the MiNiFi instance starts up
+ # JoinEnrichmentAttributes gets the ORIGINAL, releases it and holds it waiting for its pair,
+ # so nothing is joined yet.
+ Then no files are placed in the "/tmp/output" directory in 5 seconds of running time
+
+ # Graceful stop destroys JoinEnrichmentAttributes while it still holds the released ORIGINAL,
+ # then restart brings the agent back with the persistent repositories intact.
+ When MiNiFi is stopped
+ And MiNiFi is restarted
+
+ # The released ORIGINAL is recovered from the flow file repository and re-enqueued. Once its
+ # ENRICHMENT pair arrives, the join produces a single FlowFile with the ORIGINAL's content.
+ And a file with the content "enrichment_content" is placed in "/tmp/enrichment_input"
+ Then a single file with the content "original_content" is placed in the "/tmp/output" directory in less than 60 seconds
diff --git a/extensions/enrichment/tests/features/environment.py b/extensions/enrichment/tests/features/environment.py
new file mode 100644
index 000000000..079c95b50
--- /dev/null
+++ b/extensions/enrichment/tests/features/environment.py
@@ -0,0 +1,28 @@
+# 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.
+
+from minifi_behave.core.hooks import (
+ common_after_scenario,
+ common_before_scenario,
+)
+from minifi_behave.core.minifi_test_context import MinifiTestContext
+
+
+def before_scenario(context: MinifiTestContext, scenario):
+ common_before_scenario(context, scenario)
+
+
+def after_scenario(context, scenario):
+ common_after_scenario(context, scenario)
diff --git a/extensions/enrichment/tests/features/steps/steps.py b/extensions/enrichment/tests/features/steps/steps.py
new file mode 100644
index 000000000..ff5817598
--- /dev/null
+++ b/extensions/enrichment/tests/features/steps/steps.py
@@ -0,0 +1,21 @@
+# 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.
+
+from minifi_behave.steps import (
+ checking_steps, # noqa: F401
+ configuration_steps, # noqa: F401
+ core_steps, # noqa: F401
+ flow_building_steps, # noqa: F401
+)