MINIFICPP-2867 Implement ForkEnrichment/JoinEnrichmentAttributes - #2221
martinzink wants to merge 3 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR introduces a new “enrichment fork/join” mechanism to MiNiFi C++ standard processors, enabling a FlowFile to be cloned for enrichment and later rejoined with merged attributes.
Changes:
- Added new
ForkEnrichmentandJoinEnrichmentAttributesprocessors (including attribute keys shared via a small utility header). - Added unit tests and a Behave feature test validating enrichment + join behavior.
- Updated processor documentation, Behave step definitions, and the GitHub reference manifest to include the new processors.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| PROCESSORS.md | Documents the new processors, their properties, relationships, and output attributes. |
| extensions/standard-processors/utils/EnrichmentUtils.h | Introduces shared attribute-name constants for enrichment correlation/role. |
| extensions/standard-processors/processors/ForkEnrichment.h | Declares the fork processor (clones FlowFiles and sets correlation/role attributes). |
| extensions/standard-processors/processors/ForkEnrichment.cpp | Implements cloning + attribute assignment + routing logic. |
| extensions/standard-processors/processors/JoinEnrichmentAttributes.h | Declares the join processor (pairing/timeout tracking and join output). |
| extensions/standard-processors/processors/JoinEnrichmentAttributes.cpp | Implements pairing by group id, merging attributes, and timeout routing. |
| extensions/standard-processors/tests/unit/ForkEnrichmentTests.cpp | Unit coverage for basic fork behavior and max-batch handling. |
| extensions/standard-processors/tests/unit/JoinEnrichmentAttributesTests.cpp | Unit coverage for invalid inputs, pairing behavior, and timeout/max-batch. |
| extensions/standard-processors/tests/features/enrichment.feature | End-to-end Behave scenario verifying enrichment + join in a composed flow. |
| behave_framework/src/minifi_behave/steps/flow_building_steps.py | Adds “an …” variants for processor step definitions used by the new feature. |
| .github/references/ubuntu_22_04_clang_arm_manifest.json | Registers the new processors and their property/relationship metadata in the reference manifest. |
Comments suppressed due to low confidence (1)
extensions/standard-processors/processors/JoinEnrichmentAttributes.h:131
- Typo in comment: "cant" -> "can't".
// We need to track current session's FlowFiles (we cant add those)
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
bcfed19 to
f5ec2a8
Compare
| 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"}; |
| original->setAttribute(ENRICHMENT_ROLE, "ORIGINAL"); | ||
| enrichment->setAttribute(ENRICHMENT_ROLE, "ENRICHMENT"); |
There was a problem hiding this comment.
I would move the EnrichmentRole enum from JoinEnrichmentAttributes to somewhere which is visible from here, and use that instead of the "ORIGINAL" and "ENRICHMENT" strings.
| 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) const; | ||
|
|
||
| core::FlowFileStore flow_file_store_; |
There was a problem hiding this comment.
This is old code, but FlowFileStore assumes that after we move from incoming_files_, it will be empty. This is probably true in practice, but it isn't guaranteed. We should do a clear to be safe.
| return; | ||
| } | ||
|
|
||
| std::string group_id = *(flow_file->getAttribute(ENRICHMENT_GROUP_ID)); |
There was a problem hiding this comment.
we should check or assert that the attribute exists before dereferencing it
| session.transfer(previous_node.mapped(), Invalid); | ||
| if (!std::ranges::contains(session_flow_files_, previous_node.mapped()->getUUID())) { | ||
| session.add(previous_node.mapped()); | ||
| } |
There was a problem hiding this comment.
Does this work? I would expect that we need to add the flow file first before we can transfer it. If this is the correct order, then please add a comment explaining why.
There was a problem hiding this comment.
Can you add a couple more attributes to the original flow file, please? One which is not touched by the enrichment branch, and one which is overwritten by it. At the end, both (together with the existing one, all three) should show up in the log with the correct value.
| // 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); |
There was a problem hiding this comment.
are you sure this is not going to be flaky? maybe we could sleep for 2 ms, to make flakiness less likely
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you to ensure the following steps have been taken:
For all changes:
Is there a JIRA ticket associated with this PR? Is it referenced in the commit message?
Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically main)?
Is your initial contribution a single, squashed commit?
For code changes:
For documentation related changes:
Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.