Copilot commented on code in PR #2231:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2231#discussion_r3721481327
##########
libminifi/src/provenance/Provenance.cpp:
##########
@@ -454,7 +455,9 @@ void ProvenanceReporterImpl::commit() {
return;
}
- repo_->appendEvents(events_);
+ if (auto append_status = repo_->appendEvents(events_); !append_status) {
+ throw minifi::Exception(REPOSITORY_EXCEPTION, append_status.error());
+ }
Review Comment:
Throwing from `ProvenanceReporterImpl::commit()` on append failure is a
behavioral change that can now abort session/processor execution paths that
previously continued (especially under transient repo issues like full/IO
errors). If the intended behavior is to avoid crashing execution, consider
converting this to existing error-handling patterns (log + yield/rollback
signaling) or ensuring callers that invoke `commit()` have a consistent
exception boundary that translates it into controlled failure handling.
##########
libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h:
##########
@@ -28,32 +28,43 @@
#include "core/ProcessSession.h"
#include "RemoteProcessGroupPort.h"
#include "core/logging/LoggerFactory.h"
+#include "core/reporting/ReportingTaskBase.h"
namespace org::apache::nifi::minifi::core::reporting {
-class SiteToSiteProvenanceReportingTask : public
minifi::RemoteProcessGroupPort {
+class SiteToSiteProvenanceReportingTask : public ReportingTaskBase {
public:
- explicit SiteToSiteProvenanceReportingTask(std::shared_ptr<Configure>
configure)
- : minifi::RemoteProcessGroupPort(ReportTaskName, "",
std::move(configure),
- utils::IdGenerator::getIdGenerator()->generate(),
sitetosite::TransferDirection::SEND,
logging::LoggerFactory<SiteToSiteProvenanceReportingTask>::getLogger()) {
+ explicit SiteToSiteProvenanceReportingTask(ReportingTaskMetadata metadata)
+ : ReportingTaskBase{metadata},
+ remote_port_{metadata.name, metadata.uuid,
std::make_unique<RemoteProcessGroupPort>(metadata.name, "", Configure::create(),
+ metadata.uuid, sitetosite::TransferDirection::SEND, metadata.logger)}
+ {
Review Comment:
The embedded `RemoteProcessGroupPort` is constructed with
`Configure::create()` (a fresh/empty configuration) rather than the
agent/runtime configuration. This can lead to incorrect Site-to-Site behavior
(timeouts, SSL defaults, networking knobs) compared to processors/RPGs created
by the framework. Consider passing the real `Configure` instance through
`ReportingTaskMetadata` or exposing it via `ReportingTaskContext`, so the
reporting task uses the same configuration source as the rest of the flow.
##########
libminifi/src/core/flow/StructuredConfiguration.cpp:
##########
@@ -1134,4 +1064,83 @@ std::string StructuredConfiguration::serialize(const
core::ProcessGroup& process
return flow_serializer_->serialize(process_group, schema_,
sensitive_values_encryptor_, {}, parameter_contexts_);
}
+void StructuredConfiguration::parseReportingTasks(const Node&
reporting_tasks_node, core::ProcessGroup* parent_group) {
+ if (!reporting_tasks_node || !reporting_tasks_node.isSequence()) {
+ return;
+ }
+ for (const auto& reporting_task_node : reporting_tasks_node) {
+ checkRequiredField(reporting_task_node, schema_.name);
+ auto name = reporting_task_node[schema_.name].getString().value();
+ utils::Identifier id;
+ id = getOrGenerateId(reporting_task_node);
+
+ checkRequiredField(reporting_task_node, schema_.type);
+ auto type = reporting_task_node[schema_.type].getString().value();
+
+ auto reporting_task =
createProcessor(utils::string::partAfterLastOccurrenceOf(type, '.'), type,
name, id);
+ if (!reporting_task) {
+ logger_->log_error("Could not create a processor {} with id {}", name,
id.to_string());
+ throw std::invalid_argument("Could not create processor " + name);
+ }
+
+ reporting_task->setFlowIdentifier(flow_version_->getFlowIdentifier());
+
+ auto scheduling_strategy = getOptionalField(reporting_task_node,
schema_.scheduling_strategy, DEFAULT_SCHEDULING_STRATEGY);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ reporting_task->setSchedulingStrategy(core::TIMER_DRIVEN);
+ } else {
+ reporting_task->setSchedulingStrategy(core::CRON_DRIVEN);
+ }
+ auto scheduling_period_str = getOptionalField(reporting_task_node,
schema_.scheduling_period, DEFAULT_SCHEDULING_PERIOD_STR);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ if (auto scheduling_period =
utils::timeutils::StringToDuration<std::chrono::nanoseconds>(scheduling_period_str))
{
+ reporting_task->setSchedulingPeriod(*scheduling_period);
+ }
+ } else {
+ reporting_task->setCronPeriod(scheduling_period_str);
+ }
Review Comment:
Invalid `schedulingStrategy` values are silently treated as `CRON_DRIVEN`,
which can misconfigure reporting tasks without any validation. Additionally, if
the strategy is CRON-driven and `schedulingPeriod` is omitted, the default
`DEFAULT_SCHEDULING_PERIOD_STR` (e.g., \"1 sec\") is likely not a valid cron
expression but is still applied via `setCronPeriod(...)`. Recommend validating
allowed values (e.g., TIMER_DRIVEN/CRON_DRIVEN) and using a cron-appropriate
default (or requiring an explicit cron expression) when CRON is selected.
##########
libminifi/src/core/flow/StructuredConfiguration.cpp:
##########
@@ -128,12 +129,13 @@ std::unique_ptr<core::ProcessGroup>
StructuredConfiguration::getRootFrom(const N
uuids_.clear();
Node parameterContextsNode = root_node[schema_.parameter_contexts];
Node parameterProvidersNode = root_node[schema_.parameter_providers];
- Node provenanceReportNode = root_node[schema_.provenance_reporting];
+ Node reportingTasksNode = root_node[schema_.reporting_tasks];
parseParameterContexts(parameterContextsNode, parameterProvidersNode);
// Create the root process group
std::unique_ptr<core::ProcessGroup> root =
parseRootProcessGroup(root_node);
- parseProvenanceReporting(provenanceReportNode, root.get());
+
+ parseReportingTasks(reportingTasksNode, root.get());
Review Comment:
Legacy flow configs that use the previously supported provenance reporting
section (e.g., `provenanceReporting` / `Provenance Reporting`) will no longer
be parsed or executed because the old parsing path was removed and only
`reportingTasks` is read. If backward compatibility is required, consider
supporting both formats (e.g., translating the legacy provenance reporting
block into an equivalent reporting task) during a deprecation window to avoid
breaking existing deployments on upgrade.
##########
libminifi/src/core/flow/StructuredConfiguration.cpp:
##########
@@ -1134,4 +1064,83 @@ std::string StructuredConfiguration::serialize(const
core::ProcessGroup& process
return flow_serializer_->serialize(process_group, schema_,
sensitive_values_encryptor_, {}, parameter_contexts_);
}
+void StructuredConfiguration::parseReportingTasks(const Node&
reporting_tasks_node, core::ProcessGroup* parent_group) {
+ if (!reporting_tasks_node || !reporting_tasks_node.isSequence()) {
+ return;
+ }
+ for (const auto& reporting_task_node : reporting_tasks_node) {
+ checkRequiredField(reporting_task_node, schema_.name);
+ auto name = reporting_task_node[schema_.name].getString().value();
+ utils::Identifier id;
+ id = getOrGenerateId(reporting_task_node);
+
+ checkRequiredField(reporting_task_node, schema_.type);
+ auto type = reporting_task_node[schema_.type].getString().value();
+
+ auto reporting_task =
createProcessor(utils::string::partAfterLastOccurrenceOf(type, '.'), type,
name, id);
+ if (!reporting_task) {
+ logger_->log_error("Could not create a processor {} with id {}", name,
id.to_string());
+ throw std::invalid_argument("Could not create processor " + name);
+ }
+
+ reporting_task->setFlowIdentifier(flow_version_->getFlowIdentifier());
+
+ auto scheduling_strategy = getOptionalField(reporting_task_node,
schema_.scheduling_strategy, DEFAULT_SCHEDULING_STRATEGY);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ reporting_task->setSchedulingStrategy(core::TIMER_DRIVEN);
+ } else {
+ reporting_task->setSchedulingStrategy(core::CRON_DRIVEN);
+ }
+ auto scheduling_period_str = getOptionalField(reporting_task_node,
schema_.scheduling_period, DEFAULT_SCHEDULING_PERIOD_STR);
+ if (scheduling_strategy == "TIMER_DRIVEN") {
+ if (auto scheduling_period =
utils::timeutils::StringToDuration<std::chrono::nanoseconds>(scheduling_period_str))
{
+ reporting_task->setSchedulingPeriod(*scheduling_period);
+ }
+ } else {
+ reporting_task->setCronPeriod(scheduling_period_str);
+ }
+
+ if (auto penalization_node =
reporting_task_node[schema_.penalization_period]) {
+ if (auto penalization_period =
utils::timeutils::StringToDuration<std::chrono::milliseconds>(penalization_node.getString().value()))
{
+ reporting_task->setPenalizationPeriod(penalization_period.value());
+ }
+ }
+
+ if (auto yield_node = reporting_task_node[schema_.proc_yield_period]) {
+ if (auto yield_period =
utils::timeutils::StringToDuration<std::chrono::milliseconds>(yield_node.getString().value()))
{
+ reporting_task->setYieldPeriodMsec(yield_period.value());
+ }
+ }
+
+ if (auto bulletin_level_node =
reporting_task_node[schema_.bulletin_level]) {
+ if (auto bulletin_level = bulletin_level_node.getString().value();
!bulletin_level.empty()) {
+
reporting_task->setLogBulletinLevel(core::logging::mapStringToLogLevel(bulletin_level));
+ }
+ }
+
+ if (auto run_node = reporting_task_node[schema_.runduration_nanos]) {
+ if (auto run_duration_nanos =
parsing::parseIntegral<uint64_t>(run_node.getIntegerAsString().value())) {
+
reporting_task->setRunDurationNano(std::chrono::nanoseconds(*run_duration_nanos));
+ }
+ }
+
+ if (Node properties_node =
reporting_task_node[schema_.processor_properties]) {
+ parsePropertiesNode(properties_node, *reporting_task, name, nullptr);
+ }
Review Comment:
Flow schema adds `reporting_task_properties`, but reporting tasks are parsed
using `schema_.processor_properties`. This makes the new schema field unused
and increases the chance of future divergence between processor vs
reporting-task property keys. Prefer using `schema_.reporting_task_properties`
here (or remove the unused schema field) so the config contract stays explicit
and consistent.
##########
libminifi/src/core/flow/StructuredConfiguration.cpp:
##########
@@ -128,12 +129,13 @@ std::unique_ptr<core::ProcessGroup>
StructuredConfiguration::getRootFrom(const N
uuids_.clear();
Node parameterContextsNode = root_node[schema_.parameter_contexts];
Node parameterProvidersNode = root_node[schema_.parameter_providers];
- Node provenanceReportNode = root_node[schema_.provenance_reporting];
+ Node reportingTasksNode = root_node[schema_.reporting_tasks];
Review Comment:
Legacy flow configs that use the previously supported provenance reporting
section (e.g., `provenanceReporting` / `Provenance Reporting`) will no longer
be parsed or executed because the old parsing path was removed and only
`reportingTasks` is read. If backward compatibility is required, consider
supporting both formats (e.g., translating the legacy provenance reporting
block into an equivalent reporting task) during a deprecation window to avoid
breaking existing deployments on upgrade.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
