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
commit 5c21e0a97251c21bade7bbb454f2f72ec0d5967c Author: Martin Zink <[email protected]> AuthorDate: Wed Jul 8 10:30:49 2026 +0200 MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes processors --- .../ubuntu_22_04_clang_arm_manifest.json | 54 ++++++++ PROCESSORS.md | 54 ++++++++ .../processors/ForkEnrichment.cpp | 51 +++++++ .../processors/ForkEnrichment.hpp | 69 ++++++++++ .../processors/JoinEnrichmentAttributes.cpp | 152 +++++++++++++++++++++ .../processors/JoinEnrichmentAttributes.hpp | 133 ++++++++++++++++++ .../tests/features/enrichment.feature | 47 +++++++ .../tests/unit/ForkEnrichmentTests.cpp | 46 +++++++ .../tests/unit/JoinEnrichmentAttributesTests.cpp | 120 ++++++++++++++++ .../standard-processors/utils/EnrichmentUtils.h | 21 +++ 10 files changed, 747 insertions(+) diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json index bcb495d3e..6dd05d4e5 100644 --- a/.github/references/ubuntu_22_04_clang_arm_manifest.json +++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json @@ -3637,6 +3637,24 @@ "supportsDynamicProperties": "false", "type": "org.apache.nifi.minifi.processors.FocusArchiveEntry" }, + { + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "false", + "supportedRelationships": [ + { + "name": "enrichment", + "description": "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + }, + { + "name": "original", + "description": "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + } + ], + "typeDescription": "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' rela [...] + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.standard.ForkEnrichment" + }, { "propertyDescriptors": { "Batch Size": { @@ -4354,6 +4372,42 @@ "supportsDynamicProperties": "false", "type": "org.apache.nifi.minifi.processors.InvokeHTTP" }, + { + "propertyDescriptors": { + "Timeout": { + "name": "Timeout", + "description": "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.", + "validator": "TIME_PERIOD_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "true", + "supportedRelationships": [ + { + "name": "invalid", + "description": "Any FlowFiles without the requisite attributes will be routed here" + }, + { + "name": "joined", + "description": "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship" + }, + { + "name": "original", + "description": "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." + }, + { + "name": "timeout", + "description": "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." + } + ], + "typeDescription": "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).", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.standard.JoinEnrichmentAttributes" + }, { "propertyDescriptors": { "Jolt Specification": { diff --git a/PROCESSORS.md b/PROCESSORS.md index 4c7705b2a..f578ddd81 100644 --- a/PROCESSORS.md +++ b/PROCESSORS.md @@ -46,12 +46,14 @@ limitations under the License. - [FetchSFTP](#FetchSFTP) - [FetchSmb](#FetchSmb) - [FocusArchiveEntry](#FocusArchiveEntry) +- [ForkEnrichment](#ForkEnrichment) - [GenerateFlowFile](#GenerateFlowFile) - [GetCouchbaseKey](#GetCouchbaseKey) - [GetFile](#GetFile) - [GetTCP](#GetTCP) - [HashContent](#HashContent) - [InvokeHTTP](#InvokeHTTP) +- [JoinEnrichmentAttributes](#JoinEnrichmentAttributes) - [JoltTransformJSON](#JoltTransformJSON) - [ListAzureBlobStorage](#ListAzureBlobStorage) - [ListAzureDataLakeStorage](#ListAzureDataLakeStorage) @@ -1067,6 +1069,34 @@ In the list below, the names of required properties appear in bold. Any other pr | success | success operational on the flow record | +## ForkEnrichment + +### 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. + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|------|---------------|------------------|-------------| + +### Relationships + +| Name | Description | +|------------|------------------------------------------------------------------------------------------------------------| +| enrichment | A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes. | +| original | The incoming FlowFile will be routed to this relationship, after adding appropriate attributes. | + +### Output Attributes + +| Attribute | Relationship | Description | +|---------------------|----------------------|---------------------------------------------------------------------------------------------------| +| enrichment.role | enrichment, original | The role to use for enrichment. This will either be ORIGINAL or ENRICHMENT. | +| enrichment.group.id | enrichment, original | The Group ID to use in order to correlate the 'original' FlowFile with the 'enrichment' FlowFile. | + + ## GenerateFlowFile ### Description @@ -1277,6 +1307,30 @@ In the list below, the names of required properties appear in bold. Any other pr | invokehttp.tx.id | success, response, retry, no retry | The transaction ID that is returned after reading the response | +## JoinEnrichmentAttributes + +### 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). + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|---------|---------------|------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Timeout | | | 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. | + +### Relationships + +| 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 | +| 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. | + + ## JoltTransformJSON ### Description diff --git a/extensions/standard-processors/processors/ForkEnrichment.cpp b/extensions/standard-processors/processors/ForkEnrichment.cpp new file mode 100644 index 000000000..b2df6022b --- /dev/null +++ b/extensions/standard-processors/processors/ForkEnrichment.cpp @@ -0,0 +1,51 @@ +/** + * 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.hpp" + +#include "core/Resource.h" +#include "minifi-cpp/core/ProcessSession.h" + +namespace org::apache::nifi::minifi::standard { +void ForkEnrichment::initialize() { + setSupportedProperties(Properties); + setSupportedRelationships(Relationships); + ProcessorImpl::initialize(); +} + +void ForkEnrichment::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) { + ProcessorImpl::onSchedule(context, session_factory); +} + +void ForkEnrichment::onTrigger(core::ProcessContext&, core::ProcessSession& session) { + if (const auto original = session.get()) { + const auto enrichment = session.clone(*original); + + original->setAttribute(ENRICHMENT_ROLE, "ORIGINAL"); + enrichment->setAttribute(ENRICHMENT_ROLE, "ENRICHMENT"); + + const std::string group_id = utils::IdGenerator::getIdGenerator()->generate().to_string(); + original->setAttribute(ENRICHMENT_GROUP_ID, group_id); + enrichment->setAttribute(ENRICHMENT_GROUP_ID, group_id); + + session.transfer(original, Original); + session.transfer(enrichment, Enrichment); + } +} + +REGISTER_RESOURCE(ForkEnrichment, Processor); +} // namespace org::apache::nifi::minifi::standard diff --git a/extensions/standard-processors/processors/ForkEnrichment.hpp b/extensions/standard-processors/processors/ForkEnrichment.hpp new file mode 100644 index 000000000..6dffa2dbd --- /dev/null +++ b/extensions/standard-processors/processors/ForkEnrichment.hpp @@ -0,0 +1,69 @@ +/** + * 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 "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 { + +class ForkEnrichment : public 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 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, 0>{}; + 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}; + + ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS + + void initialize() override; + void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) override; + void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override; +}; +} // namespace org::apache::nifi::minifi::standard diff --git a/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp b/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp new file mode 100644 index 000000000..bcb28d606 --- /dev/null +++ b/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp @@ -0,0 +1,152 @@ +/** + * 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.hpp" + +#include "core/Resource.h" +#include "minifi-cpp/core/ProcessSession.h" +#include "utils/AttributeErrors.h" +#include "utils/EnrichmentUtils.h" +#include "utils/ProcessorConfigUtils.h" + +namespace org::apache::nifi::minifi::standard { +const core::Relationship JoinEnrichmentAttributes::Self("__self__", "Marks the FlowFile to be owned by this processor"); + +void JoinEnrichmentAttributes::initialize() { + setSupportedProperties(Properties); + setSupportedRelationships(Relationships); + ProcessorImpl::initialize(); +} + +void JoinEnrichmentAttributes::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) { + using namespace std::literals::chrono_literals; + if (const auto timeout = utils::parseOptionalDurationProperty(context, TimeoutProperty); timeout && *timeout > 0ms) { + time_out_tracker_.emplace(*timeout); + } + 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(); +} +} // namespace + +void JoinEnrichmentAttributes::join(const std::shared_ptr<core::FlowFile>& one, const std::shared_ptr<core::FlowFile>& two, + core::ProcessSession& session, const EnrichmentRole role_one) const { + auto [original, enrichment] = (role_one == EnrichmentRole::ORIGINAL) ? std::tie(one, two) : std::tie(two, one); + 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); + } + session.transfer(original, Original); + session.transfer(enrichment, Original); + session.transfer(cloned, Joined); +} + +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()); + session.transfer(flow_file, Invalid); + return; + } + + /// SAFETY: we checked that the required attributes are present in checkRequiredAttributes + const auto role = flow_file->getAttribute(ENRICHMENT_ROLE) | utils::toExpected(make_error_code(core::AttributeErrorCode::MissingAttribute)) | + utils::andThen(parsing::parseEnum<EnrichmentRole>); + if (!role) { + logger_->log_warn("{} has invalid role due to {}", flow_file->getId(), role.error()); + session.transfer(flow_file, Invalid); + return; + } + + std::string group_id = *(flow_file->getAttribute(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(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; + } + + if (const auto pair_node = pair_map.extract(group_id)) { + logger_->log_trace("Match found"); + join(flow_file, std::move(pair_node.mapped()), session, *role); + } else { + logger_->log_trace("Missing other half routing {} to Self", flow_file->getId()); + my_map.insert({group_id, flow_file}); + session.transfer(flow_file, Self); + if (time_out_tracker_) { + time_out_tracker_->track(std::move(group_id), current_time); + } + } +} + +void JoinEnrichmentAttributes::onTrigger(core::ProcessContext&, core::ProcessSession& session) { + const auto current_time = std::chrono::steady_clock::now(); + for (const auto& flow_file : flow_file_store_.getNewFlowFiles()) { + handleFlowFile(flow_file, session, current_time); + } + while (const auto flow_file = session.get()) { + session_flow_files_.push_back(flow_file->getUUID()); + handleFlowFile(flow_file, session, current_time); + } + + if (time_out_tracker_) { + for (auto timed_out_group : time_out_tracker_->getTimedOutFlowFiles(current_time)) { + const auto removeFromMap = [&](MapType& 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()); + session.transfer(timed_out_node.mapped(), TimeoutRelationship); + } + }; + removeFromMap(originals_); + removeFromMap(enrichments_); + } + } + + session_flow_files_.clear(); +} + +void JoinEnrichmentAttributes::restore(const std::shared_ptr<core::FlowFile>& flowFile) { + if (!flowFile) { + return; + } + flow_file_store_.put(flowFile); +} + +REGISTER_RESOURCE(JoinEnrichmentAttributes, Processor); + +} // namespace org::apache::nifi::minifi::standard diff --git a/extensions/standard-processors/processors/JoinEnrichmentAttributes.hpp b/extensions/standard-processors/processors/JoinEnrichmentAttributes.hpp new file mode 100644 index 000000000..868adebd2 --- /dev/null +++ b/extensions/standard-processors/processors/JoinEnrichmentAttributes.hpp @@ -0,0 +1,133 @@ +/** + * 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 <memory> +#include <string> + +#include "core/FlowFileStore.h" +#include "core/ProcessorImpl.h" +#include "core/PropertyDefinitionBuilder.h" +#include "minifi-cpp/core/PropertyDefinition.h" +#include "utils/Enum.h" +#include "utils/RegexUtils.h" + +namespace org::apache::nifi::minifi::standard { + +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 MapType = std::unordered_map<std::string, std::shared_ptr<core::FlowFile>, utils::string::transparent_string_hash, std::equal_to<>>; + +class JoinEnrichmentAttributes : public 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 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, 1>{TimeoutProperty}; + 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; + + ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS + + EXTENSIONAPI static const core::Relationship Self; + + void initialize() override; + void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) override; + void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override; + 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, + EnrichmentRole role) const; + + core::FlowFileStore flow_file_store_; + // We need to track current session's FlowFiles (we cant add those) + std::vector<utils::Identifier> session_flow_files_; + + std::optional<join_enrichment_attributes::TimeOutTracker> time_out_tracker_; + MapType originals_; + MapType enrichments_; +}; +} // namespace org::apache::nifi::minifi::standard diff --git a/extensions/standard-processors/tests/features/enrichment.feature b/extensions/standard-processors/tests/features/enrichment.feature new file mode 100644 index 000000000..91e92d3bf --- /dev/null +++ b/extensions/standard-processors/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. + +@CORE @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 seconds + And the Minifi logs contain the following message: "original_content" in less than 1 seconds diff --git a/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp b/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp new file mode 100644 index 000000000..8e9d5664b --- /dev/null +++ b/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp @@ -0,0 +1,46 @@ +/** + * + * 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.hpp" +#include "unit/Catch.h" +#include "unit/ProcessorUtils.h" +#include "unit/SingleProcessorTestController.h" + +namespace org::apache::nifi::minifi::standard::test { +TEST_CASE("Fork Enrichment processor works") { + minifi::test::SingleProcessorTestController test_controller(minifi::test::utils::make_processor<ForkEnrichment>("ForkEnrichment")); + 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)); +} +} // namespace org::apache::nifi::minifi::standard::test diff --git a/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp b/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp new file mode 100644 index 000000000..937e29645 --- /dev/null +++ b/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp @@ -0,0 +1,120 @@ +/** + * + * 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.hpp" +#include "unit/Catch.h" +#include "unit/ProcessorUtils.h" +#include "unit/SingleProcessorTestController.h" +#include "utils/EnrichmentUtils.h" + +namespace org::apache::nifi::minifi::standard::test { +TEST_CASE("JoinEnrichmentAttributes input without appropriate attributes") { + minifi::test::SingleProcessorTestController + test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes")); + const auto trigger_result = test_controller.trigger("test_content"); + CHECK(trigger_result.at(JoinEnrichmentAttributes::Invalid).size() == 1); + CHECK(trigger_result.at(JoinEnrichmentAttributes::Original).size() == 0); + CHECK(trigger_result.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 0); + CHECK(trigger_result.at(JoinEnrichmentAttributes::Joined).size() == 0); +} + +TEST_CASE("JoinEnrichmentAttributes invalid role") { + minifi::test::SingleProcessorTestController + test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes")); + 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).size() == 0); + CHECK(trigger_result.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 0); + CHECK(trigger_result.at(JoinEnrichmentAttributes::Joined).size() == 0); +} + +TEST_CASE("JoinEnrichmentAttributes same id same role same session") { + minifi::test::SingleProcessorTestController + test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes")); + 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).size() == 0); + CHECK(trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 0); + CHECK(trigger.at(JoinEnrichmentAttributes::Joined).size() == 0); +} + +TEST_CASE("JoinEnrichmentAttributes same id same role different session") { + minifi::test::SingleProcessorTestController + test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes")); + 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).size() == 0); + CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 0); + CHECK(second_trigger.at(JoinEnrichmentAttributes::Joined).size() == 0); +} + +TEST_CASE("JoinEnrichmentAttributes same id diff role different session") { + minifi::test::SingleProcessorTestController + test_controller(minifi::test::utils::make_processor<JoinEnrichmentAttributes>("JoinEnrichmentAttributes")); + 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).size() == 0); + CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 0); + 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_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"}, {"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).size() == 0); + CHECK(second_trigger.at(JoinEnrichmentAttributes::Invalid).size() == 0); + CHECK(second_trigger.at(JoinEnrichmentAttributes::TimeoutRelationship).size() == 1); + REQUIRE(second_trigger.at(JoinEnrichmentAttributes::Joined).size() == 0); +} +} // namespace org::apache::nifi::minifi::standard::test diff --git a/extensions/standard-processors/utils/EnrichmentUtils.h b/extensions/standard-processors/utils/EnrichmentUtils.h new file mode 100644 index 000000000..e711585f0 --- /dev/null +++ b/extensions/standard-processors/utils/EnrichmentUtils.h @@ -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. + */ + +#pragma once + +constexpr std::string_view ENRICHMENT_ROLE = "enrichment.role"; +constexpr std::string_view ENRICHMENT_GROUP_ID = "enrichment.group.id";
