You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "martinzink (via GitHub)" <gi...@apache.org> on 2026/02/25 12:24:50 UTC
[PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
martinzink opened a new pull request, #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120
This is required to support fluid streaming api.
Currently if we opened the OutputStream it will always replace the old content, but lets say a processor wants to process flowfile content as a stream (read current content and write the new content at the same time), but if something comes up it might want to drop all changes and route the FlowFile to failure.
---
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:
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file?
- [ ] If applicable, have you updated the NOTICE file?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2853135774
##########
libminifi/src/core/ForwardingContentSession.cpp:
##########
@@ -51,6 +51,11 @@ std::shared_ptr<io::BaseStream> ForwardingContentSession::append(const std::shar
return repository_->write(*resource_id, true);
}
+void ForwardingContentSession::remove(const std::shared_ptr<ResourceClaim>& resource_id) {
Review Comment:
The remove() method only erases the claim from the internal tracking sets but doesn't delete the actual content from the repository. Since ForwardingContentSession writes directly to the repository (line 43), when a write is cancelled, the content file will remain on disk as an orphan. Add a call to repository_->remove(*resource_id) to properly clean up the physical file when remove() is called.
```suggestion
void ForwardingContentSession::remove(const std::shared_ptr<ResourceClaim>& resource_id) {
if (repository_ && resource_id) {
repository_->remove(*resource_id);
}
```
##########
libminifi/test/libtest/unit/ContentRepositoryDependentTests.h:
##########
@@ -171,4 +171,64 @@ void testReadFromZeroLengthFlowFile(std::shared_ptr<core::ContentRepository> con
REQUIRE_NOTHROW(process_session.readBuffer(flow_file));
REQUIRE_NOTHROW(process_session.read(flow_file, ReadUntilItCan{}));
}
+
+inline void testErrWrite(std::shared_ptr<core::ContentRepository> content_repo) {
+ const auto fixture = Fixture(std::move(content_repo));
+ core::ProcessSession& process_session = fixture.processSession();
+ const auto flow_file = process_session.create();
+ fixture.writeToFlowFile(flow_file, "original_content");
+ fixture.transferAndCommit(flow_file);
+
+ REQUIRE_THROWS(
+ process_session.write(flow_file, [](const std::shared_ptr<minifi::io::OutputStream>& output_stream) {
+ output_stream->write("new_content");
+ return MinifiIoStatus::MINIFI_IO_ERROR;
+ }));
+
+ CHECK(flow_file->getSize() == 16);
+ ReadUntilItCan read_until_it_can_callback;
+ const auto read_result = process_session.readBuffer(flow_file);
+ process_session.read(flow_file, std::ref(read_until_it_can_callback));
+ CHECK(to_string(read_result) == "original_content");
+}
+
+inline void testOkWrite(std::shared_ptr<core::ContentRepository> content_repo) {
+ const auto fixture = Fixture(std::move(content_repo));
+ core::ProcessSession& process_session = fixture.processSession();
+ const auto flow_file = process_session.create();
+ fixture.writeToFlowFile(flow_file, "original_content");
+ fixture.transferAndCommit(flow_file);
+
+ CHECK(flow_file->getSize() == 16);
+
+ process_session.write(flow_file, [](const std::shared_ptr<minifi::io::OutputStream>& output_stream) {
+ constexpr std::string str = "new_content";
Review Comment:
Using constexpr with std::string is unnecessary here. The string is not used for compile-time computation and is only needed at runtime within the lambda. Consider using a simple string literal, const char*, or std::string_view instead of constexpr std::string for clarity and to avoid confusion about the intended usage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "martinzink (via GitHub)" <gi...@apache.org>.
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3008012372
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
I've opened a followup PR that addresses this issue.
https://github.com/apache/nifi-minifi-cpp/pull/2148
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2852764279
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
Review Comment:
`MinifiIoStatus` is declared as an unscoped enum in `minifi-c.h` (`typedef enum MinifiIoStatus : int64_t { ... }`), so the enumerators are not members of the type. `MinifiIoStatus::MINIFI_IO_CANCEL` will not compile; use `MINIFI_IO_CANCEL` (unqualified) or change the enum to a scoped `enum class` and adjust the comparison/casts accordingly.
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
+ }
Review Comment:
The new cancellation path (`callback_result == ...CANCEL`) changes write semantics but doesn't appear to be covered by existing unit/integration tests. Please add a test that returns the cancel status from the callback and asserts that the FlowFile's original content/claim is preserved and the temporary claim is released (no orphan content left behind).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "fgerlits (via GitHub)" <gi...@apache.org>.
fgerlits commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3000932337
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
If someone is writing a new `io::OutputStreamCallback`, they will not know that they can signal cancellation, or how to do it, unless they happen to find and read this part of the source code.
Could we change `io::OutputStreamCallback` to return a `MinifiIoStatus` instead of an integer, to make it easier for future developers?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "fgerlits (via GitHub)" <gi...@apache.org>.
fgerlits closed pull request #2120: MINIFICPP-2731 Allow cancellation from ProcessSession::write
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "fgerlits (via GitHub)" <gi...@apache.org>.
fgerlits commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3001594370
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
yes, that sounds good
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "martinzink (via GitHub)" <gi...@apache.org>.
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2853104528
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
Review Comment:
its C header so cant use enum class afaik
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "martinzink (via GitHub)" <gi...@apache.org>.
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3001526161
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
forcing the bytes written number into that enum feels weird, what if instead std::expected<int64_t, MinifiIoStatus>? we can keep the normal int64_t on the C api side and wrap it in an unexpected if its < 0 (we can even throw if the return is out of the range of our defined (negative) error values)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "martinzink (via GitHub)" <gi...@apache.org>.
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3002880114
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
i've started the refactor but it snowballed and it seems way to big and error prone to include it here without discarding existing reviews (im on 78 files modified and counting)
how about I do this refactor in a followup PR?
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
i've started the refactor but it snowballed and it seems way too big and error prone to include it here without discarding existing reviews (im on 78 files modified and counting)
how about I do this refactor in a followup PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Posted by "martinzink (via GitHub)" <gi...@apache.org>.
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2853199097
##########
libminifi/src/core/ProcessSession.cpp:
##########
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for write");
}
- if (callback(stream) < 0) {
+ const auto callback_result = callback(stream);
+ if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
+ }
Review Comment:
[added unit tests](https://github.com/apache/nifi-minifi-cpp/pull/2120/commits/f901307f1ec9950422eefc0a5c080c1da635fe82)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org