Skip to content

MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes #2221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
martinzink wants to merge 1 commit into
base: main
Choose a base branch
from
Open

MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes #2221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 72 additions & 0 deletions .github/references/ubuntu_22_04_clang_arm_manifest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": {
Expand Down Expand Up @@ -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": {
Expand Down
56 changes: 56 additions & 0 deletions PROCESSORS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
10 changes: 10 additions & 0 deletions behave_framework/src/minifi_behave/steps/flow_building_steps.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -49,50 +51,58 @@ 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)
processor.add_property(property_name, property_value)
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)
processor.add_property(property_name, property_value)
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')
Expand Down
61 changes: 61 additions & 0 deletions extensions/standard-processors/processors/ForkEnrichment.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading