Skip to content

MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes processors in rust #2267

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

Draft
martinzink wants to merge 5 commits into
base: minifi_rust_impr_3
Choose a base branch
from
Draft

MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes processors in rust #2267

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1b89551
MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes proc…
martinzink Aug 24, 2026
ea70eaf
cpp stable api enrichment
martinzink Aug 24, 2026
9ba3599
Revert "cpp stable api enrichment"
martinzink Sep 18, 2026
3f5d23e
cargo fmt
martinzink Sep 18, 2026
0e86a73
wip
martinzink Sep 25, 2026
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
166 changes: 166 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 @@ -4144,6 +4144,89 @@
"version": "1.0.0"
}
},
{
"bundles": {
"componentManifest": {
"processors": [
{
"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.enrichment.ForkEnrichment"
},
{
"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.enrichment.JoinEnrichmentAttributes"
}
]
},
"group": "org.apache.nifi.minifi",
"artifact": "minifi-enrichment",
"version": "1.0.0"
}
},
{
"bundles": {
"componentManifest": {
Expand Down Expand Up @@ -12104,6 +12187,89 @@
"version": "1.0.0"
}
},
{
"bundles": {
"componentManifest": {
"processors": [
{
"propertyDescriptors": {
"Batch Size": {
"name": "Batch Size",
"description": "The maximum number of FlowFiles to fork in each trigger",
"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": "minifi_enrichment.processors.fork_enrichment.ForkEnrichment"
},
{
"propertyDescriptors": {
"Batch Size": {
"name": "Batch Size",
"description": "The maximum number of FlowFiles to process in each trigger",
"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": "minifi_enrichment.processors.join_enrichment_attributes.JoinEnrichmentAttributes"
}
]
},
"group": "org.apache.nifi.minifi.rust.enrichment",
"artifact": "minifi_enrichment",
"version": "0.1.0"
}
},
{
"bundles": {
"componentManifest": {
Expand Down
1 change: 0 additions & 1 deletion CONTROLLERS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,4 +393,3 @@ In the list below, the names of required properties appear in bold. Any other pr
| **Pretty Print XML** | false | true<br/>false | Specifies whether or not the XML should be pretty printed |
| **Name of Record Tag** | | | Specifies the name of the XML record tag wrapping the record fields. |
| **Name of Root Tag** | | | Specifies the name of the XML root tag wrapping the record set. |

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
1 change: 1 addition & 0 deletions cmake/MiNiFiOptions.cmake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@ add_minifi_option(ENABLE_CONTROLLER "Enables the build of MiNiFi controller bina
add_minifi_option(ENABLE_LLAMACPP "Enables llama.cpp support." ON)
add_minifi_option(ENABLE_OPC "Instructs the build system to enable the OPC extension" ON)
add_minifi_option(MINIFI_RUST "Enables the build of rust based extensions." OFF)
add_minifi_dependent_option(MINIFI_EXTENSION_ENRICHMENT "Enables the build of enrichment (rust) extension." ON "MINIFI_RUST" OFF)
add_minifi_option(MINIFI_LMDB "Enables the LMDB extension." OFF)

set_minifi_cache_variable(CUSTOM_MALLOC OFF "Overwrite malloc implementation.")
Expand Down
Loading
Loading