diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json index bcb495d3e9..b018058b5c 100644 --- a/.github/references/ubuntu_22_04_clang_arm_manifest.json +++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json @@ -3637,6 +3637,34 @@ "supportsDynamicProperties": "false", "type": "org.apache.nifi.minifi.processors.FocusArchiveEntry" }, + { + "propertyDescriptors": { + "Max Batch Size": { + "name": "Max Batch Size", + "description": "The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "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' relationship.", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.standard.ForkEnrichment" + }, { "propertyDescriptors": { "Batch Size": { @@ -4354,6 +4382,50 @@ "supportsDynamicProperties": "false", "type": "org.apache.nifi.minifi.processors.InvokeHTTP" }, + { + "propertyDescriptors": { + "Max Batch Size": { + "name": "Max Batch Size", + "description": "The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "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 4c7705b2a6..68487069c5 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,35 @@ 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 | +|----------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------| +| Max Batch Size | | | The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once. | + +### 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 +1308,31 @@ 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. | +| Max Batch Size | | | The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once. | + +### 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/behave_framework/src/minifi_behave/steps/flow_building_steps.py b/behave_framework/src/minifi_behave/steps/flow_building_steps.py index 40414a9e90..d0415f51cf 100644 --- a/behave_framework/src/minifi_behave/steps/flow_building_steps.py +++ b/behave_framework/src/minifi_behave/steps/flow_building_steps.py @@ -40,6 +40,8 @@ def transient_flow_with_logondestructionprocessor(context: MinifiTestContext): Processor("LogOnDestructionProcessor", "LogOnDestructionProcessor")) +@given( + 'an {processor_type} processor with the name "{processor_name}" and the "{property_name}" property set to "{property_value}"') @given( 'a {processor_type} processor with the name "{processor_name}" and the "{property_name}" property set to "{property_value}"') def processor_with_name_and_property(context: MinifiTestContext, processor_type: str, processor_name: str, property_name: str, @@ -49,12 +51,14 @@ def processor_with_name_and_property(context: MinifiTestContext, processor_type: context.get_or_create_default_minifi_container().flow_definition.add_processor(processor) +@step('an {processor_type} processor with the "{property_name}" property set to "{property_value}"') @step('a {processor_type} processor with the "{property_name}" property set to "{property_value}"') def processor_with_property(context: MinifiTestContext, processor_type: str, property_name: str, property_value: str): context.execute_steps( f'Given a {processor_type} processor with the name "{processor_type}" and the "{property_name}" property set to "{property_value}"') +@step('an {processor_type} processor with the "{property_name}" property set to "{property_value}" in the "{minifi_container_name}" flow') @step('a {processor_type} processor with the "{property_name}" property set to "{property_value}" in the "{minifi_container_name}" flow') def processor_with_property_in_minifi_flow(context: MinifiTestContext, processor_type: str, property_name: str, property_value: str, minifi_container_name: str): processor = Processor(processor_type, processor_type) @@ -62,6 +66,7 @@ def processor_with_property_in_minifi_flow(context: MinifiTestContext, processor context.get_or_create_minifi_container(minifi_container_name).flow_definition.add_processor(processor) +@step('an {processor_type} processor with the "{property_name}" property set to "{property_value}" in the NiFi flow') @step('a {processor_type} processor with the "{property_name}" property set to "{property_value}" in the NiFi flow') def processor_with_property_in_nifi_flow(context: MinifiTestContext, processor_type: str, property_name: str, property_value: str): processor = Processor(processor_type, processor_type) @@ -69,30 +74,35 @@ def processor_with_property_in_nifi_flow(context: MinifiTestContext, processor_t context.containers["nifi"].flow_definition.add_processor(processor) +@given('an {processor_type} processor with the name "{processor_name}" in the "{minifi_container_name}" flow') @given('a {processor_type} processor with the name "{processor_name}" in the "{minifi_container_name}" flow') def processor_with_name_in_minifi_flow(context: MinifiTestContext, processor_type: str, processor_name: str, minifi_container_name: str): processor = Processor(processor_type, processor_name) context.get_or_create_minifi_container(minifi_container_name).flow_definition.add_processor(processor) +@given('an {processor_type} processor with the name "{processor_name}"') @given('a {processor_type} processor with the name "{processor_name}"') def processor_with_name(context: MinifiTestContext, processor_type: str, processor_name: str): context.execute_steps( f'given a {processor_type} processor with the name "{processor_name}" in the "{DEFAULT_MINIFI_CONTAINER_NAME}" flow') +@given("an {processor_type} processor in the \"{minifi_container_name}\" flow") @given("a {processor_type} processor in the \"{minifi_container_name}\" flow") def processor_in_minifi_flow(context: MinifiTestContext, processor_type: str, minifi_container_name: str): processor = Processor(processor_type, processor_type) context.get_or_create_minifi_container(minifi_container_name).flow_definition.add_processor(processor) +@given("an {processor_type} processor in the NiFi flow") @given("a {processor_type} processor in the NiFi flow") def processor_in_nifi_flow(context: MinifiTestContext, processor_type: str): processor = Processor(processor_type, processor_type) context.containers["nifi"].flow_definition.add_processor(processor) +@given("an {processor_type} processor") @given("a {processor_type} processor") def processor_setup(context: MinifiTestContext, processor_type: str): context.execute_steps(f'given a {processor_type} processor in the "{DEFAULT_MINIFI_CONTAINER_NAME}" flow') diff --git a/extensions/standard-processors/processors/ForkEnrichment.cpp b/extensions/standard-processors/processors/ForkEnrichment.cpp new file mode 100644 index 0000000000..9cc3b34de6 --- /dev/null +++ b/extensions/standard-processors/processors/ForkEnrichment.cpp @@ -0,0 +1,61 @@ +/** + * 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 "core/Resource.h" +#include "minifi-cpp/core/ProcessSession.h" +#include "utils/ProcessorConfigUtils.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) { + max_batch_size_ = utils::parseOptionalU64Property(context, MaxBatchSize); + if (max_batch_size_ && *max_batch_size_ == 0) { + throw Exception(PROCESSOR_EXCEPTION, "Max Batch Size property is invalid"); + } + + ProcessorImpl::onSchedule(context, session_factory); +} + +void ForkEnrichment::onTrigger(core::ProcessContext&, core::ProcessSession& session) { + uint64_t processed = 0; + while (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); + if (max_batch_size_ && ++processed >= max_batch_size_) { + break; + } + } +} + +REGISTER_RESOURCE(ForkEnrichment, Processor); +} // namespace org::apache::nifi::minifi::standard diff --git a/extensions/standard-processors/processors/ForkEnrichment.h b/extensions/standard-processors/processors/ForkEnrichment.h new file mode 100644 index 0000000000..a8f103bfc0 --- /dev/null +++ b/extensions/standard-processors/processors/ForkEnrichment.h @@ -0,0 +1,76 @@ +/** + * 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 + +#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 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{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{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; + + private: + std::optional max_batch_size_; +}; +} // 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 0000000000..258e625c88 --- /dev/null +++ b/extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp @@ -0,0 +1,160 @@ +/** + * 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 "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); + } + max_batch_size_ = utils::parseOptionalU64Property(context, MaxBatchSize); + if (max_batch_size_ && *max_batch_size_ == 0) { + throw Exception(PROCESSOR_EXCEPTION, "Max Batch Size property is invalid"); + } + 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& original, const std::shared_ptr& enrichment, + core::ProcessSession& session) const { + 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 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; + } + + const auto role = flow_file->getAttribute(ENRICHMENT_ROLE) | utils::toExpected(make_error_code(core::AttributeErrorCode::MissingAttribute)) | + utils::andThen(parsing::parseEnum); + 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"); + auto [original, + enrichment] = (*role == EnrichmentRole::ORIGINAL) ? std::tie(flow_file, pair_node.mapped()) : std::tie(pair_node.mapped(), flow_file); + join(original, enrichment, session); + } 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 (auto flow_file : flow_file_store_.getNewFlowFiles()) { + handleFlowFile(std::move(flow_file), session, current_time); + } + uint64_t processed = 0; + while (auto flow_file = session.get()) { + session_flow_files_.push_back(flow_file->getUUID()); + 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 = [&](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& 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.h b/extensions/standard-processors/processors/JoinEnrichmentAttributes.h new file mode 100644 index 0000000000..1dbebaf2e8 --- /dev/null +++ b/extensions/standard-processors/processors/JoinEnrichmentAttributes.h @@ -0,0 +1,139 @@ +/** + * 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 +#include + +#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 getTimedOutFlowFiles(std::chrono::steady_clock::time_point current_time) { + std::vector 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 queue_; +}; +} // namespace join_enrichment_attributes + +using MapType = std::unordered_map, 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 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{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; + + 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& flowFile) override; + + private: + enum class EnrichmentRole { + ORIGINAL, + ENRICHMENT, + }; + + void handleFlowFile(std::shared_ptr flow_file, core::ProcessSession& session, std::chrono::steady_clock::time_point current_time); + void join(const std::shared_ptr& original, const std::shared_ptr& enrichment, core::ProcessSession& session) const; + + core::FlowFileStore flow_file_store_; + // We need to track current session's FlowFiles (we cant add those) + std::vector session_flow_files_; + + std::optional time_out_tracker_; + MapType originals_; + MapType enrichments_; + std::optional max_batch_size_; +}; +} // 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 0000000000..91e92d3bf7 --- /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 0000000000..a81ae14080 --- /dev/null +++ b/extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp @@ -0,0 +1,66 @@ +/** + * + * 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 "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")); + 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_processor("ForkEnrichment")); + 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_processor("ForkEnrichment")); + 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::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 0000000000..8a138198db --- /dev/null +++ b/extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp @@ -0,0 +1,134 @@ +/** + * + * 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 "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")); + const auto trigger_result = test_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_processor("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).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_processor("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).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_processor("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).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_processor("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).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_processor("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).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_processor("JoinEnrichmentAttributes")); + 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_processor("JoinEnrichmentAttributes")); + 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::standard::test diff --git a/extensions/standard-processors/utils/EnrichmentUtils.h b/extensions/standard-processors/utils/EnrichmentUtils.h new file mode 100644 index 0000000000..e711585f0c --- /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";