You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ad...@apache.org on 2026/08/05 14:34:20 UTC
(nifi-minifi-cpp) branch MINIFICPP-2845 created (now f4ccd1854)
This is an automated email from the ASF dual-hosted git repository.
adamdebreceni pushed a change to branch MINIFICPP-2845
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
at f4ccd1854 MINIFICPP-2845 - Remove extra file
This branch includes the following new commits:
new a74b7d16a MINIFICPP-2840 - Refactor provenance part 1
new c1f65b296 MINIFICPP-2840 - More refactor
new 3685104d6 MINIFICPP-2845 - Monotonic provenance event ids, iteration support, refactor
new f4ccd1854 MINIFICPP-2845 - Remove extra file
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(nifi-minifi-cpp) 01/04: MINIFICPP-2840 - Refactor provenance part 1
Posted by ad...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
adamdebreceni pushed a commit to branch MINIFICPP-2845
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit a74b7d16afade55833d7df8ea1a7b6320329a5ee
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Wed Jun 3 12:51:09 2026 +0200
MINIFICPP-2840 - Refactor provenance part 1
---
core-framework/include/core/Repository.h | 4 +-
extensions/grafana-loki/PushGrafanaLokiGrpc.cpp | 2 +-
extensions/grafana-loki/PushGrafanaLokiREST.cpp | 2 +-
extensions/rocksdb-repos/ProvenanceRepository.cpp | 18 +--
extensions/rocksdb-repos/ProvenanceRepository.h | 2 +-
extensions/rocksdb-repos/tests/ProvenanceTests.cpp | 4 -
.../processors/LogAttribute.cpp | 2 +-
.../tests/unit/ProcessorTests.cpp | 10 +-
libminifi/include/core/FlowFile.h | 2 +-
.../reporting/SiteToSiteProvenanceReportingTask.h | 3 +-
libminifi/include/provenance/Provenance.h | 160 ++++++++++-----------
libminifi/src/core/FlowFile.cpp | 2 +-
libminifi/src/core/ProcessSession.cpp | 6 +-
libminifi/src/core/flow/FlowSchema.cpp | 6 +-
.../SiteToSiteProvenanceReportingTask.cpp | 32 +++--
libminifi/src/provenance/Provenance.cpp | 100 ++++++-------
libminifi/test/libtest/unit/ProvenanceTestHelper.h | 23 +--
minifi-api/include/minifi-cpp/core/FlowFile.h | 2 +-
minifi-api/include/minifi-cpp/core/Repository.h | 2 +-
.../include/minifi-cpp/provenance/Provenance.h | 6 +-
20 files changed, 184 insertions(+), 204 deletions(-)
diff --git a/core-framework/include/core/Repository.h b/core-framework/include/core/Repository.h
index ee2d3a040..7a817ae2f 100644
--- a/core-framework/include/core/Repository.h
+++ b/core-framework/include/core/Repository.h
@@ -106,8 +106,8 @@ class RepositoryImpl : public core::CoreComponentImpl, public core::RepositoryMe
return false;
}
- bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>>& /*store*/, size_t& /*max_size*/) override {
- return true;
+ std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t /*max_size*/) override {
+ return {};
}
bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) override;
diff --git a/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp b/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp
index 53428988f..100d60d52 100644
--- a/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp
+++ b/extensions/grafana-loki/PushGrafanaLokiGrpc.cpp
@@ -116,7 +116,7 @@ std::expected<void, std::string> PushGrafanaLokiGrpc::submitRequest(const std::v
for (const auto& flow_file : batched_flow_files) {
logproto::EntryAdapter *entry = stream->add_entries();
- auto timestamp_str = std::to_string(flow_file->getlineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
+ auto timestamp_str = std::to_string(flow_file->getLineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
auto timestamp_nanos = std::stoll(timestamp_str);
*entry->mutable_timestamp() = google::protobuf::util::TimeUtil::NanosecondsToTimestamp(timestamp_nanos);
diff --git a/extensions/grafana-loki/PushGrafanaLokiREST.cpp b/extensions/grafana-loki/PushGrafanaLokiREST.cpp
index fd01e15d4..1a9591888 100644
--- a/extensions/grafana-loki/PushGrafanaLokiREST.cpp
+++ b/extensions/grafana-loki/PushGrafanaLokiREST.cpp
@@ -133,7 +133,7 @@ std::string PushGrafanaLokiREST::createLokiJson(const std::vector<std::shared_pt
for (const auto& flow_file : batched_flow_files) {
rapidjson::Value log_line(rapidjson::kArrayType);
- auto timestamp_str = std::to_string(flow_file->getlineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
+ auto timestamp_str = std::to_string(flow_file->getLineageStartDate().time_since_epoch() / std::chrono::nanoseconds(1));
rapidjson::Value timestamp;
timestamp.SetString(timestamp_str.c_str(), gsl::narrow<rapidjson::SizeType>(timestamp_str.length()), allocator);
rapidjson::Value log_line_value;
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.cpp b/extensions/rocksdb-repos/ProvenanceRepository.cpp
index e14a97299..5c380545f 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.cpp
+++ b/extensions/rocksdb-repos/ProvenanceRepository.cpp
@@ -73,28 +73,30 @@ bool ProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::m
return true;
}
-bool ProvenanceRepository::getElements(std::vector<std::shared_ptr<core::SerializableComponent>> &records, size_t &max_size) {
+std::vector<std::shared_ptr<core::SerializableComponent>> ProvenanceRepository::getElements(size_t max_size) {
+ if (max_size == 0) {
+ return {};
+ }
auto opendb = db_->open();
if (!opendb) {
- return false;
+ return {};
}
+ std::vector<std::shared_ptr<core::SerializableComponent>> records;
rocksdb::ReadOptions options;
options.verify_checksums = verify_checksums_in_rocksdb_reads_;
std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(options));
- size_t requested_batch = max_size;
- max_size = 0;
for (it->SeekToFirst(); it->Valid(); it->Next()) {
- if (max_size >= requested_batch)
- break;
auto eventRead = ProvenanceEventRecord::create();
const auto slice = it->value();
io::BufferStream stream(std::as_bytes(std::span(slice.data(), slice.size())));
if (eventRead->deserialize(stream)) {
- max_size++;
records.push_back(eventRead);
+ if (--max_size == 0) {
+ break;
+ }
}
}
- return max_size > 0;
+ return records;
}
void ProvenanceRepository::destroy() {
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.h b/extensions/rocksdb-repos/ProvenanceRepository.h
index 3984c9911..950ecd31c 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.h
+++ b/extensions/rocksdb-repos/ProvenanceRepository.h
@@ -72,7 +72,7 @@ class ProvenanceRepository : public core::repository::RocksDbRepository {
// The repo is cleaned up by itself, there is no need to delete items.
return true;
}
- bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>> &records, size_t &max_size) override;
+ std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t max_size) override;
void destroy();
diff --git a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
index 95f129883..ecd3a77cf 100644
--- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
+++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
@@ -84,8 +84,6 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") {
REQUIRE(record2->getChildrenUuids().size() == 1);
utils::Identifier childId = record2->getChildrenUuids().at(0);
REQUIRE(childId == ffr1->getUUID());
- record2->removeChildUuid(childId);
- REQUIRE(record2->getChildrenUuids().empty());
}
TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
@@ -136,8 +134,6 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test
REQUIRE(record2->getChildrenUuids().size() == 1);
utils::Identifier childId = record2->getChildrenUuids().at(0);
REQUIRE(childId == ffr1->getUUID());
- record2->removeChildUuid(childId);
- REQUIRE(record2->getChildrenUuids().empty());
}
TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
diff --git a/extensions/standard-processors/processors/LogAttribute.cpp b/extensions/standard-processors/processors/LogAttribute.cpp
index 66cd7c040..4d8170569 100644
--- a/extensions/standard-processors/processors/LogAttribute.cpp
+++ b/extensions/standard-processors/processors/LogAttribute.cpp
@@ -82,7 +82,7 @@ std::string LogAttribute::generateLogMessage(core::ProcessSession& session, cons
message << "\nStandard FlowFile Attributes";
message << "\n" << "UUID:" << flow_file->getUUIDStr();
message << "\n" << "EntryDate:" << utils::timeutils::getTimeStr(flow_file->getEntryDate());
- message << "\n" << "lineageStartDate:" << utils::timeutils::getTimeStr(flow_file->getlineageStartDate());
+ message << "\n" << "lineageStartDate:" << utils::timeutils::getTimeStr(flow_file->getLineageStartDate());
message << "\n" << "Size:" << flow_file->getSize() << " Offset:" << flow_file->getOffset();
message << "\nFlowFile Attributes Map Content";
for (const auto& [attr_key, attr_value] : flow_file->getAttributes()) {
diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
index aa0e048ca..cb5905af2 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -466,19 +466,15 @@ TEST_CASE("Test Find file", "[getfileCreate3]") {
}
auto taskReport = &processorReport.get();
taskReport->setBatchSize(1);
- std::vector<std::shared_ptr<core::SerializableComponent>> recordsReport;
- recordsReport.push_back(std::make_shared<minifi::provenance::ProvenanceEventRecordImpl>());
processorReport->incrementActiveTasks();
processorReport->setScheduledState(core::ScheduledState::RUNNING);
- std::string jsonStr;
- std::size_t deserialized = 0;
- repo->getElements(recordsReport, deserialized);
+ auto recordsReport = repo->getElements(1);
std::function<void(const std::shared_ptr<core::ProcessContext> &, const std::shared_ptr<core::ProcessSession>&)> verifyReporter =
[&](const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
- taskReport->getJsonReport(*context, *session, recordsReport, jsonStr);
+ auto json_str = taskReport->getJsonReport(*context, *session, recordsReport);
REQUIRE(recordsReport.size() == 1);
REQUIRE(taskReport->getName() == std::string(minifi::core::reporting::SiteToSiteProvenanceReportingTask::ReportTaskName));
- REQUIRE(jsonStr.find("\"componentType\": \"getfileCreate2\"") != std::string::npos);
+ REQUIRE(json_str.find("\"componentType\": \"getfileCreate2\"") != std::string::npos);
};
testController.runSession(plan, false, verifyReporter);
diff --git a/libminifi/include/core/FlowFile.h b/libminifi/include/core/FlowFile.h
index 72f1aea2f..931bb1055 100644
--- a/libminifi/include/core/FlowFile.h
+++ b/libminifi/include/core/FlowFile.h
@@ -115,7 +115,7 @@ class FlowFileImpl : public CoreComponentImpl, public ReferenceContainerImpl, pu
* Get lineage start date
* @return lineage start date uint64_t
*/
- [[nodiscard]] std::chrono::system_clock::time_point getlineageStartDate() const override;
+ [[nodiscard]] std::chrono::system_clock::time_point getLineageStartDate() const override;
/**
* Sets the lineage start date
diff --git a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
index 182e6f15d..29767ad60 100644
--- a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
+++ b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
@@ -45,9 +45,8 @@ class SiteToSiteProvenanceReportingTask : public minifi::RemoteProcessGroupPort
static constexpr char const* ReportTaskName = "SiteToSiteProvenanceReportingTask";
static const char *ProvenanceAppStr;
- static void getJsonReport(core::ProcessContext& context, core::ProcessSession& session, std::vector<std::shared_ptr<core::SerializableComponent>> &records, std::string &report); // NOLINT
+ static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector<std::shared_ptr<core::SerializableComponent>> &records); // NOLINT
- void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& session_factory) override;
void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override;
void initialize() override;
diff --git a/libminifi/include/provenance/Provenance.h b/libminifi/include/provenance/Provenance.h
index 37e24f680..9d6aa11e2 100644
--- a/libminifi/include/provenance/Provenance.h
+++ b/libminifi/include/provenance/Provenance.h
@@ -49,7 +49,7 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
ProvenanceEventRecordImpl()
: core::SerializableComponentImpl(core::className<ProvenanceEventRecord>()) {
- _eventTime = std::chrono::system_clock::now();
+ event_time_ = std::chrono::system_clock::now();
}
~ProvenanceEventRecordImpl() override = default;
@@ -63,47 +63,47 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
}
std::map<std::string, std::string> getAttributes() const override {
- return _attributes;
+ return attributes_;
}
uint64_t getFileSize() const override {
- return _size;
+ return size_;
}
uint64_t getFileOffset() const override {
- return _offset;
+ return offset_;
}
std::chrono::system_clock::time_point getFlowFileEntryDate() const override {
- return _entryDate;
+ return entry_date_;
}
- std::chrono::system_clock::time_point getlineageStartDate() const override {
- return _lineageStartDate;
+ std::chrono::system_clock::time_point getLineageStartDate() const override {
+ return lineage_start_date_;
}
std::chrono::system_clock::time_point getEventTime() const override {
- return _eventTime;
+ return event_time_;
}
std::chrono::milliseconds getEventDuration() const override {
- return _eventDuration;
+ return event_duration_;
}
void setEventDuration(std::chrono::milliseconds duration) override {
- _eventDuration = duration;
+ event_duration_ = duration;
}
ProvenanceEventType getEventType() const override {
- return _eventType;
+ return event_type_;
}
std::string getComponentId() const override {
- return _componentId;
+ return component_id_;
}
std::string getComponentType() const override {
- return _componentType;
+ return component_type_;
}
utils::Identifier getFlowFileUuid() const override {
@@ -111,69 +111,61 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
}
std::string getContentFullPath() const override {
- return _contentFullPath;
+ return content_full_path;
}
std::vector<utils::Identifier> getLineageIdentifiers() const override {
- return _lineageIdentifiers;
+ return lineage_identifiers;
}
std::string getDetails() const override {
- return _details;
+ return details_;
}
void setDetails(const std::string& details) override {
- _details = details;
+ details_ = details;
}
std::string getTransitUri() override {
- return _transitUri;
+ return transit_uri_;
}
void setTransitUri(const std::string& uri) override {
- _transitUri = uri;
+ transit_uri_ = uri;
}
std::string getSourceSystemFlowFileIdentifier() const override {
- return _sourceSystemFlowFileIdentifier;
+ return source_system_flow_file_identifier_;
}
void setSourceSystemFlowFileIdentifier(const std::string& identifier) override {
- _sourceSystemFlowFileIdentifier = identifier;
+ source_system_flow_file_identifier_ = identifier;
}
std::vector<utils::Identifier> getParentUuids() const override {
- return _parentUuids;
+ return parent_uuids_;
}
void addParentUuid(const utils::Identifier& uuid) override {
- if (std::find(_parentUuids.begin(), _parentUuids.end(), uuid) != _parentUuids.end())
+ if (std::find(parent_uuids_.begin(), parent_uuids_.end(), uuid) != parent_uuids_.end())
return;
else
- _parentUuids.push_back(uuid);
+ parent_uuids_.push_back(uuid);
}
void addParentFlowFile(const core::FlowFile& flow_file) override {
addParentUuid(flow_file.getUUID());
}
- void removeParentUuid(const utils::Identifier& uuid) override {
- _parentUuids.erase(std::remove(_parentUuids.begin(), _parentUuids.end(), uuid), _parentUuids.end());
- }
-
- void removeParentFlowFile(const core::FlowFile& flow_file) override {
- removeParentUuid(flow_file.getUUID());
- }
-
std::vector<utils::Identifier> getChildrenUuids() const override {
- return _childrenUuids;
+ return children_uuids_;
}
void addChildUuid(const utils::Identifier& uuid) override {
- if (std::find(_childrenUuids.begin(), _childrenUuids.end(), uuid) != _childrenUuids.end())
+ if (std::find(children_uuids_.begin(), children_uuids_.end(), uuid) != children_uuids_.end())
return;
else
- _childrenUuids.push_back(uuid);
+ children_uuids_.push_back(uuid);
}
void addChildFlowFile(const core::FlowFile& flow_file) override {
@@ -181,50 +173,42 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
return;
}
- void removeChildUuid(const utils::Identifier& uuid) override {
- _childrenUuids.erase(std::remove(_childrenUuids.begin(), _childrenUuids.end(), uuid), _childrenUuids.end());
- }
-
- void removeChildFlowFile(const core::FlowFile& flow_file) override {
- removeChildUuid(flow_file.getUUID());
- }
-
std::string getAlternateIdentifierUri() const override {
- return _alternateIdentifierUri;
+ return alternate_identifier_uri_;
}
void setAlternateIdentifierUri(const std::string& uri) override {
- _alternateIdentifierUri = uri;
+ alternate_identifier_uri_ = uri;
}
std::string getRelationship() const override {
- return _relationship;
+ return relationship_;
}
void setRelationship(const std::string& relation) override {
- _relationship = relation;
+ relationship_ = relation;
}
std::string getSourceQueueIdentifier() const override {
- return _sourceQueueIdentifier;
+ return source_queue_identifier_;
}
void setSourceQueueIdentifier(const std::string& identifier) override {
- _sourceQueueIdentifier = identifier;
+ source_queue_identifier_ = identifier;
}
void fromFlowFile(const core::FlowFile& flow_file) override {
- _entryDate = flow_file.getEntryDate();
- _lineageStartDate = flow_file.getlineageStartDate();
- _lineageIdentifiers = flow_file.getlineageIdentifiers();
+ entry_date_ = flow_file.getEntryDate();
+ lineage_start_date_ = flow_file.getLineageStartDate();
+ lineage_identifiers = flow_file.getlineageIdentifiers();
flow_uuid_ = flow_file.getUUID();
- _attributes = flow_file.getAttributes();
- _size = flow_file.getSize();
- _offset = flow_file.getOffset();
+ attributes_ = flow_file.getAttributes();
+ size_ = flow_file.getSize();
+ offset_ = flow_file.getOffset();
if (flow_file.getConnection())
- _sourceQueueIdentifier = flow_file.getConnection()->getName();
+ source_queue_identifier_ = flow_file.getConnection()->getName();
if (flow_file.getResourceClaim()) {
- _contentFullPath = flow_file.getResourceClaim()->getContentFullPath();
+ content_full_path = flow_file.getResourceClaim()->getContentFullPath();
}
}
@@ -233,32 +217,32 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
bool loadFromRepository(const std::shared_ptr<core::Repository> &repo) override;
protected:
- ProvenanceEventType _eventType;
+ ProvenanceEventType event_type_;
// Date at which the event was created
- std::chrono::system_clock::time_point _eventTime{};
+ std::chrono::system_clock::time_point event_time_{};
// Date at which the flow file entered the flow
- std::chrono::system_clock::time_point _entryDate{};
+ std::chrono::system_clock::time_point entry_date_{};
// Date at which the origin of this flow file entered the flow
- std::chrono::system_clock::time_point _lineageStartDate{};
- std::chrono::milliseconds _eventDuration{};
- std::string _componentId;
- std::string _componentType;
+ std::chrono::system_clock::time_point lineage_start_date_{};
+ std::chrono::milliseconds event_duration_{};
+ std::string component_id_;
+ std::string component_type_;
// Size in bytes of the data corresponding to this flow file
- uint64_t _size = 0;
+ uint64_t size_ = 0;
utils::Identifier flow_uuid_;
- uint64_t _offset = 0;
- std::string _contentFullPath;
- std::map<std::string, std::string> _attributes;
+ uint64_t offset_ = 0;
+ std::string content_full_path;
+ std::map<std::string, std::string> attributes_;
// UUID string for all parents
- std::vector<utils::Identifier> _lineageIdentifiers;
- std::string _transitUri;
- std::string _sourceSystemFlowFileIdentifier;
- std::vector<utils::Identifier> _parentUuids;
- std::vector<utils::Identifier> _childrenUuids;
- std::string _details;
- std::string _sourceQueueIdentifier;
- std::string _relationship;
- std::string _alternateIdentifierUri;
+ std::vector<utils::Identifier> lineage_identifiers;
+ std::string transit_uri_;
+ std::string source_system_flow_file_identifier_;
+ std::vector<utils::Identifier> parent_uuids_;
+ std::vector<utils::Identifier> children_uuids_;
+ std::string details_;
+ std::string source_queue_identifier_;
+ std::string relationship_;
+ std::string alternate_identifier_uri_;
private:
ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl &parent);
@@ -271,8 +255,8 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
public:
ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, std::string componentId, std::string componentType)
: logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()) {
- _componentId = componentId;
- _componentType = componentType;
+ component_id_ = componentId;
+ component_type_ = componentType;
repo_ = repo;
}
@@ -281,21 +265,21 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
}
std::set<std::shared_ptr<ProvenanceEventRecord>> getEvents() const override {
- return _events;
+ return events_;
}
void add(const std::shared_ptr<ProvenanceEventRecord> &event) override {
- _events.insert(event);
+ events_.insert(event);
}
void remove(const std::shared_ptr<ProvenanceEventRecord> &event) override {
- if (_events.find(event) != _events.end()) {
- _events.erase(event);
+ if (events_.find(event) != events_.end()) {
+ events_.erase(event);
}
}
void clear() override {
- _events.clear();
+ events_.clear();
}
void commit() override;
@@ -317,19 +301,19 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
return nullptr;
}
- auto event = std::make_shared<ProvenanceEventRecordImpl>(eventType, _componentId, _componentType);
+ auto event = std::make_shared<ProvenanceEventRecordImpl>(eventType, component_id_, component_type_);
if (event)
event->fromFlowFile(flow_file);
return event;
}
- std::string _componentId;
- std::string _componentType;
+ std::string component_id_;
+ std::string component_type_;
private:
std::shared_ptr<core::logging::Logger> logger_;
- std::set<std::shared_ptr<ProvenanceEventRecord>> _events;
+ std::set<std::shared_ptr<ProvenanceEventRecord>> events_;
std::shared_ptr<core::Repository> repo_;
ProvenanceReporterImpl(const ProvenanceReporterImpl &parent);
diff --git a/libminifi/src/core/FlowFile.cpp b/libminifi/src/core/FlowFile.cpp
index 7f0819a28..0a19365a3 100644
--- a/libminifi/src/core/FlowFile.cpp
+++ b/libminifi/src/core/FlowFile.cpp
@@ -131,7 +131,7 @@ std::chrono::system_clock::time_point FlowFileImpl::getEventTime() const {
return event_time_;
}
// ! Get Lineage Start Date
-std::chrono::system_clock::time_point FlowFileImpl::getlineageStartDate() const {
+std::chrono::system_clock::time_point FlowFileImpl::getLineageStartDate() const {
return lineage_start_date_;
}
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index bd59eeaaa..d44c71947 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -69,7 +69,7 @@ ProcessSessionImpl::ProcessSessionImpl(std::shared_ptr<ProcessContext> processCo
stateManager_(process_context_->getStateManager()) {
logger_->log_trace("ProcessSession created for {}", process_context_->getProcessor().getName());
auto repo = process_context_->getProvenanceRepository();
- provenance_report_ = std::make_shared<provenance::ProvenanceReporterImpl>(repo, process_context_->getProcessor().getName(), process_context_->getProcessor().getName());
+ provenance_report_ = std::make_shared<provenance::ProvenanceReporterImpl>(repo, process_context_->getProcessor().getUUIDStr(), process_context_->getProcessor().getName());
content_session_ = process_context_->getContentRepository()->createSession();
if (stateManager_ && !stateManager_->beginTransaction()) {
@@ -113,7 +113,7 @@ std::shared_ptr<core::FlowFile> ProcessSessionImpl::create(const core::FlowFile*
}
record->setAttribute(attribute.first, attribute.second);
}
- record->setLineageStartDate(parent->getlineageStartDate());
+ record->setLineageStartDate(parent->getLineageStartDate());
record->setLineageIdentifiers(parent->getlineageIdentifiers());
record->getlineageIdentifiers().push_back(parent->getUUID());
}
@@ -163,7 +163,7 @@ std::shared_ptr<core::FlowFile> ProcessSessionImpl::cloneDuringTransfer(const co
}
record->setAttribute(attribute.first, attribute.second);
}
- record->setLineageStartDate(parent.getlineageStartDate());
+ record->setLineageStartDate(parent.getLineageStartDate());
record->setLineageIdentifiers(parent.getlineageIdentifiers());
record->getlineageIdentifiers().push_back(parent.getUUID());
diff --git a/libminifi/src/core/flow/FlowSchema.cpp b/libminifi/src/core/flow/FlowSchema.cpp
index 06ad9fe11..76d48f427 100644
--- a/libminifi/src/core/flow/FlowSchema.cpp
+++ b/libminifi/src/core/flow/FlowSchema.cpp
@@ -135,9 +135,9 @@ FlowSchema FlowSchema::getNiFiFlowJson() {
.controller_services = {"controllerServices"},
.controller_service_properties = {"properties"},
.remote_process_group = {"remoteProcessGroups"},
- .provenance_reporting = {},
- .provenance_reporting_port_uuid = {},
- .provenance_reporting_batch_size = {},
+ .provenance_reporting = {"provenanceReporting"},
+ .provenance_reporting_port_uuid = {"portUuid"},
+ .provenance_reporting_batch_size = {"batchSize"},
.funnels = {"funnels"},
.input_ports = {"inputPorts"},
.output_ports = {"outputPorts"},
diff --git a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
index 363fddede..6fb3d5d3c 100644
--- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
+++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
@@ -88,8 +88,7 @@ void appendJsonStr(const utils::SmallString<N>& value, rapidjson::Value& parent,
parent.PushBack(valueVal, alloc);
}
-void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&,
- std::vector<std::shared_ptr<core::SerializableComponent>> &records, std::string &report) {
+std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector<std::shared_ptr<core::SerializableComponent>> &records) {
rapidjson::Document array(rapidjson::kArrayType);
rapidjson::Document::AllocatorType &alloc = array.GetAllocator();
@@ -106,7 +105,7 @@ void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, cor
recordJson.AddMember("timestampMillis", int64_t{std::chrono::duration_cast<std::chrono::milliseconds>(record->getEventTime().time_since_epoch()).count()}, alloc);
recordJson.AddMember("durationMillis", int64_t{record->getEventDuration().count()}, alloc);
- recordJson.AddMember("lineageStart", int64_t{std::chrono::duration_cast<std::chrono::milliseconds>(record->getlineageStartDate().time_since_epoch()).count()}, alloc);
+ recordJson.AddMember("lineageStart", int64_t{std::chrono::duration_cast<std::chrono::milliseconds>(record->getLineageStartDate().time_since_epoch()).count()}, alloc);
recordJson.AddMember("entitySize", record->getFileSize(), alloc);
recordJson.AddMember("entityOffset", record->getFileOffset(), alloc);
@@ -148,24 +147,27 @@ void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, cor
rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(buffer);
array.Accept(writer);
- report = buffer.GetString();
-}
-
-void SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext&, core::ProcessSessionFactory&) {
+ return buffer.GetString();
}
void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
- logger_->log_debug("SiteToSiteProvenanceReportingTask -- onTrigger");
- std::vector<std::shared_ptr<core::SerializableComponent>> records;
- logger_->log_debug("batch size {} records", batch_size_);
- size_t deserialized = batch_size_;
+ auto* state_manager = context.getStateManager();
+ if (!state_manager) {
+ logger_->log_error("Failed to get StateManager");
+ context.yield();
+ return;
+ }
std::shared_ptr<core::Repository> repo = context.getProvenanceRepository();
- if (!repo->getElements(records, deserialized) && deserialized == 0) {
+ if (!repo) {
+ throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository");
+ }
+ auto records = repo->getElements(batch_size_);
+ if (records.empty()) {
+ logger_->log_debug("No new provenance records");
return;
}
- logger_->log_debug("Captured {} records", deserialized);
- std::string jsonStr;
- this->getJsonReport(context, session, records, jsonStr);
+ logger_->log_debug("Captured {} records", records.size());
+ std::string jsonStr = getJsonReport(context, session, records);
if (jsonStr.empty()) {
return;
}
diff --git a/libminifi/src/provenance/Provenance.cpp b/libminifi/src/provenance/Provenance.cpp
index d8ee5e723..ed4d0638a 100644
--- a/libminifi/src/provenance/Provenance.cpp
+++ b/libminifi/src/provenance/Provenance.cpp
@@ -42,10 +42,10 @@ const char *ProvenanceEventRecord::ProvenanceEventTypeStr[REPLAY + 1] = { "CREAT
ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, std::string componentId, std::string componentType)
: core::SerializableComponentImpl(core::className<ProvenanceEventRecord>()),
- _eventType(event),
- _eventTime(std::chrono::system_clock::now()),
- _componentId(std::move(componentId)),
- _componentType(std::move(componentType)) {
+ event_type_(event),
+ event_time_(std::chrono::system_clock::now()),
+ component_id_(std::move(componentId)),
+ component_type_(std::move(componentType)) {
}
bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr<core::Repository> &repo) {
@@ -70,9 +70,9 @@ bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr<core::R
ret = deserialize(stream);
if (ret) {
- logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {} success", getUUIDStr(), stream.size(), magic_enum::enum_name(_eventType));
+ logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {} success", getUUIDStr(), stream.size(), magic_enum::enum_name(event_type_));
} else {
- logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {} fail", getUUIDStr(), stream.size(), magic_enum::enum_name(_eventType));
+ logger_->log_debug("NiFi Provenance retrieve event {} size {} eventType {} fail", getUUIDStr(), stream.size(), magic_enum::enum_name(event_type_));
}
return ret;
@@ -86,48 +86,48 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- uint32_t eventType = this->_eventType;
+ uint32_t eventType = this->event_type_;
const auto ret = output_stream.write(eventType);
if (ret != 4) {
return false;
}
}
{
- uint64_t event_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(_eventTime.time_since_epoch()).count();
+ uint64_t event_time_ms = std::chrono::duration_cast<std::chrono::milliseconds>(event_time_.time_since_epoch()).count();
const auto ret = output_stream.write(event_time_ms);
if (ret != 8) {
return false;
}
}
{
- uint64_t entry_date_ms = std::chrono::duration_cast<std::chrono::milliseconds>(_entryDate.time_since_epoch()).count();
+ uint64_t entry_date_ms = std::chrono::duration_cast<std::chrono::milliseconds>(entry_date_.time_since_epoch()).count();
const auto ret = output_stream.write(entry_date_ms);
if (ret != 8) {
return false;
}
}
{
- uint64_t event_duration_ms = this->_eventDuration.count();
+ uint64_t event_duration_ms = this->event_duration_.count();
const auto ret = output_stream.write(event_duration_ms);
if (ret != 8) {
return false;
}
}
{
- uint64_t lineage_start_date_ms = std::chrono::duration_cast<std::chrono::milliseconds>(_lineageStartDate.time_since_epoch()).count();
+ uint64_t lineage_start_date_ms = std::chrono::duration_cast<std::chrono::milliseconds>(lineage_start_date_.time_since_epoch()).count();
const auto ret = output_stream.write(lineage_start_date_ms);
if (ret != 8) {
return false;
}
}
{
- const auto ret = output_stream.write(this->_componentId);
+ const auto ret = output_stream.write(this->component_id_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->_componentType);
+ const auto ret = output_stream.write(this->component_type_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -139,20 +139,20 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- const auto ret = output_stream.write(this->_details);
+ const auto ret = output_stream.write(this->details_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
// write flow attributes
{
- const auto numAttributes = gsl::narrow<uint32_t>(this->_attributes.size());
+ const auto numAttributes = gsl::narrow<uint32_t>(this->attributes_.size());
const auto ret = output_stream.write(numAttributes);
if (ret != 4) {
return false;
}
}
- for (const auto& itAttribute : _attributes) {
+ for (const auto& itAttribute : attributes_) {
{
const auto ret = output_stream.write(itAttribute.first);
if (ret == 0 || io::isError(ret)) {
@@ -167,71 +167,71 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- const auto ret = output_stream.write(this->_contentFullPath);
+ const auto ret = output_stream.write(this->content_full_path);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->_size);
+ const auto ret = output_stream.write(this->size_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = output_stream.write(this->_offset);
+ const auto ret = output_stream.write(this->offset_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = output_stream.write(this->_sourceQueueIdentifier);
+ const auto ret = output_stream.write(this->source_queue_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- if (this->_eventType == ProvenanceEventRecord::FORK || this->_eventType == ProvenanceEventRecord::CLONE || this->_eventType == ProvenanceEventRecord::JOIN) {
+ if (this->event_type_ == ProvenanceEventRecord::FORK || this->event_type_ == ProvenanceEventRecord::CLONE || this->event_type_ == ProvenanceEventRecord::JOIN) {
// write UUIDs
{
- const auto parent_uuids_count = gsl::narrow<uint32_t>(this->_parentUuids.size());
+ const auto parent_uuids_count = gsl::narrow<uint32_t>(this->parent_uuids_.size());
const auto ret = output_stream.write(parent_uuids_count);
if (ret != 4) {
return false;
}
}
- for (const auto& parentUUID : _parentUuids) {
+ for (const auto& parentUUID : parent_uuids_) {
const auto ret = output_stream.write(parentUUID);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto children_uuids_count = gsl::narrow<uint32_t>(this->_childrenUuids.size());
+ const auto children_uuids_count = gsl::narrow<uint32_t>(this->children_uuids_.size());
const auto ret = output_stream.write(children_uuids_count);
if (ret != 4) {
return false;
}
}
- for (const auto& childUUID : _childrenUuids) {
+ for (const auto& childUUID : children_uuids_) {
const auto ret = output_stream.write(childUUID);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- } else if (this->_eventType == ProvenanceEventRecord::SEND || this->_eventType == ProvenanceEventRecord::FETCH) {
- const auto ret = output_stream.write(this->_transitUri);
+ } else if (this->event_type_ == ProvenanceEventRecord::SEND || this->event_type_ == ProvenanceEventRecord::FETCH) {
+ const auto ret = output_stream.write(this->transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
- } else if (this->_eventType == ProvenanceEventRecord::RECEIVE) {
+ } else if (this->event_type_ == ProvenanceEventRecord::RECEIVE) {
{
- const auto ret = output_stream.write(this->_transitUri);
+ const auto ret = output_stream.write(this->transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->_sourceSystemFlowFileIdentifier);
+ const auto ret = output_stream.write(this->source_system_flow_file_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -258,7 +258,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
}
if (auto event_type_opt = magic_enum::enum_cast<ProvenanceEventRecord::ProvenanceEventType>(eventType)) {
- _eventType = *event_type_opt;
+ event_type_ = *event_type_opt;
} else {
return false;
}
@@ -269,7 +269,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
if (ret != 8) {
return false;
}
- _eventTime = std::chrono::system_clock::time_point() + std::chrono::milliseconds(event_time_in_ms);
+ event_time_ = std::chrono::system_clock::time_point() + std::chrono::milliseconds(event_time_in_ms);
}
{
@@ -278,7 +278,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
if (ret != 8) {
return false;
}
- _entryDate = std::chrono::system_clock::time_point() + std::chrono::milliseconds(entry_date_in_ms);
+ entry_date_ = std::chrono::system_clock::time_point() + std::chrono::milliseconds(entry_date_in_ms);
}
{
@@ -287,7 +287,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
if (ret != 8) {
return false;
}
- _eventDuration = std::chrono::milliseconds(event_duration_ms);
+ event_duration_ = std::chrono::milliseconds(event_duration_ms);
}
{
@@ -296,18 +296,18 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
if (ret != 8) {
return false;
}
- _lineageStartDate = std::chrono::system_clock::time_point() + std::chrono::milliseconds(lineage_start_date_in_ms);
+ lineage_start_date_ = std::chrono::system_clock::time_point() + std::chrono::milliseconds(lineage_start_date_in_ms);
}
{
- const auto ret = input_stream.read(this->_componentId);
+ const auto ret = input_stream.read(this->component_id_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->_componentType);
+ const auto ret = input_stream.read(this->component_type_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -321,7 +321,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
}
{
- const auto ret = input_stream.read(this->_details);
+ const auto ret = input_stream.read(this->details_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -351,38 +351,38 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
return false;
}
}
- this->_attributes[key] = value;
+ this->attributes_[key] = value;
}
{
- const auto ret = input_stream.read(this->_contentFullPath);
+ const auto ret = input_stream.read(this->content_full_path);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->_size);
+ const auto ret = input_stream.read(this->size_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = input_stream.read(this->_offset);
+ const auto ret = input_stream.read(this->offset_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = input_stream.read(this->_sourceQueueIdentifier);
+ const auto ret = input_stream.read(this->source_queue_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- if (this->_eventType == ProvenanceEventRecord::FORK || this->_eventType == ProvenanceEventRecord::CLONE || this->_eventType == ProvenanceEventRecord::JOIN) {
+ if (this->event_type_ == ProvenanceEventRecord::FORK || this->event_type_ == ProvenanceEventRecord::CLONE || this->event_type_ == ProvenanceEventRecord::JOIN) {
// read UUIDs
uint32_t number = 0;
{
@@ -419,22 +419,22 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
}
this->addChildUuid(childUUID);
}
- } else if (this->_eventType == ProvenanceEventRecord::SEND || this->_eventType == ProvenanceEventRecord::FETCH) {
+ } else if (this->event_type_ == ProvenanceEventRecord::SEND || this->event_type_ == ProvenanceEventRecord::FETCH) {
{
- const auto ret = input_stream.read(this->_transitUri);
+ const auto ret = input_stream.read(this->transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- } else if (this->_eventType == ProvenanceEventRecord::RECEIVE) {
+ } else if (this->event_type_ == ProvenanceEventRecord::RECEIVE) {
{
- const auto ret = input_stream.read(this->_transitUri);
+ const auto ret = input_stream.read(this->transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->_sourceSystemFlowFileIdentifier);
+ const auto ret = input_stream.read(this->source_system_flow_file_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -456,7 +456,7 @@ void ProvenanceReporterImpl::commit() {
std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> flowData;
- for (auto& event : _events) {
+ for (auto& event : events_) {
auto stramptr = std::make_unique<io::BufferStream>();
event->serialize(*stramptr);
diff --git a/libminifi/test/libtest/unit/ProvenanceTestHelper.h b/libminifi/test/libtest/unit/ProvenanceTestHelper.h
index 0d0407a9c..03d6e0c32 100644
--- a/libminifi/test/libtest/unit/ProvenanceTestHelper.h
+++ b/libminifi/test/libtest/unit/ProvenanceTestHelper.h
@@ -34,6 +34,7 @@
#include "properties/Configure.h"
#include "minifi-cpp/SwapManager.h"
#include "io/BufferStream.h"
+#include "minifi-cpp/provenance/Provenance.h"
using namespace std::literals::chrono_literals;
@@ -88,19 +89,23 @@ class TestRepositoryBase : public T_BaseRepository {
}
}
- bool getElements(std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> &store, size_t &max_size) override {
+ std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> getElements(size_t max_size) override {
+ if (max_size == 0) {
+ return {};
+ }
+ std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> store;
std::lock_guard<std::mutex> lock{repository_results_mutex_};
- max_size = 0;
for (const auto &entry : repository_results_) {
- if (max_size >= store.size()) {
- break;
- }
- const auto eventRead = store.at(max_size);
+ const auto eventRead = org::apache::nifi::minifi::provenance::ProvenanceEventRecord::create();
org::apache::nifi::minifi::io::BufferStream stream(std::as_bytes(std::span(entry.second)));
- eventRead->deserialize(stream);
- ++max_size;
+ if (eventRead->deserialize(stream)) {
+ store.push_back(std::move(eventRead));
+ if (--max_size == 0) {
+ break;
+ }
+ }
}
- return true;
+ return store;
}
std::map<std::string, std::string> getRepoMap() const {
diff --git a/minifi-api/include/minifi-cpp/core/FlowFile.h b/minifi-api/include/minifi-cpp/core/FlowFile.h
index 811f9e8a4..951829b45 100644
--- a/minifi-api/include/minifi-cpp/core/FlowFile.h
+++ b/minifi-api/include/minifi-cpp/core/FlowFile.h
@@ -54,7 +54,7 @@ class FlowFile : public virtual CoreComponent, public virtual ReferenceContainer
virtual void setDeleted(bool deleted) = 0;
[[nodiscard]] virtual std::chrono::system_clock::time_point getEntryDate() const = 0;
[[nodiscard]] virtual std::chrono::system_clock::time_point getEventTime() const = 0;
- [[nodiscard]] virtual std::chrono::system_clock::time_point getlineageStartDate() const = 0;
+ [[nodiscard]] virtual std::chrono::system_clock::time_point getLineageStartDate() const = 0;
virtual void setLineageStartDate(std::chrono::system_clock::time_point date) = 0;
virtual void setLineageIdentifiers(const std::vector<utils::Identifier>& lineage_Identifiers) = 0;
virtual void setAttribute(std::string_view name, std::string value) = 0;
diff --git a/minifi-api/include/minifi-cpp/core/Repository.h b/minifi-api/include/minifi-cpp/core/Repository.h
index f2177f00c..33b5c3747 100644
--- a/minifi-api/include/minifi-cpp/core/Repository.h
+++ b/minifi-api/include/minifi-cpp/core/Repository.h
@@ -54,7 +54,7 @@ class Repository : public virtual core::CoreComponent, public virtual core::Repo
virtual bool Get(const std::string& /*key*/, std::string& /*value*/) = 0;
- virtual bool getElements(std::vector<std::shared_ptr<core::SerializableComponent>>& /*store*/, size_t& /*max_size*/) = 0;
+ virtual std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t max_size) = 0;
virtual bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) = 0;
diff --git a/minifi-api/include/minifi-cpp/provenance/Provenance.h b/minifi-api/include/minifi-cpp/provenance/Provenance.h
index ef5c5a9b5..ef396958a 100644
--- a/minifi-api/include/minifi-cpp/provenance/Provenance.h
+++ b/minifi-api/include/minifi-cpp/provenance/Provenance.h
@@ -154,7 +154,7 @@ class ProvenanceEventRecord : public virtual core::SerializableComponent {
virtual uint64_t getFileSize() const = 0;
virtual uint64_t getFileOffset() const = 0;
virtual std::chrono::system_clock::time_point getFlowFileEntryDate() const = 0;
- virtual std::chrono::system_clock::time_point getlineageStartDate() const = 0;
+ virtual std::chrono::system_clock::time_point getLineageStartDate() const = 0;
virtual std::chrono::system_clock::time_point getEventTime() const = 0;
virtual std::chrono::milliseconds getEventDuration() const = 0;
virtual void setEventDuration(std::chrono::milliseconds duration) = 0;
@@ -173,13 +173,9 @@ class ProvenanceEventRecord : public virtual core::SerializableComponent {
virtual std::vector<utils::Identifier> getParentUuids() const = 0;
virtual void addParentUuid(const utils::Identifier& uuid) = 0;
virtual void addParentFlowFile(const core::FlowFile& flow_file) = 0;
- virtual void removeParentUuid(const utils::Identifier& uuid) = 0;
- virtual void removeParentFlowFile(const core::FlowFile& flow_file) = 0;
virtual std::vector<utils::Identifier> getChildrenUuids() const = 0;
virtual void addChildUuid(const utils::Identifier& uuid) = 0;
virtual void addChildFlowFile(const core::FlowFile& flow_file) = 0;
- virtual void removeChildUuid(const utils::Identifier& uuid) = 0;
- virtual void removeChildFlowFile(const core::FlowFile& flow_file) = 0;
virtual std::string getAlternateIdentifierUri() const = 0;
virtual void setAlternateIdentifierUri(const std::string& uri) = 0;
virtual std::string getRelationship() const = 0;
(nifi-minifi-cpp) 03/04: MINIFICPP-2845 - Monotonic provenance event ids, iteration support, refactor
Posted by ad...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
adamdebreceni pushed a commit to branch MINIFICPP-2845
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3685104d65e786c9d86378a5a4101fa25c539efa
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Mon Jul 6 11:13:01 2026 +0200
MINIFICPP-2845 - Monotonic provenance event ids, iteration support, refactor
---
cmake/FetchSQLite.cmake | 29 +++
core-framework/common/src/utils/Id.cpp | 12 ++
core-framework/include/core/Repository.h | 6 -
core-framework/src/core/Repository.cpp | 16 --
extensions/rocksdb-repos/ProvenanceRepository.cpp | 108 -----------
.../rocksdb-repos/RocksDbProvenanceRepository.cpp | 207 +++++++++++++++++++++
...eRepository.h => RocksDbProvenanceRepository.h} | 30 ++-
.../tests/DBProvenanceRepositoryTests.cpp | 142 +++++++++++++-
extensions/rocksdb-repos/tests/ProvenanceTests.cpp | 41 ++--
extensions/rocksdb-repos/tests/RepoTests.cpp | 12 +-
.../tests/integration/VerifyInvokeHTTP.h | 2 +-
.../tests/unit/ProcessorTests.cpp | 7 +-
libminifi/include/CronDrivenSchedulingAgent.h | 2 +-
libminifi/include/EventDrivenSchedulingAgent.h | 2 +-
libminifi/include/FlowController.h | 6 +-
libminifi/include/SchedulingAgent.h | 4 +-
libminifi/include/ThreadedSchedulingAgent.h | 2 +-
libminifi/include/TimerDrivenSchedulingAgent.h | 2 +-
libminifi/include/core/ProcessContextImpl.h | 8 +-
.../reporting/SiteToSiteProvenanceReportingTask.h | 2 +-
.../core/repository/NoOpThreadedRepository.h | 18 +-
.../core/repository/VolatileProvenanceRepository.h | 35 +++-
libminifi/include/provenance/Provenance.h | 25 ++-
libminifi/src/FlowController.cpp | 2 +-
libminifi/src/core/ProcessContextImpl.cpp | 4 +-
.../src/core/flow/StructuredConfiguration.cpp | 4 +
.../SiteToSiteProvenanceReportingTask.cpp | 51 ++++-
libminifi/src/provenance/Provenance.cpp | 12 +-
libminifi/test/flow-tests/SessionTests.cpp | 2 +-
libminifi/test/integration/C2PauseResumeTest.cpp | 2 +-
.../ControllerServiceIntegrationTests.cpp | 2 +-
.../test/libtest/integration/IntegrationBase.cpp | 2 +-
libminifi/test/libtest/unit/ProvenanceTestHelper.h | 27 ++-
libminifi/test/libtest/unit/TestBase.cpp | 4 +-
libminifi/test/libtest/unit/TestBase.h | 6 +-
.../test/libtest/unit/TestControllerWithFlow.cpp | 2 +-
.../test/persistence-tests/PersistenceTests.cpp | 6 +-
libminifi/test/unit/SchedulingAgentTests.cpp | 2 +-
minifi-api/common/include/minifi-cpp/utils/Id.h | 3 +
.../include/minifi-cpp/core/ProcessContext.h | 3 +-
minifi-api/include/minifi-cpp/core/Repository.h | 4 -
.../include/minifi-cpp/provenance/Provenance.h | 5 +-
.../minifi-cpp/provenance/ProvenanceRepository.h | 44 +++++
minifi_main/MiNiFiMain.cpp | 2 +-
44 files changed, 665 insertions(+), 242 deletions(-)
diff --git a/cmake/FetchSQLite.cmake b/cmake/FetchSQLite.cmake
new file mode 100644
index 000000000..65a9fb62f
--- /dev/null
+++ b/cmake/FetchSQLite.cmake
@@ -0,0 +1,29 @@
+# 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(FetchContent)
+
+FetchContent_Declare(
+ sqlite3
+ URL "https://www.sqlite.org/2026/sqlite-amalgamation-3530200.zip"
+ URL_HASH "SHA256=8a310d0a16c7a90cacd4c884e70faa51c902afed2a89f63aaa0126ab83558a32"
+)
+
+FetchContent_MakeAvailable(sqlite3)
+
+add_library(sqlite3 STATIC ${sqlite3_SOURCE_DIR}/sqlite3.c)
+target_include_directories(sqlite3 PUBLIC ${sqlite3_SOURCE_DIR})
diff --git a/core-framework/common/src/utils/Id.cpp b/core-framework/common/src/utils/Id.cpp
index d401d0a5e..09bdee3eb 100644
--- a/core-framework/common/src/utils/Id.cpp
+++ b/core-framework/common/src/utils/Id.cpp
@@ -43,6 +43,18 @@ bool Identifier::isNil() const {
return *this == Identifier{};
}
+Identifier& Identifier::operator++() {
+ for (int byte_idx = 15; byte_idx >= 0 && ++data_[byte_idx] == 0; --byte_idx) {}
+ return *this;
+}
+
+Identifier Identifier::operator++(int) {
+ auto result = *this;
+ ++*this;
+ return result;
+}
+
+
bool Identifier::operator!=(const Identifier& other) const {
return !(*this == other);
}
diff --git a/core-framework/include/core/Repository.h b/core-framework/include/core/Repository.h
index 7a817ae2f..1615a20b8 100644
--- a/core-framework/include/core/Repository.h
+++ b/core-framework/include/core/Repository.h
@@ -106,12 +106,6 @@ class RepositoryImpl : public core::CoreComponentImpl, public core::RepositoryMe
return false;
}
- std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t /*max_size*/) override {
- return {};
- }
-
- bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) override;
-
void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) override {
}
diff --git a/core-framework/src/core/Repository.cpp b/core-framework/src/core/Repository.cpp
index 722e82e7a..7004e05c0 100644
--- a/core-framework/src/core/Repository.cpp
+++ b/core-framework/src/core/Repository.cpp
@@ -27,20 +27,4 @@ bool RepositoryImpl::Delete(std::vector<std::shared_ptr<core::SerializableCompon
return found;
}
-bool RepositoryImpl::storeElement(const std::shared_ptr<core::SerializableComponent>& element) {
- if (!element) {
- return false;
- }
-
- org::apache::nifi::minifi::io::BufferStream stream;
-
- element->serialize(stream);
-
- if (!Put(element->getUUIDStr(), reinterpret_cast<const uint8_t*>(stream.getBuffer().data()), stream.size())) {
- logger_->log_error("NiFi Provenance Store event {} size {} fail", element->getUUIDStr(), stream.size());
- return false;
- }
- return true;
-}
-
} // namespace org::apache::nifi::minifi::core
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.cpp b/extensions/rocksdb-repos/ProvenanceRepository.cpp
deleted file mode 100644
index 5c380545f..000000000
--- a/extensions/rocksdb-repos/ProvenanceRepository.cpp
+++ /dev/null
@@ -1,108 +0,0 @@
-/**
- * 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 "ProvenanceRepository.h"
-
-#include <string>
-
-#include "core/Resource.h"
-
-namespace org::apache::nifi::minifi::provenance {
-
-bool ProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) {
- std::string value;
- if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) {
- directory_ = value;
- }
- logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_);
- if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) {
- max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size"));
- }
- logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_);
- if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) {
- if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value))
- max_partition_millis_ = *max_partition;
- }
- logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_);
-
- verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false);
- logger_->log_debug("{} checksum verification in ProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using");
-
- auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) {
- minifi::internal::setCommonRocksDbOptions(db_opts);
- };
-
- // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after
- // This shouldn't go above 16MB and the configured total size of the db should cap it as well
- auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) {
- int64_t max_buffer_size = 16 << 20;
- cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_));
- cf_opts.max_write_buffer_number = 4;
- cf_opts.min_write_buffer_number_to_merge = 1;
-
- cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO;
- cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false);
- if (max_partition_millis_ > std::chrono::milliseconds(0)) {
- cf_opts.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count();
- }
- };
-
- db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_,
- minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options));
- if (db_->open()) {
- logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_);
- } else {
- logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_);
- return false;
- }
-
- return true;
-}
-
-std::vector<std::shared_ptr<core::SerializableComponent>> ProvenanceRepository::getElements(size_t max_size) {
- if (max_size == 0) {
- return {};
- }
- auto opendb = db_->open();
- if (!opendb) {
- return {};
- }
- std::vector<std::shared_ptr<core::SerializableComponent>> records;
- rocksdb::ReadOptions options;
- options.verify_checksums = verify_checksums_in_rocksdb_reads_;
- std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(options));
- for (it->SeekToFirst(); it->Valid(); it->Next()) {
- auto eventRead = ProvenanceEventRecord::create();
- const auto slice = it->value();
- io::BufferStream stream(std::as_bytes(std::span(slice.data(), slice.size())));
- if (eventRead->deserialize(stream)) {
- records.push_back(eventRead);
- if (--max_size == 0) {
- break;
- }
- }
- }
- return records;
-}
-
-void ProvenanceRepository::destroy() {
- db_.reset();
-}
-
-REGISTER_RESOURCE_AS(ProvenanceRepository, InternalResource, ("ProvenanceRepository", "provenancerepository"));
-
-} // namespace org::apache::nifi::minifi::provenance
diff --git a/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp b/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp
new file mode 100644
index 000000000..f3397ea62
--- /dev/null
+++ b/extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp
@@ -0,0 +1,207 @@
+/**
+ * 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 "RocksDbProvenanceRepository.h"
+
+#include <string>
+
+#include "core/Resource.h"
+
+namespace org::apache::nifi::minifi::provenance {
+
+namespace {
+class EventCursor : public ProvenanceRepository::Cursor {
+public:
+ explicit EventCursor(std::string event_id): event_id_(std::move(event_id)) {}
+ [[nodiscard]]
+ std::string toString() const override {
+ return event_id_;
+ }
+ ~EventCursor() override = default;
+
+ std::string event_id_;
+};
+} // namespace
+
+static const std::string_view NEXT_EVENT_UUID_KEY = "next_event_uuid";
+
+bool RocksDbProvenanceRepository::initialize(const std::shared_ptr<org::apache::nifi::minifi::Configure> &config) {
+ if (!RocksDbRepository::initialize(config)) {
+ return false;
+ }
+ std::string value;
+ if (config->get(Configure::nifi_provenance_repository_directory_default, value) && !value.empty()) {
+ directory_ = value;
+ }
+ logger_->log_debug("MiNiFi Provenance Repository Directory {}", directory_);
+ if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) {
+ max_partition_bytes_ = gsl::narrow<int64_t>(parsing::parseDataSize(value) | utils::orThrow("expected parsable data size"));
+ }
+ logger_->log_debug("MiNiFi Provenance Max Partition Bytes {}", max_partition_bytes_);
+ if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) {
+ if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value))
+ max_partition_millis_ = *max_partition;
+ }
+ logger_->log_debug("MiNiFi Provenance Max Storage Time: [{}]", max_partition_millis_);
+
+ verify_checksums_in_rocksdb_reads_ = (config->get(Configure::nifi_provenance_repository_rocksdb_read_verify_checksums) | utils::andThen(&utils::string::toBool)).value_or(false);
+ logger_->log_debug("{} checksum verification in RocksDbProvenanceRepository", verify_checksums_in_rocksdb_reads_ ? "Using" : "Not using");
+
+ auto db_options = [] (minifi::internal::Writable<rocksdb::DBOptions>& db_opts) {
+ minifi::internal::setCommonRocksDbOptions(db_opts);
+ };
+
+ // Rocksdb write buffers act as a log of database operation: grow till reaching the limit, serialized after
+ // This shouldn't go above 16MB and the configured total size of the db should cap it as well
+ auto cf_options = [this] (rocksdb::ColumnFamilyOptions& cf_opts) {
+ int64_t max_buffer_size = 16 << 20;
+ cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_));
+ cf_opts.max_write_buffer_number = 4;
+ cf_opts.min_write_buffer_number_to_merge = 1;
+
+ cf_opts.compaction_style = rocksdb::CompactionStyle::kCompactionStyleFIFO;
+ cf_opts.compaction_options_fifo = rocksdb::CompactionOptionsFIFO(max_partition_bytes_, false);
+ if (max_partition_millis_ > std::chrono::milliseconds(0)) {
+ cf_opts.ttl = std::chrono::duration_cast<std::chrono::seconds>(max_partition_millis_).count();
+ }
+ };
+
+ db_ = minifi::internal::RocksDatabase::create(db_options, cf_options, directory_,
+ minifi::internal::getRocksDbOptionsToOverride(config, Configure::nifi_provenance_repository_rocksdb_options));
+ std::string internal_state_db_uri = [&] {
+ const std::string_view minifidb_scheme = "minifidb://";
+ if (directory_.starts_with(minifidb_scheme)) {
+ return directory_ + "-internal-state";
+ }
+ std::string uri = utils::string::join_pack(minifidb_scheme, directory_);
+ if (uri.ends_with("/") || uri.ends_with("\\")) {
+ uri.pop_back();
+ }
+ return uri + "/internal-state";
+ }();
+ internal_state_db_ = minifi::internal::RocksDatabase::create(db_options, {}, internal_state_db_uri, {});
+ if (auto open_state_db = internal_state_db_->open()) {
+ rocksdb::ReadOptions options;
+ options.verify_checksums = verify_checksums_in_rocksdb_reads_;
+ std::string next_event_uuid_str;
+ if (open_state_db->Get(options, NEXT_EVENT_UUID_KEY, &next_event_uuid_str).ok()) {
+ next_event_id_ = next_event_uuid_str;
+ } else {
+ logger_->log_error("Could not find '{}'", NEXT_EVENT_UUID_KEY);
+ next_event_id_ = utils::IdGenerator::getIdGenerator()->generate();
+ }
+ logger_->log_trace("Using next event uuid: {}", next_event_id_.to_string());
+ } else {
+ logger_->log_error("Could not open internal state column in provenance repository {}", internal_state_db_uri);
+ return false;
+ }
+ if (db_->open()) {
+ logger_->log_debug("MiNiFi Provenance Repository database open {} success", directory_);
+ } else {
+ logger_->log_error("MiNiFi Provenance Repository database open {} failed", directory_);
+ return false;
+ }
+
+ return true;
+}
+
+void RocksDbProvenanceRepository::destroy() {
+ db_.reset();
+}
+
+std::unique_ptr<ProvenanceRepository::Cursor> RocksDbProvenanceRepository::cursorFromString(std::optional<std::string> cursor_str) {
+ if (cursor_str.has_value()) {
+ return std::make_unique<EventCursor>(cursor_str.value());
+ }
+ return std::make_unique<EventCursor>("");
+}
+
+std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> RocksDbProvenanceRepository::getEvents(size_t max_size, Cursor* cursor) {
+ auto* event_cursor = dynamic_cast<EventCursor*>(cursor);
+ if (cursor && !event_cursor) {
+ return std::unexpected{"Invalid cursor"};
+ }
+ if (max_size == 0) {
+ return {};
+ }
+ auto opendb = db_->open();
+ if (!opendb) {
+ return std::unexpected{"Failed to open database"};
+ }
+ std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records;
+ rocksdb::ReadOptions options;
+ options.verify_checksums = verify_checksums_in_rocksdb_reads_;
+ std::unique_ptr<rocksdb::Iterator> it(opendb->NewIterator(options));
+ std::string last_event_id;
+ if (event_cursor) {
+ last_event_id = event_cursor->event_id_;
+ it->Seek(event_cursor->event_id_);
+ if (it->Valid() && it->key() == event_cursor->event_id_) {
+ it->Next();
+ }
+ } else {
+ it->SeekToFirst();
+ }
+ for (; it->Valid(); it->Next()) {
+ last_event_id = it->key().ToString();
+ auto eventRead = ProvenanceEventRecord::create();
+ const auto slice = it->value();
+ io::BufferStream stream(std::as_bytes(std::span(slice.data(), slice.size())));
+ if (eventRead->deserialize(stream)) {
+ records.push_back(eventRead);
+ if (--max_size == 0) {
+ break;
+ }
+ }
+ }
+ if (event_cursor) {
+ event_cursor->event_id_ = last_event_id;
+ }
+ return records;
+}
+
+std::expected<void, std::string> RocksDbProvenanceRepository::appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) {
+ std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> data;
+ data.reserve(events.size());
+ std::lock_guard guard(next_event_id_mtx_);
+ for (auto& event : events) {
+ event->setUUID(next_event_id_++);
+ }
+ {
+ auto open_state_db = internal_state_db_->open();
+ if (!open_state_db) {
+ return std::unexpected{"Failed to open internal state column in provenance database"};
+ }
+ auto operation = [this, &open_state_db]() { return open_state_db->Put(rocksdb::WriteOptions(), NEXT_EVENT_UUID_KEY, next_event_id_.to_string().view()); };
+ if (!ExecuteWithRetry(operation)) {
+ return std::unexpected{"Failed to update next provenance event id"};
+ }
+ }
+ for (auto& event : events) {
+ data.emplace_back(event->getUUIDStr(), std::make_unique<io::BufferStream>());
+ event->serialize(*data.back().second);
+ }
+ if (MultiPut(data)) {
+ return {};
+ }
+
+ return std::unexpected{"Failed to append provenance events"};
+}
+
+REGISTER_RESOURCE_AS(RocksDbProvenanceRepository, InternalResource, ("RocksDbProvenanceRepository", "ProvenanceRepository", "provenancerepository"));
+
+} // namespace org::apache::nifi::minifi::provenance
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.h b/extensions/rocksdb-repos/RocksDbProvenanceRepository.h
similarity index 69%
rename from extensions/rocksdb-repos/ProvenanceRepository.h
rename to extensions/rocksdb-repos/RocksDbProvenanceRepository.h
index 950ecd31c..d841fecd8 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.h
+++ b/extensions/rocksdb-repos/RocksDbProvenanceRepository.h
@@ -29,6 +29,7 @@
#include "core/Core.h"
#include "core/logging/LoggerFactory.h"
#include "minifi-cpp/provenance/Provenance.h"
+#include "minifi-cpp/provenance/ProvenanceRepository.h"
#include "minifi-cpp/utils/Literals.h"
#include "RocksDbRepository.h"
@@ -39,22 +40,22 @@ constexpr auto MAX_PROVENANCE_STORAGE_SIZE = 10_MiB;
constexpr auto MAX_PROVENANCE_ENTRY_LIFE_TIME = std::chrono::minutes(1);
constexpr auto PROVENANCE_PURGE_PERIOD = std::chrono::milliseconds(2500);
-class ProvenanceRepository : public core::repository::RocksDbRepository {
+class RocksDbProvenanceRepository : public core::repository::RocksDbRepository, public ProvenanceRepository {
public:
- ProvenanceRepository(std::string_view name, const utils::Identifier& /*uuid*/)
- : ProvenanceRepository(name) {
+ RocksDbProvenanceRepository(std::string_view name, const utils::Identifier& /*uuid*/)
+ : RocksDbProvenanceRepository(name) {
}
- explicit ProvenanceRepository(std::string_view repo_name = "",
+ explicit RocksDbProvenanceRepository(std::string_view repo_name = "",
std::string directory = PROVENANCE_DIRECTORY,
std::chrono::milliseconds maxPartitionMillis = MAX_PROVENANCE_ENTRY_LIFE_TIME,
int64_t maxPartitionBytes = MAX_PROVENANCE_STORAGE_SIZE,
std::chrono::milliseconds purgePeriod = PROVENANCE_PURGE_PERIOD)
- : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className<ProvenanceRepository>(),
- directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory<ProvenanceRepository>::getLogger()) {
+ : RocksDbRepository(repo_name.length() > 0 ? repo_name : core::className<RocksDbProvenanceRepository>(),
+ directory, maxPartitionMillis, maxPartitionBytes, purgePeriod, core::logging::LoggerFactory<RocksDbProvenanceRepository>::getLogger()) {
}
- ~ProvenanceRepository() override {
+ ~RocksDbProvenanceRepository() override {
stop();
}
@@ -72,17 +73,26 @@ class ProvenanceRepository : public core::repository::RocksDbRepository {
// The repo is cleaned up by itself, there is no need to delete items.
return true;
}
- std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t max_size) override;
void destroy();
- ProvenanceRepository(const ProvenanceRepository &parent) = delete;
+ RocksDbProvenanceRepository(const RocksDbProvenanceRepository &parent) = delete;
- ProvenanceRepository &operator=(const ProvenanceRepository &parent) = delete;
+ RocksDbProvenanceRepository &operator=(const RocksDbProvenanceRepository &parent) = delete;
+
+ std::unique_ptr<Cursor> cursorFromString(std::optional<std::string> cursor_str) override;
+
+ std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t max_size, Cursor* cursor) override;
+
+ std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) override;
private:
// Run function for the thread
void run() override {};
+
+ std::unique_ptr<minifi::internal::RocksDatabase> internal_state_db_;
+ std::mutex next_event_id_mtx_;
+ utils::Identifier next_event_id_;
};
} // namespace org::apache::nifi::minifi::provenance
diff --git a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
index 8f6677c35..154a9961d 100644
--- a/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
+++ b/extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp
@@ -21,7 +21,7 @@
#include <random>
#include <vector>
-#include "ProvenanceRepository.h"
+#include "RocksDbProvenanceRepository.h"
#include "unit/TestBase.h"
#include "unit/Catch.h"
@@ -61,13 +61,24 @@ void verifyMaxKeyCount(const minifi::provenance::ProvenanceRepository& repo, uin
REQUIRE(k < keyCount);
}
+std::vector<std::byte> serializeEvent(minifi::provenance::ProvenanceEventRecord& event) {
+ minifi::io::BufferStream stream;
+ event.serialize(stream);
+ return stream.moveBuffer();
+}
+
+template<typename T>
+void appendAll(std::vector<T>& sink, const std::vector<T>& source) {
+ sink.insert(sink.end(), source.begin(), source.end());
+}
+
TEST_CASE("Test size limit", "[sizeLimitTest]") {
TestController testController;
auto temp_dir = testController.createTempDirectory();
REQUIRE(!temp_dir.empty());
// 60 sec, 100 KB - going to exceed the size limit
- minifi::provenance::ProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1min, TEST_PROVENANCE_STORAGE_SIZE, 1s);
+ minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1min, TEST_PROVENANCE_STORAGE_SIZE, 1s);
auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>();
configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string());
@@ -87,7 +98,7 @@ TEST_CASE("Test time limit", "[timeLimitTest]") {
REQUIRE(!temp_dir.empty());
// 1 sec, 100 MB - going to exceed TTL
- minifi::provenance::ProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+ minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>();
configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string());
@@ -114,3 +125,128 @@ TEST_CASE("Test time limit", "[timeLimitTest]") {
verifyMaxKeyCount(provdb, 400);
}
+
+TEST_CASE("Test query elements after cursor", "[iterationTest]") {
+ TestController testController;
+ auto temp_dir = testController.createTempDirectory();
+ REQUIRE(!temp_dir.empty());
+
+ minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+
+ auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>();
+ configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string());
+
+ REQUIRE(provdb.initialize(configuration));
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> events;
+ for (size_t i = 0; i < 8; ++i) {
+ events.push_back(minifi::provenance::ProvenanceEventRecord::create());
+ }
+
+ REQUIRE(provdb.appendEvents(events));
+
+ auto cursor = provdb.cursorFromString(std::nullopt);
+ REQUIRE(cursor);
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> queried_events;
+
+ appendAll(queried_events, provdb.getEvents(3, cursor.get()).value());
+ REQUIRE(queried_events.size() == 3);
+ appendAll(queried_events, provdb.getEvents(3, cursor.get()).value());
+ REQUIRE(queried_events.size() == 6);
+ appendAll(queried_events, provdb.getEvents(3, cursor.get()).value());
+ REQUIRE(queried_events.size() == 8);
+
+ std::string last_event_id;
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()});
+ last_event_id = queried_events.at(i)->getUUIDStr();
+ REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i)));
+ }
+}
+
+TEST_CASE("Test loading cursor from string", "[cursorSerializationTest]") {
+ TestController testController;
+ auto temp_dir = testController.createTempDirectory();
+ REQUIRE(!temp_dir.empty());
+
+ minifi::provenance::RocksDbProvenanceRepository provdb("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+
+ auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>();
+ configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string());
+
+ REQUIRE(provdb.initialize(configuration));
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> events;
+ for (size_t i = 0; i < 8; ++i) {
+ events.push_back(minifi::provenance::ProvenanceEventRecord::create());
+ }
+
+ REQUIRE(provdb.appendEvents(events));
+
+ auto cursor = provdb.cursorFromString(std::nullopt);
+ REQUIRE(cursor);
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> queried_events;
+
+ appendAll(queried_events, provdb.getEvents(3, cursor.get()).value());
+ REQUIRE(queried_events.size() == 3);
+
+ cursor = provdb.cursorFromString(cursor->toString());
+ REQUIRE(cursor);
+
+ appendAll(queried_events, provdb.getEvents(3, cursor.get()).value());
+ REQUIRE(queried_events.size() == 6);
+
+ std::string last_event_id;
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()});
+ last_event_id = queried_events.at(i)->getUUIDStr();
+ REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i)));
+ }
+}
+
+TEST_CASE("Test opening existing database loads monotonic counter", "[eventUuidMonotonicTest]") {
+ TestController testController;
+ auto temp_dir = testController.createTempDirectory();
+ REQUIRE(!temp_dir.empty());
+
+ auto provdb = std::make_unique<minifi::provenance::RocksDbProvenanceRepository>("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+
+ auto configuration = std::make_shared<org::apache::nifi::minifi::ConfigureImpl>();
+ configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, temp_dir.string());
+
+ REQUIRE(provdb->initialize(configuration));
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> events;
+ for (size_t i = 0; i < 4; ++i) {
+ events.push_back(minifi::provenance::ProvenanceEventRecord::create());
+ }
+
+ REQUIRE(provdb->appendEvents(events));
+
+ provdb = std::make_unique<minifi::provenance::RocksDbProvenanceRepository>("TestProvRepo", temp_dir.string(), 1s, TEST_MAX_PROVENANCE_STORAGE_SIZE, 1s);
+
+ REQUIRE(provdb->initialize(configuration));
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> new_events;
+ for (size_t i = 0; i < 4; ++i) {
+ new_events.push_back(minifi::provenance::ProvenanceEventRecord::create());
+ events.push_back(new_events.back());
+ }
+
+ REQUIRE(provdb->appendEvents(new_events));
+
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> queried_events = provdb->getEvents(8, nullptr).value();
+ REQUIRE(queried_events.size() == 8);
+
+ std::string last_event_id;
+
+ for (size_t i = 0; i < queried_events.size(); ++i) {
+ REQUIRE(last_event_id < std::string{queried_events.at(i)->getUUIDStr()});
+ last_event_id = queried_events.at(i)->getUUIDStr();
+ REQUIRE(serializeEvent(*events.at(i)) == serializeEvent(*queried_events.at(i)));
+ }
+}
diff --git a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
index fca1926bb..bc40ebe19 100644
--- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
+++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
@@ -34,13 +34,13 @@ namespace provenance = minifi::provenance;
using namespace std::literals::chrono_literals;
TEST_CASE("Test Provenance record create", "[Testprovenance::ProvenanceEventRecord]") {
- auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "blah", "blahblah");
+ auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000001").value(), "blahblah");
REQUIRE(record1->getAttributes().empty());
REQUIRE(record1->getAlternateIdentifierUri().empty());
}
TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
- auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype");
+ auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000002").value(), "componenttype");
utils::Identifier eventId = record1->getEventId();
@@ -48,10 +48,10 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv
record1->setDetails(smileyface);
auto sample = 65555ms;
- std::shared_ptr<core::Repository> testRepository = std::make_shared<TestRepository>();
+ auto testRepository = std::make_shared<TestRepository>();
record1->setEventDuration(sample);
- testRepository->storeElement(record1);
+ testRepository->appendEvents({record1});
auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
@@ -64,7 +64,7 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv
}
TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") {
- auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, "componentid", "componenttype");
+ auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, utils::Identifier::parse("00000000-0000-0000-0000-000000000003").value(), "componenttype");
utils::Identifier eventId = record1->getEventId();
std::shared_ptr<minifi::FlowFileRecord> ffr1 = std::make_shared<minifi::FlowFileRecordImpl>();
ffr1->setAttribute("potato", "potatoe");
@@ -73,10 +73,10 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") {
record1->addChildFlowFile(*ffr1);
auto sample = 65555ms;
- std::shared_ptr<core::Repository> testRepository = std::make_shared<TestRepository>();
+ auto testRepository = std::make_shared<TestRepository>();
record1->setEventDuration(sample);
- testRepository->storeElement(record1);
+ testRepository->appendEvents({record1});
auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
@@ -87,20 +87,21 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") {
}
TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
- auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype");
-
- utils::Identifier eventId = record1->getEventId();
+ auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000004").value(), "componenttype");
std::string smileyface = ":)";
record1->setDetails(smileyface);
auto sample = 65555ms;
- std::shared_ptr<core::Repository> testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>();
+ auto testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>();
testRepository->initialize(nullptr);
record1->setEventDuration(sample);
- testRepository->storeElement(record1);
+ testRepository->appendEvents({record1});
+
+ utils::Identifier eventId = record1->getEventId();
+
auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
@@ -113,8 +114,7 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro
}
TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[TestFlowAndProv1]") {
- auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, "componentid", "componenttype");
- utils::Identifier eventId = record1->getEventId();
+ auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CLONE, utils::Identifier::parse("00000000-0000-0000-0000-000000000005").value(), "componenttype");
std::shared_ptr<minifi::FlowFileRecord> ffr1 = std::make_shared<minifi::FlowFileRecordImpl>();
ffr1->setAttribute("potato", "potatoe");
ffr1->setAttribute("tomato", "tomatoe");
@@ -122,11 +122,14 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test
record1->addChildFlowFile(*ffr1);
auto sample = 65555ms;
- std::shared_ptr<core::Repository> testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>();
+ auto testRepository = std::make_shared<core::repository::VolatileProvenanceRepository>();
testRepository->initialize(nullptr);
record1->setEventDuration(sample);
- testRepository->storeElement(record1);
+ testRepository->appendEvents({record1});
+
+ utils::Identifier eventId = record1->getEventId();
+
auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
@@ -137,7 +140,7 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test
}
TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::ProvenanceEventRecordSerializeDeser]") {
- auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, "componentid", "componenttype");
+ auto record1 = std::make_shared<provenance::ProvenanceEventRecordImpl>(provenance::ProvenanceEventRecord::ProvenanceEventType::CREATE, utils::Identifier::parse("00000000-0000-0000-0000-000000000006").value(), "componenttype");
utils::Identifier eventId = record1->getEventId();
@@ -146,11 +149,11 @@ TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::Provena
auto sample = 65555ms;
- std::shared_ptr<core::Repository> testRepository = core::createRepository("nooprepository");
+ std::shared_ptr testRepository = utils::dynamic_unique_cast<minifi::provenance::ProvenanceRepository>(core::createRepository("nooprepository"));
testRepository->initialize(nullptr);
record1->setEventDuration(sample);
- REQUIRE(testRepository->storeElement(record1));
+ REQUIRE(testRepository->appendEvents({record1}));
auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == false);
diff --git a/extensions/rocksdb-repos/tests/RepoTests.cpp b/extensions/rocksdb-repos/tests/RepoTests.cpp
index f3a8fd76f..771c8f2a8 100644
--- a/extensions/rocksdb-repos/tests/RepoTests.cpp
+++ b/extensions/rocksdb-repos/tests/RepoTests.cpp
@@ -28,7 +28,7 @@
#include "core/RepositoryFactory.h"
#include "FlowFileRecord.h"
#include "FlowFileRepository.h"
-#include "ProvenanceRepository.h"
+#include "RocksDbProvenanceRepository.h"
#include "properties/Configure.h"
#include "unit/ProvenanceTestHelper.h"
#include "unit/TestBase.h"
@@ -280,7 +280,7 @@ TEST_CASE("Test FlowFile Restore", "[TestFFR6]") {
config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string());
config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string());
- std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>();
+ auto prov_repo = std::make_shared<TestThreadedRepository>();
auto ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository");
std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::FileSystemRepository>();
ff_repository->initialize(config);
@@ -549,8 +549,8 @@ TEST_CASE("Test getting flow file repository size properties", "[TestGettingRepo
expected_rocksdb_stats = true;
}
- SECTION("ProvenanceRepository") {
- repository = std::make_shared<minifi::provenance::ProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms);
+ SECTION("RocksDbProvenanceRepository") {
+ repository = std::make_shared<minifi::provenance::RocksDbProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms);
expected_rocksdb_stats = true;
}
@@ -707,8 +707,8 @@ TEST_CASE("Flow file repositories can be stopped", "[TestRepoIsRunning]") {
repository = std::make_shared<core::repository::FlowFileRepository>("ff", dir.string(), 0ms, 0, 1ms);
}
- SECTION("ProvenanceRepository") {
- repository = std::make_shared<minifi::provenance::ProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms);
+ SECTION("RocksDbProvenanceRepository") {
+ repository = std::make_shared<minifi::provenance::RocksDbProvenanceRepository>("ff", dir.string(), 0ms, 0, 1ms);
}
SECTION("VolatileProvenanceRepository") {
diff --git a/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h b/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h
index 0f6c74a03..499e5e41b 100644
--- a/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h
+++ b/extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h
@@ -85,7 +85,7 @@ class VerifyInvokeHTTP : public HTTPIntegrationBase {
virtual void setupFlow() {
testSetup();
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>();
+ auto test_repo = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
if (flow_config_path_.config_path) {
diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
index 6dd308c7c..3f3c861b0 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -468,11 +468,12 @@ TEST_CASE("Test Find file", "[getfileCreate3]") {
taskReport->setBatchSize(1);
processorReport->incrementActiveTasks();
processorReport->setScheduledState(core::ScheduledState::RUNNING);
- auto recordsReport = repo->getElements(1);
+ auto recordsReport = repo->getEvents(1, nullptr);
+ REQUIRE(recordsReport);
std::function<void(const std::shared_ptr<core::ProcessContext> &, const std::shared_ptr<core::ProcessSession>&)> verifyReporter =
[&](const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
- auto json_str = taskReport->getJsonReport(*context, *session, recordsReport);
- REQUIRE(recordsReport.size() == 1);
+ auto json_str = taskReport->getJsonReport(*context, *session, recordsReport.value());
+ REQUIRE(recordsReport->size() == 1);
REQUIRE(taskReport->getName() == std::string(minifi::core::reporting::SiteToSiteProvenanceReportingTask::ReportTaskName));
REQUIRE(json_str.find("\"componentType\": \"getfileCreate2\"") != std::string::npos);
};
diff --git a/libminifi/include/CronDrivenSchedulingAgent.h b/libminifi/include/CronDrivenSchedulingAgent.h
index d08ac5a4f..46b0d68c0 100644
--- a/libminifi/include/CronDrivenSchedulingAgent.h
+++ b/libminifi/include/CronDrivenSchedulingAgent.h
@@ -37,7 +37,7 @@ namespace org::apache::nifi::minifi {
class CronDrivenSchedulingAgent : public ThreadedSchedulingAgent {
public:
CronDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider,
- std::shared_ptr<core::Repository> repo,
+ std::shared_ptr<provenance::ProvenanceRepository> repo,
std::shared_ptr<core::Repository> flow_repo,
std::shared_ptr<core::ContentRepository> content_repo,
std::shared_ptr<Configure> configuration,
diff --git a/libminifi/include/EventDrivenSchedulingAgent.h b/libminifi/include/EventDrivenSchedulingAgent.h
index 060fff588..d5ccd54b2 100644
--- a/libminifi/include/EventDrivenSchedulingAgent.h
+++ b/libminifi/include/EventDrivenSchedulingAgent.h
@@ -35,7 +35,7 @@ namespace org::apache::nifi::minifi {
class EventDrivenSchedulingAgent : public ThreadedSchedulingAgent {
public:
- EventDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo,
+ EventDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo,
std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration,
utils::ThreadPool &thread_pool)
: ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) {
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index 00ff3f756..f3f654c1a 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -66,7 +66,7 @@ class ProcessorController;
class FlowController : public core::controller::ForwardingControllerServiceProvider, public state::StateMonitor {
public:
- FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
+ FlowController(std::shared_ptr<provenance::ProvenanceRepository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
std::shared_ptr<Configure> configure, std::shared_ptr<core::FlowConfiguration> flow_configuration,
std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<state::MetricsPublisherStore> metrics_publisher_store = nullptr,
std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>(), std::function<void()> request_restart = []{},
@@ -74,7 +74,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
~FlowController() override;
- virtual std::shared_ptr<core::Repository> getProvenanceRepository() {
+ virtual std::shared_ptr<provenance::ProvenanceRepository> getProvenanceRepository() {
return this->provenance_repo_;
}
@@ -187,7 +187,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
// Thread pool for schedulers
utils::ThreadPool thread_pool_;
std::shared_ptr<Configure> configuration_;
- std::shared_ptr<core::Repository> provenance_repo_;
+ std::shared_ptr<provenance::ProvenanceRepository> provenance_repo_;
std::shared_ptr<core::Repository> flow_file_repo_;
std::shared_ptr<core::ContentRepository> content_repo_;
std::shared_ptr<core::FlowConfiguration> flow_configuration_;
diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h
index d5cab7af5..03e5111bd 100644
--- a/libminifi/include/SchedulingAgent.h
+++ b/libminifi/include/SchedulingAgent.h
@@ -50,7 +50,7 @@ namespace org::apache::nifi::minifi {
class SchedulingAgent {
public:
- SchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_repo,
+ SchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo, std::shared_ptr<core::Repository> flow_repo,
std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, utils::ThreadPool& thread_pool)
: admin_yield_duration_(),
bored_yield_duration_(0),
@@ -122,7 +122,7 @@ class SchedulingAgent {
std::shared_ptr<Configure> configure_;
- std::shared_ptr<core::Repository> repo_;
+ std::shared_ptr<provenance::ProvenanceRepository> repo_;
std::shared_ptr<core::Repository> flow_repo_;
diff --git a/libminifi/include/ThreadedSchedulingAgent.h b/libminifi/include/ThreadedSchedulingAgent.h
index 7a3ad2299..2668bab0d 100644
--- a/libminifi/include/ThreadedSchedulingAgent.h
+++ b/libminifi/include/ThreadedSchedulingAgent.h
@@ -38,7 +38,7 @@ namespace org::apache::nifi::minifi {
*/
class ThreadedSchedulingAgent : public SchedulingAgent {
public:
- ThreadedSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo,
+ ThreadedSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo,
std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo,
std::shared_ptr<Configure> configuration, utils::ThreadPool &thread_pool)
: SchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) {
diff --git a/libminifi/include/TimerDrivenSchedulingAgent.h b/libminifi/include/TimerDrivenSchedulingAgent.h
index 99073ce99..055a81eb7 100644
--- a/libminifi/include/TimerDrivenSchedulingAgent.h
+++ b/libminifi/include/TimerDrivenSchedulingAgent.h
@@ -30,7 +30,7 @@
namespace org::apache::nifi::minifi {
class TimerDrivenSchedulingAgent : public ThreadedSchedulingAgent {
public:
- TimerDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo,
+ TimerDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<provenance::ProvenanceRepository> repo,
std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure,
utils::ThreadPool &thread_pool)
: ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configure, thread_pool) {
diff --git a/libminifi/include/core/ProcessContextImpl.h b/libminifi/include/core/ProcessContextImpl.h
index 9466e69df..0f761013d 100644
--- a/libminifi/include/core/ProcessContextImpl.h
+++ b/libminifi/include/core/ProcessContextImpl.h
@@ -50,11 +50,11 @@ class Processor;
class ProcessContextImpl : public core::VariableRegistryImpl, public virtual ProcessContext {
public:
ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage,
- const std::shared_ptr<core::Repository>& repo, const std::shared_ptr<core::Repository>& flow_repo,
+ const std::shared_ptr<provenance::ProvenanceRepository>& repo, const std::shared_ptr<core::Repository>& flow_repo,
const std::shared_ptr<core::ContentRepository>& content_repo = repository::createFileSystemRepository());
ProcessContextImpl(Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage,
- const std::shared_ptr<core::Repository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<minifi::Configure>& configuration,
+ const std::shared_ptr<provenance::ProvenanceRepository>& repo, const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<minifi::Configure>& configuration,
const std::shared_ptr<core::ContentRepository>& content_repo = repository::createFileSystemRepository());
// Get Processor associated with the Process Context
@@ -85,7 +85,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro
void yield() override;
- std::shared_ptr<core::Repository> getProvenanceRepository() override { return repo_; }
+ std::shared_ptr<provenance::ProvenanceRepository> getProvenanceRepository() override { return repo_; }
/**
* Returns a reference to the content repository for the running instance.
@@ -198,7 +198,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro
std::shared_ptr<logging::Logger> logger_;
controller::ControllerServiceProvider* controller_service_provider_;
std::shared_ptr<core::StateStorage> state_storage_;
- std::shared_ptr<core::Repository> repo_;
+ std::shared_ptr<provenance::ProvenanceRepository> repo_;
std::shared_ptr<core::Repository> flow_repo_;
std::shared_ptr<core::ContentRepository> content_repo_;
Processor& processor_;
diff --git a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
index 29767ad60..b53a78013 100644
--- a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
+++ b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
@@ -45,7 +45,7 @@ class SiteToSiteProvenanceReportingTask : public minifi::RemoteProcessGroupPort
static constexpr char const* ReportTaskName = "SiteToSiteProvenanceReportingTask";
static const char *ProvenanceAppStr;
- static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector<std::shared_ptr<core::SerializableComponent>> &records); // NOLINT
+ static std::string getJsonReport(core::ProcessContext& context, core::ProcessSession& session, const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> &records); // NOLINT
void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override;
diff --git a/libminifi/include/core/repository/NoOpThreadedRepository.h b/libminifi/include/core/repository/NoOpThreadedRepository.h
index df23a1b32..06bc01f37 100644
--- a/libminifi/include/core/repository/NoOpThreadedRepository.h
+++ b/libminifi/include/core/repository/NoOpThreadedRepository.h
@@ -21,10 +21,11 @@
#include <thread>
#include "core/ThreadedRepository.h"
+#include "minifi-cpp/provenance/ProvenanceRepository.h"
namespace org::apache::nifi::minifi::core::repository {
-class NoOpThreadedRepository : public core::ThreadedRepositoryImpl {
+class NoOpThreadedRepository : public core::ThreadedRepositoryImpl, public provenance::ProvenanceRepository {
public:
explicit NoOpThreadedRepository(std::string_view repo_name)
: ThreadedRepositoryImpl(repo_name) {
@@ -47,6 +48,21 @@ class NoOpThreadedRepository : public core::ThreadedRepositoryImpl {
return 0;
}
+ std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>& /*events*/) override {
+ return {};
+ }
+
+ std::unique_ptr<ProvenanceRepository::Cursor> cursorFromString(std::optional<std::string> /*cursor_str*/) override {
+ return nullptr;
+ }
+
+ std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t /*max_size*/, Cursor* cursor) override {
+ if (cursor) {
+ return std::unexpected{"Cursor based query is not supported"};
+ }
+ return {};
+ }
+
private:
void run() override {
}
diff --git a/libminifi/include/core/repository/VolatileProvenanceRepository.h b/libminifi/include/core/repository/VolatileProvenanceRepository.h
index da5df86fc..278f0b8f1 100644
--- a/libminifi/include/core/repository/VolatileProvenanceRepository.h
+++ b/libminifi/include/core/repository/VolatileProvenanceRepository.h
@@ -21,10 +21,11 @@
#include <string_view>
#include "VolatileRepository.h"
+#include "minifi-cpp/provenance/ProvenanceRepository.h"
namespace org::apache::nifi::minifi::core::repository {
-class VolatileProvenanceRepository : public VolatileRepository {
+class VolatileProvenanceRepository : public VolatileRepository, public provenance::ProvenanceRepository {
public:
explicit VolatileProvenanceRepository(std::string_view repo_name = "",
std::string /*dir*/ = REPOSITORY_DIRECTORY,
@@ -38,6 +39,36 @@ class VolatileProvenanceRepository : public VolatileRepository {
stop();
}
+ bool initialize(const std::shared_ptr<Configure> &configure) override {
+ if (!VolatileRepository::initialize(configure)) {
+ return false;
+ }
+ next_event_id_ = utils::IdGenerator::getIdGenerator()->generate();
+ return true;
+ }
+
+ std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>& events) override {
+ std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> data;
+ data.reserve(events.size());
+ std::lock_guard guard(next_event_id_mtx_);
+ for (auto& event : events) {
+ event->setUUID(next_event_id_++);
+ data.emplace_back(event->getUUIDStr(), std::make_unique<io::BufferStream>());
+ event->serialize(*data.back().second);
+ }
+ MultiPut(data);
+
+ return {};
+ }
+
+ std::unique_ptr<ProvenanceRepository::Cursor> cursorFromString(std::optional<std::string> /*cursor_str*/) override {
+ return nullptr;
+ }
+
+ std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t /*max_size*/, Cursor* /*cursor*/) override {
+ return std::unexpected{"Querying events is not yet supported"};
+ }
+
private:
void run() override {
}
@@ -51,6 +82,8 @@ class VolatileProvenanceRepository : public VolatileRepository {
}
std::thread thread_;
+ std::mutex next_event_id_mtx_;
+ utils::Identifier next_event_id_;
};
} // namespace org::apache::nifi::minifi::core::repository
diff --git a/libminifi/include/provenance/Provenance.h b/libminifi/include/provenance/Provenance.h
index 5539dc5c2..91d4b91dd 100644
--- a/libminifi/include/provenance/Provenance.h
+++ b/libminifi/include/provenance/Provenance.h
@@ -38,6 +38,7 @@
#include "utils/Id.h"
#include "utils/TimeUtil.h"
#include "minifi-cpp/provenance/Provenance.h"
+#include "minifi-cpp/provenance/ProvenanceRepository.h"
namespace org::apache::nifi::minifi::provenance {
@@ -54,6 +55,14 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
~ProvenanceEventRecordImpl() override = default;
+ std::optional<uint64_t> getEventOrdinal() const override {
+ return event_ordinal_;
+ }
+
+ void setEventOrdinal(uint64_t event_ordinal) override {
+ event_ordinal_ = event_ordinal;
+ }
+
utils::Identifier getEventId() const override {
return getUUID();
}
@@ -218,6 +227,8 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
protected:
ProvenanceEventType event_type_;
+ // the index of the event
+ std::optional<uint64_t> event_ordinal_;
// Date at which the event was created
std::chrono::system_clock::time_point event_time_{};
// Date at which the flow file entered the flow
@@ -251,7 +262,7 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
class ProvenanceReporterImpl : public virtual ProvenanceReporter {
public:
- ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, utils::Identifier component_id, std::string component_type)
+ ProvenanceReporterImpl(std::shared_ptr<provenance::ProvenanceRepository> repo, utils::Identifier component_id, std::string component_type)
: component_id_(component_id),
component_type_(std::move(component_type)),
logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()),
@@ -266,16 +277,12 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
clear();
}
- std::set<std::shared_ptr<ProvenanceEventRecord>> getEvents() const override {
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> getEvents() const override {
return events_;
}
void add(const std::shared_ptr<ProvenanceEventRecord> &event) override {
- events_.insert(event);
- }
-
- void remove(const std::shared_ptr<ProvenanceEventRecord> &event) override {
- events_.erase(event);
+ events_.push_back(event);
}
void clear() final {
@@ -313,8 +320,8 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
private:
std::shared_ptr<core::logging::Logger> logger_;
- std::set<std::shared_ptr<ProvenanceEventRecord>> events_;
- std::shared_ptr<core::Repository> repo_;
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> events_;
+ std::shared_ptr<provenance::ProvenanceRepository> repo_;
};
} // namespace org::apache::nifi::minifi::provenance
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index 916299805..107c44b37 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -46,7 +46,7 @@
namespace org::apache::nifi::minifi {
-FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
+FlowController::FlowController(std::shared_ptr<provenance::ProvenanceRepository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
std::shared_ptr<Configure> configure, std::shared_ptr<core::FlowConfiguration> flow_configuration,
std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<state::MetricsPublisherStore> metrics_publisher_store,
std::shared_ptr<utils::file::FileSystem> filesystem, std::function<void()> request_restart,
diff --git a/libminifi/src/core/ProcessContextImpl.cpp b/libminifi/src/core/ProcessContextImpl.cpp
index 841b74ca5..6ac0372cf 100644
--- a/libminifi/src/core/ProcessContextImpl.cpp
+++ b/libminifi/src/core/ProcessContextImpl.cpp
@@ -41,7 +41,7 @@ class StandardProcessorInfo : public ProcessorInfo {
} // namespace
ProcessContextImpl::ProcessContextImpl(
- Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<core::Repository>& repo,
+ Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<provenance::ProvenanceRepository>& repo,
const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<core::ContentRepository>& content_repo)
: VariableRegistryImpl(static_cast<std::shared_ptr<Configure>>(minifi::Configure::create())),
logger_(logging::LoggerFactory<ProcessContext>::getLogger()),
@@ -55,7 +55,7 @@ ProcessContextImpl::ProcessContextImpl(
info_(std::make_unique<StandardProcessorInfo>(processor)) {}
ProcessContextImpl::ProcessContextImpl(
- Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<core::Repository>& repo,
+ Processor& processor, controller::ControllerServiceProvider* controller_service_provider, const std::shared_ptr<core::StateStorage>& state_storage, const std::shared_ptr<provenance::ProvenanceRepository>& repo,
const std::shared_ptr<core::Repository>& flow_repo, const std::shared_ptr<minifi::Configure>& configuration,
const std::shared_ptr<core::ContentRepository>& content_repo)
: VariableRegistryImpl(configuration),
diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp b/libminifi/src/core/flow/StructuredConfiguration.cpp
index e64b5a34d..d6a94a330 100644
--- a/libminifi/src/core/flow/StructuredConfiguration.cpp
+++ b/libminifi/src/core/flow/StructuredConfiguration.cpp
@@ -586,6 +586,10 @@ void StructuredConfiguration::parseProvenanceReporting(const Node& node, core::P
auto report_task = createProvenanceReportTask();
+ utils::Identifier task_uuid;
+ task_uuid = getOrGenerateId(node);
+ report_task->setUUID(task_uuid);
+
checkRequiredField(node, schema_.scheduling_strategy);
auto schedulingStrategyStr = node[schema_.scheduling_strategy].getString().value();
checkRequiredField(node, schema_.scheduling_period);
diff --git a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
index 6fb3d5d3c..38bc15d64 100644
--- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
+++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
@@ -88,7 +88,7 @@ void appendJsonStr(const utils::SmallString<N>& value, rapidjson::Value& parent,
parent.PushBack(valueVal, alloc);
}
-std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector<std::shared_ptr<core::SerializableComponent>> &records) {
+std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, core::ProcessSession&, const std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> &records) {
rapidjson::Document array(rapidjson::kArrayType);
rapidjson::Document::AllocatorType &alloc = array.GetAllocator();
@@ -111,6 +111,10 @@ std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContex
recordJson.AddMember("entityType", "org.apache.nifi.flowfile.FlowFile", alloc);
+ if (auto event_ordinal = record->getEventOrdinal()) {
+ recordJson.AddMember("eventOrdinal", event_ordinal.value(), alloc);
+ }
+
recordJson.AddMember("eventId", getStringValue(record->getEventId().to_string(), alloc), alloc);
recordJson.AddMember("eventType", getStringValue(provenance::ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()], alloc), alloc);
recordJson.AddMember("details", getStringValue(record->getDetails(), alloc), alloc);
@@ -151,17 +155,36 @@ std::string SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContex
}
void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
+ std::shared_ptr<provenance::ProvenanceRepository> repo = context.getProvenanceRepository();
+ if (!repo) {
+ throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository");
+ }
auto* state_manager = context.getStateManager();
if (!state_manager) {
logger_->log_error("Failed to get StateManager");
context.yield();
return;
}
- std::shared_ptr<core::Repository> repo = context.getProvenanceRepository();
- if (!repo) {
- throw minifi::Exception(ExceptionType::REPOSITORY_EXCEPTION, "Failed to retrieve provenance repository");
+ std::optional<std::string> cursor_str;
+ {
+ std::unordered_map<std::string, std::string> state_map;
+ if (state_manager->get(state_map)) {
+ if (auto it = state_map.find("cursor"); it != state_map.end()) {
+ cursor_str = it->second;
+ }
+ }
+ }
+ std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>> records;
+ auto cursor = repo->cursorFromString(cursor_str);
+ if (cursor_str && !cursor) {
+ logger_->log_error("Failed to parse cursor, falling back to enumerating from the beginning");
+ cursor = repo->cursorFromString(std::nullopt);
+ }
+ if (auto result = repo->getEvents(batch_size_, cursor.get())) {
+ records = std::move(result.value());
+ } else {
+ throw minifi::Exception(GENERAL_EXCEPTION, "Failed to retrieve records: " + result.error());
}
- auto records = repo->getElements(batch_size_);
if (records.empty()) {
logger_->log_debug("No new provenance records");
return;
@@ -189,8 +212,22 @@ void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context,
return;
}
- // we transfer the record, purge the record from DB
- repo->Delete(records);
+ if (cursor) {
+ // no need to delete just update the state
+ std::unordered_map<std::string, std::string> state_map;
+ state_map["cursor"] = cursor->toString();
+ if (!state_manager->set(state_map)) {
+ logger_->log_error("Failed to update cursor state");
+ }
+ } else {
+ // we transfer the record, purge the record from DB
+ std::vector<std::shared_ptr<core::SerializableComponent>> entries;
+ entries.reserve(records.size());
+ for (const auto& record : records) {
+ entries.push_back(record);
+ }
+ repo->Delete(entries);
+ }
returnProtocol(context, std::move(protocol_));
}
diff --git a/libminifi/src/provenance/Provenance.cpp b/libminifi/src/provenance/Provenance.cpp
index 66bcff976..8fa948642 100644
--- a/libminifi/src/provenance/Provenance.cpp
+++ b/libminifi/src/provenance/Provenance.cpp
@@ -454,15 +454,7 @@ void ProvenanceReporterImpl::commit() {
return;
}
- std::vector<std::pair<std::string, std::unique_ptr<io::BufferStream>>> flowData;
-
- for (auto& event : events_) {
- auto stramptr = std::make_unique<io::BufferStream>();
- event->serialize(*stramptr);
-
- flowData.emplace_back(event->getUUIDStr(), std::move(stramptr));
- }
- repo_->MultiPut(flowData);
+ repo_->appendEvents(events_);
}
void ProvenanceReporterImpl::create(const core::FlowFile& flow_file, const std::string& detail) {
@@ -543,7 +535,7 @@ void ProvenanceReporterImpl::send(const core::FlowFile& flow_file, const std::st
add(event);
} else {
if (!repo_->isFull())
- repo_->storeElement(event);
+ repo_->appendEvents({event});
}
}
}
diff --git a/libminifi/test/flow-tests/SessionTests.cpp b/libminifi/test/flow-tests/SessionTests.cpp
index 533e3bf56..876a22375 100644
--- a/libminifi/test/flow-tests/SessionTests.cpp
+++ b/libminifi/test/flow-tests/SessionTests.cpp
@@ -62,7 +62,7 @@ TEST_CASE("Import null data") {
config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string());
config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string());
- std::shared_ptr<core::Repository> prov_repo = core::createRepository("nooprepository");
+ std::shared_ptr prov_repo = utils::dynamic_unique_cast<minifi::provenance::ProvenanceRepository>(core::createRepository("nooprepository"));
std::shared_ptr<core::Repository> ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository");
std::shared_ptr<core::ContentRepository> content_repo;
SECTION("VolatileContentRepository") {
diff --git a/libminifi/test/integration/C2PauseResumeTest.cpp b/libminifi/test/integration/C2PauseResumeTest.cpp
index b9e6efe15..39afd2bef 100644
--- a/libminifi/test/integration/C2PauseResumeTest.cpp
+++ b/libminifi/test/integration/C2PauseResumeTest.cpp
@@ -122,7 +122,7 @@ TEST_CASE("C2PauseResumeTest", "[c2test]") {
VerifyC2PauseResume harness{test_file_path, flow_resumed_successfully};
PauseResumeHandler responder{flow_resumed_successfully, harness.getConfiguration()};
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>();
+ auto test_repo = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::ConfigureImpl>();
configuration->set(minifi::Configure::nifi_flow_configuration_file, test_file_path.string());
diff --git a/libminifi/test/integration/ControllerServiceIntegrationTests.cpp b/libminifi/test/integration/ControllerServiceIntegrationTests.cpp
index d5586ca73..99f80b401 100644
--- a/libminifi/test/integration/ControllerServiceIntegrationTests.cpp
+++ b/libminifi/test/integration/ControllerServiceIntegrationTests.cpp
@@ -52,7 +52,7 @@ TEST_CASE("ControllerServiceIntegrationTests", "[controller]") {
using org::apache::nifi::minifi::test::utils::verifyEventHappenedInPollTime;
std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::ConfigureImpl>();
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>();
+ auto test_repo = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
const auto test_file_path = std::filesystem::path(TEST_RESOURCES) / "TestControllerServices.yml";
diff --git a/libminifi/test/libtest/integration/IntegrationBase.cpp b/libminifi/test/libtest/integration/IntegrationBase.cpp
index 8a767546a..b70ffa066 100644
--- a/libminifi/test/libtest/integration/IntegrationBase.cpp
+++ b/libminifi/test/libtest/integration/IntegrationBase.cpp
@@ -59,7 +59,7 @@ void IntegrationBase::run() {
using namespace std::literals::chrono_literals;
testSetup();
- std::shared_ptr<core::Repository> test_repo = std::make_shared<TestThreadedRepository>();
+ auto test_repo = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>();
if (flow_config_path_.config_path) {
diff --git a/libminifi/test/libtest/unit/ProvenanceTestHelper.h b/libminifi/test/libtest/unit/ProvenanceTestHelper.h
index 03d6e0c32..1901c707a 100644
--- a/libminifi/test/libtest/unit/ProvenanceTestHelper.h
+++ b/libminifi/test/libtest/unit/ProvenanceTestHelper.h
@@ -41,7 +41,7 @@ using namespace std::literals::chrono_literals;
const int64_t TEST_MAX_REPOSITORY_STORAGE_SIZE = 100;
template <typename T_BaseRepository>
-class TestRepositoryBase : public T_BaseRepository {
+class TestRepositoryBase : public T_BaseRepository, public org::apache::nifi::minifi::provenance::ProvenanceRepository {
public:
TestRepositoryBase()
: T_BaseRepository("repo_name", "./dir", 1s, TEST_MAX_REPOSITORY_STORAGE_SIZE, 0ms) {
@@ -89,11 +89,14 @@ class TestRepositoryBase : public T_BaseRepository {
}
}
- std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> getElements(size_t max_size) override {
+ std::expected<std::vector<std::shared_ptr<org::apache::nifi::minifi::provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t max_size, Cursor* cursor) override {
+ if (cursor) {
+ return std::unexpected{"Cursor based query is not supported"};
+ }
if (max_size == 0) {
return {};
}
- std::vector<std::shared_ptr<org::apache::nifi::minifi::core::SerializableComponent>> store;
+ std::vector<std::shared_ptr<org::apache::nifi::minifi::provenance::ProvenanceEventRecord>> store;
std::lock_guard<std::mutex> lock{repository_results_mutex_};
for (const auto &entry : repository_results_) {
const auto eventRead = org::apache::nifi::minifi::provenance::ProvenanceEventRecord::create();
@@ -113,6 +116,24 @@ class TestRepositoryBase : public T_BaseRepository {
return repository_results_;
}
+ std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<org::apache::nifi::minifi::provenance::ProvenanceEventRecord>>& events) override {
+ std::vector<std::pair<std::string, std::unique_ptr<org::apache::nifi::minifi::io::BufferStream>>> data;
+ data.reserve(events.size());
+ for (auto& event : events) {
+ data.emplace_back(event->getUUIDStr(), std::make_unique<org::apache::nifi::minifi::io::BufferStream>());
+ event->serialize(*data.back().second);
+ }
+ if (!MultiPut(data)) {
+ return std::unexpected{"Failed to store provenance events"};
+ }
+
+ return {};
+ }
+
+ std::unique_ptr<ProvenanceRepository::Cursor> cursorFromString(std::optional<std::string> /*cursor_str*/) override {
+ return nullptr;
+ }
+
protected:
mutable std::mutex repository_results_mutex_;
std::map<std::string, std::string> repository_results_;
diff --git a/libminifi/test/libtest/unit/TestBase.cpp b/libminifi/test/libtest/unit/TestBase.cpp
index 23a967ae9..c5477e67c 100644
--- a/libminifi/test/libtest/unit/TestBase.cpp
+++ b/libminifi/test/libtest/unit/TestBase.cpp
@@ -208,7 +208,7 @@ LogTestController::LogTestController(const std::shared_ptr<logging::LoggerProper
init(loggerProps);
}
-TestPlan::TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::core::Repository> prov_repo,
+TestPlan::TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::provenance::ProvenanceRepository> prov_repo,
std::shared_ptr<minifi::state::response::FlowVersion> flow_version, std::shared_ptr<minifi::Configure> configuration, const char* state_dir)
: configuration_(std::move(configuration)),
content_repo_(std::move(content_repo)),
@@ -637,7 +637,7 @@ std::shared_ptr<minifi::core::FlowFile> TestPlan::getFlowFileProducedByCurrentPr
return nullptr;
}
-std::set<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> TestPlan::getProvenanceRecords() {
+std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> TestPlan::getProvenanceRecords() {
return process_sessions_.at(location)->getProvenanceReporter()->getEvents();
}
diff --git a/libminifi/test/libtest/unit/TestBase.h b/libminifi/test/libtest/unit/TestBase.h
index 9e5557194..b2d20cde8 100644
--- a/libminifi/test/libtest/unit/TestBase.h
+++ b/libminifi/test/libtest/unit/TestBase.h
@@ -243,7 +243,7 @@ class TypedProcessorWrapper {
class TestPlan {
public:
- explicit TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::core::Repository> prov_repo,
+ explicit TestPlan(std::shared_ptr<minifi::core::ContentRepository> content_repo, std::shared_ptr<minifi::core::Repository> flow_repo, std::shared_ptr<minifi::provenance::ProvenanceRepository> prov_repo,
std::shared_ptr<minifi::state::response::FlowVersion> flow_version, std::shared_ptr<minifi::Configure> configuration, const char* state_dir);
virtual ~TestPlan();
@@ -305,7 +305,7 @@ class TestPlan {
bool runCurrentProcessor();
bool runCurrentProcessorUntilFlowfileIsProduced(std::chrono::milliseconds wait_duration);
- std::set<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> getProvenanceRecords();
+ std::vector<std::shared_ptr<minifi::provenance::ProvenanceEventRecord>> getProvenanceRecords();
std::shared_ptr<minifi::core::FlowFile> getCurrentFlowFile();
std::vector<minifi::Connection*> getProcessorOutboundConnections(minifi::core::Processor* processor);
@@ -357,7 +357,7 @@ class TestPlan {
std::shared_ptr<minifi::core::ContentRepository> content_repo_;
std::shared_ptr<minifi::core::Repository> flow_repo_;
- std::shared_ptr<minifi::core::Repository> prov_repo_;
+ std::shared_ptr<minifi::provenance::ProvenanceRepository> prov_repo_;
std::shared_ptr<minifi::core::controller::ControllerServiceProvider> controller_services_provider_;
diff --git a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp
index 26f5408b4..4d8b6988a 100644
--- a/libminifi/test/libtest/unit/TestControllerWithFlow.cpp
+++ b/libminifi/test/libtest/unit/TestControllerWithFlow.cpp
@@ -55,7 +55,7 @@ TestControllerWithFlow::TestControllerWithFlow(const char* yamlConfigContent, bo
}
void TestControllerWithFlow::setupFlow() {
- std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>();
+ auto prov_repo = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::Repository> ff_repo = std::make_shared<TestFlowRepository>();
std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
diff --git a/libminifi/test/persistence-tests/PersistenceTests.cpp b/libminifi/test/persistence-tests/PersistenceTests.cpp
index 11538a8f0..70642304d 100644
--- a/libminifi/test/persistence-tests/PersistenceTests.cpp
+++ b/libminifi/test/persistence-tests/PersistenceTests.cpp
@@ -60,7 +60,7 @@ class TestProcessor : public minifi::core::ProcessorImpl {
};
struct TestFlow{
- TestFlow(const std::shared_ptr<core::Repository>& ff_repository, const std::shared_ptr<core::ContentRepository>& content_repo, const std::shared_ptr<core::Repository>& prov_repo,
+ TestFlow(const std::shared_ptr<core::Repository>& ff_repository, const std::shared_ptr<core::ContentRepository>& content_repo, const std::shared_ptr<minifi::provenance::ProvenanceRepository>& prov_repo,
const std::function<std::unique_ptr<core::Processor>(utils::Identifier&)>& processorGenerator, const core::Relationship& relationshipToOutput)
: ff_repository(ff_repository), content_repo(content_repo), prov_repo(prov_repo) {
// setup processor
@@ -178,7 +178,7 @@ TEST_CASE("Processors Can Store FlowFiles", "[TestP1]") {
config->set(minifi::Configure::nifi_dbcontent_repository_directory_default, (dir / "content_repository").string());
config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string());
- std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>();
+ auto prov_repo = std::make_shared<TestThreadedRepository>();
auto ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository");
std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::FileSystemRepository>();
ff_repository->initialize(config);
@@ -292,7 +292,7 @@ TEST_CASE("Persisted flowFiles are updated on modification", "[TestP1]") {
config->set(minifi::Configure::nifi_flowfile_repository_directory_default, (dir / "flowfile_repository").string());
config->set(minifi::Configure::nifi_dbcontent_repository_purge_period, "0 s");
- std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestThreadedRepository>();
+ auto prov_repo = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::Repository> ff_repository = std::make_shared<core::repository::FlowFileRepository>("flowFileRepository");
std::shared_ptr<core::ContentRepository> content_repo;
SECTION("VolatileContentRepository") {
diff --git a/libminifi/test/unit/SchedulingAgentTests.cpp b/libminifi/test/unit/SchedulingAgentTests.cpp
index 0e087cf1a..8c2f0fee4 100644
--- a/libminifi/test/unit/SchedulingAgentTests.cpp
+++ b/libminifi/test/unit/SchedulingAgentTests.cpp
@@ -78,7 +78,7 @@ class SchedulingAgentTestFixture {
}
protected:
- std::shared_ptr<core::Repository> test_repo_ = std::make_shared<TestThreadedRepository>();
+ std::shared_ptr<provenance::ProvenanceRepository> test_repo_ = std::make_shared<TestThreadedRepository>();
std::shared_ptr<core::ContentRepository> content_repo_ = std::make_shared<core::repository::VolatileContentRepository>();
TestController test_controller_;
diff --git a/minifi-api/common/include/minifi-cpp/utils/Id.h b/minifi-api/common/include/minifi-cpp/utils/Id.h
index 5c435dcf9..95a675985 100644
--- a/minifi-api/common/include/minifi-cpp/utils/Id.h
+++ b/minifi-api/common/include/minifi-cpp/utils/Id.h
@@ -46,6 +46,9 @@ class Identifier {
return !isNil();
}
+ Identifier& operator++();
+ Identifier operator++(int);
+
bool operator!=(const Identifier& other) const;
bool operator==(const Identifier& other) const;
bool operator<(const Identifier& other) const;
diff --git a/minifi-api/include/minifi-cpp/core/ProcessContext.h b/minifi-api/include/minifi-cpp/core/ProcessContext.h
index 6fa663a7a..dd7b086e3 100644
--- a/minifi-api/include/minifi-cpp/core/ProcessContext.h
+++ b/minifi-api/include/minifi-cpp/core/ProcessContext.h
@@ -29,6 +29,7 @@
#include "minifi-cpp/core/StateStorage.h"
#include "minifi-cpp/core/VariableRegistry.h"
#include "minifi-cpp/core/controller/ControllerServiceHandle.h"
+#include "minifi-cpp/provenance/ProvenanceRepository.h"
namespace org::apache::nifi::minifi::core {
@@ -79,7 +80,7 @@ class ProcessContext : public virtual core::VariableRegistry, public virtual uti
virtual bool isRunning() const = 0;
virtual bool isAutoTerminated(Relationship relationship) const = 0;
virtual uint8_t getMaxConcurrentTasks() const = 0;
- virtual std::shared_ptr<core::Repository> getProvenanceRepository() = 0;
+ virtual std::shared_ptr<provenance::ProvenanceRepository> getProvenanceRepository() = 0;
virtual std::shared_ptr<core::ContentRepository> getContentRepository() const = 0;
virtual std::shared_ptr<core::Repository> getFlowFileRepository() const = 0;
diff --git a/minifi-api/include/minifi-cpp/core/Repository.h b/minifi-api/include/minifi-cpp/core/Repository.h
index 33b5c3747..6158d9d51 100644
--- a/minifi-api/include/minifi-cpp/core/Repository.h
+++ b/minifi-api/include/minifi-cpp/core/Repository.h
@@ -54,10 +54,6 @@ class Repository : public virtual core::CoreComponent, public virtual core::Repo
virtual bool Get(const std::string& /*key*/, std::string& /*value*/) = 0;
- virtual std::vector<std::shared_ptr<core::SerializableComponent>> getElements(size_t max_size) = 0;
-
- virtual bool storeElement(const std::shared_ptr<core::SerializableComponent>& element) = 0;
-
virtual void loadComponent(const std::shared_ptr<core::ContentRepository>& /*content_repo*/) = 0;
virtual std::string getDirectory() const = 0;
diff --git a/minifi-api/include/minifi-cpp/provenance/Provenance.h b/minifi-api/include/minifi-cpp/provenance/Provenance.h
index ef396958a..4cf5f1924 100644
--- a/minifi-api/include/minifi-cpp/provenance/Provenance.h
+++ b/minifi-api/include/minifi-cpp/provenance/Provenance.h
@@ -148,6 +148,8 @@ class ProvenanceEventRecord : public virtual core::SerializableComponent {
~ProvenanceEventRecord() override = default;
+ virtual std::optional<uint64_t> getEventOrdinal() const = 0;
+ virtual void setEventOrdinal(uint64_t value) = 0;
virtual utils::Identifier getEventId() const = 0;
virtual void setEventId(const utils::Identifier &id) = 0;
virtual std::map<std::string, std::string> getAttributes() const = 0;
@@ -192,9 +194,8 @@ class ProvenanceReporter {
public:
virtual ~ProvenanceReporter() = default;
- virtual std::set<std::shared_ptr<ProvenanceEventRecord>> getEvents() const = 0;
+ virtual std::vector<std::shared_ptr<ProvenanceEventRecord>> getEvents() const = 0;
virtual void add(const std::shared_ptr<ProvenanceEventRecord> &event) = 0;
- virtual void remove(const std::shared_ptr<ProvenanceEventRecord> &event) = 0;
virtual void clear() = 0;
virtual void commit() = 0;
diff --git a/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h b/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h
new file mode 100644
index 000000000..cfebebe35
--- /dev/null
+++ b/minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h
@@ -0,0 +1,44 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include "minifi-cpp/core/Repository.h"
+#include "utils/Id.h"
+#include "minifi-cpp/provenance/Provenance.h"
+
+namespace org::apache::nifi::minifi::provenance {
+
+class ProvenanceRepository : public virtual core::Repository {
+ public:
+ class Cursor {
+ public:
+ [[nodiscard]]
+ virtual std::string toString() const = 0;
+ virtual ~Cursor() = default;
+ };
+
+ virtual std::expected<void, std::string> appendEvents(const std::vector<std::shared_ptr<ProvenanceEventRecord>>& events) = 0;
+
+ // if @cursor_str is nullopt it creates a cursor to the beginning if the underlying repository supports it
+ virtual std::unique_ptr<Cursor> cursorFromString(std::optional<std::string> cursor_str) = 0;
+
+ virtual std::expected<std::vector<std::shared_ptr<provenance::ProvenanceEventRecord>>, std::string> getEvents(size_t max_size, Cursor* cursor) = 0;
+};
+
+} // namespace org::apache::nifi::minifi::provenance
diff --git a/minifi_main/MiNiFiMain.cpp b/minifi_main/MiNiFiMain.cpp
index fcc2ba9d6..ea7e44fdc 100644
--- a/minifi_main/MiNiFiMain.cpp
+++ b/minifi_main/MiNiFiMain.cpp
@@ -343,7 +343,7 @@ int main(int argc, char **argv) {
configure->get(minifi::Configure::nifi_provenance_repository_class_name, prov_repo_class);
// Create repos for flow record and provenance
- std::shared_ptr prov_repo = core::createRepository(prov_repo_class, "provenance");
+ std::shared_ptr prov_repo = utils::dynamic_unique_cast<minifi::provenance::ProvenanceRepository>(core::createRepository(prov_repo_class, "provenance"));
if (!prov_repo || !prov_repo->initialize(configure)) {
logger->log_error("Provenance repository failed to initialize, exiting..");
(nifi-minifi-cpp) 04/04: MINIFICPP-2845 - Remove extra file
Posted by ad...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
adamdebreceni pushed a commit to branch MINIFICPP-2845
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit f4ccd18546a6c48af957a37801cb3143bab4b208
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Mon Jul 6 11:31:30 2026 +0200
MINIFICPP-2845 - Remove extra file
---
cmake/FetchSQLite.cmake | 29 -----------------------------
1 file changed, 29 deletions(-)
diff --git a/cmake/FetchSQLite.cmake b/cmake/FetchSQLite.cmake
deleted file mode 100644
index 65a9fb62f..000000000
--- a/cmake/FetchSQLite.cmake
+++ /dev/null
@@ -1,29 +0,0 @@
-# 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(FetchContent)
-
-FetchContent_Declare(
- sqlite3
- URL "https://www.sqlite.org/2026/sqlite-amalgamation-3530200.zip"
- URL_HASH "SHA256=8a310d0a16c7a90cacd4c884e70faa51c902afed2a89f63aaa0126ab83558a32"
-)
-
-FetchContent_MakeAvailable(sqlite3)
-
-add_library(sqlite3 STATIC ${sqlite3_SOURCE_DIR}/sqlite3.c)
-target_include_directories(sqlite3 PUBLIC ${sqlite3_SOURCE_DIR})
(nifi-minifi-cpp) 02/04: MINIFICPP-2840 - More refactor
Posted by ad...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
adamdebreceni pushed a commit to branch MINIFICPP-2845
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit c1f65b296f97d6d1e273ae17550459fdb0dd754a
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Wed Jun 3 14:14:09 2026 +0200
MINIFICPP-2840 - More refactor
---
extensions/rocksdb-repos/tests/ProvenanceTests.cpp | 10 +--
.../tests/unit/ProcessorTests.cpp | 16 ++---
libminifi/include/provenance/Provenance.h | 39 +++++-----
libminifi/src/core/ProcessSession.cpp | 2 +-
libminifi/src/provenance/Provenance.cpp | 83 +++++++++++-----------
5 files changed, 73 insertions(+), 77 deletions(-)
diff --git a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
index ecd3a77cf..fca1926bb 100644
--- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
+++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp
@@ -52,7 +52,7 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv
record1->setEventDuration(sample);
testRepository->storeElement(record1);
- auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>();
+ auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
REQUIRE(record2->getEventId() == record1->getEventId());
@@ -77,7 +77,7 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") {
record1->setEventDuration(sample);
testRepository->storeElement(record1);
- auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>();
+ auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
REQUIRE(record1->getChildrenUuids().size() == 1);
@@ -101,7 +101,7 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro
record1->setEventDuration(sample);
testRepository->storeElement(record1);
- auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>();
+ auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
REQUIRE(record2->getEventId() == record1->getEventId());
@@ -127,7 +127,7 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test
record1->setEventDuration(sample);
testRepository->storeElement(record1);
- auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>();
+ auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == true);
REQUIRE(record1->getChildrenUuids().size() == 1);
@@ -151,7 +151,7 @@ TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::Provena
record1->setEventDuration(sample);
REQUIRE(testRepository->storeElement(record1));
- auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>();
+ auto record2 = provenance::ProvenanceEventRecord::create();
record2->setEventId(eventId);
REQUIRE(record2->loadFromRepository(testRepository) == false);
}
diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
index cb5905af2..6dd308c7c 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -444,18 +444,18 @@ TEST_CASE("Test Find file", "[getfileCreate3]") {
REQUIRE(2 == repo->getRepoMap().size());
for (auto entry : repo->getRepoMap()) {
- minifi::provenance::ProvenanceEventRecordImpl newRecord;
+ auto newRecord = minifi::provenance::ProvenanceEventRecordImpl::create();
minifi::io::BufferStream stream(std::as_bytes(std::span(entry.second)));
- newRecord.deserialize(stream);
+ newRecord->deserialize(stream);
bool found = false;
for (const auto& provRec : records) {
- if (provRec->getEventId() == newRecord.getEventId()) {
- REQUIRE(provRec->getEventId() == newRecord.getEventId());
- REQUIRE(provRec->getComponentId() == newRecord.getComponentId());
- REQUIRE(provRec->getComponentType() == newRecord.getComponentType());
- REQUIRE(provRec->getDetails() == newRecord.getDetails());
- REQUIRE(provRec->getEventDuration() == newRecord.getEventDuration());
+ if (provRec->getEventId() == newRecord->getEventId()) {
+ REQUIRE(provRec->getEventId() == newRecord->getEventId());
+ REQUIRE(provRec->getComponentId() == newRecord->getComponentId());
+ REQUIRE(provRec->getComponentType() == newRecord->getComponentType());
+ REQUIRE(provRec->getDetails() == newRecord->getDetails());
+ REQUIRE(provRec->getEventDuration() == newRecord->getEventDuration());
found = true;
break;
}
diff --git a/libminifi/include/provenance/Provenance.h b/libminifi/include/provenance/Provenance.h
index 9d6aa11e2..5539dc5c2 100644
--- a/libminifi/include/provenance/Provenance.h
+++ b/libminifi/include/provenance/Provenance.h
@@ -45,12 +45,12 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
public:
static const char *ProvenanceEventTypeStr[REPLAY + 1];
- ProvenanceEventRecordImpl(ProvenanceEventType event, std::string componentId, std::string componentType);
+ ProvenanceEventRecordImpl(ProvenanceEventType event, utils::Identifier component_id, std::string component_type);
- ProvenanceEventRecordImpl()
- : core::SerializableComponentImpl(core::className<ProvenanceEventRecord>()) {
- event_time_ = std::chrono::system_clock::now();
- }
+ ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl&) = delete;
+ ProvenanceEventRecordImpl(ProvenanceEventRecordImpl&&) = delete;
+ ProvenanceEventRecordImpl& operator=(const ProvenanceEventRecordImpl&) = delete;
+ ProvenanceEventRecordImpl& operator=(ProvenanceEventRecordImpl&&) = delete;
~ProvenanceEventRecordImpl() override = default;
@@ -245,20 +245,22 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public
std::string alternate_identifier_uri_;
private:
- ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl &parent);
- ProvenanceEventRecordImpl &operator=(const ProvenanceEventRecordImpl &parent);
static std::shared_ptr<core::logging::Logger> logger_;
static std::shared_ptr<utils::IdGenerator> id_generator_;
};
class ProvenanceReporterImpl : public virtual ProvenanceReporter {
public:
- ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, std::string componentId, std::string componentType)
- : logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()) {
- component_id_ = componentId;
- component_type_ = componentType;
- repo_ = repo;
- }
+ ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, utils::Identifier component_id, std::string component_type)
+ : component_id_(component_id),
+ component_type_(std::move(component_type)),
+ logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()),
+ repo_(std::move(repo)) {}
+
+ ProvenanceReporterImpl(const ProvenanceReporterImpl&) = delete;
+ ProvenanceReporterImpl(ProvenanceReporterImpl&&) = delete;
+ ProvenanceReporterImpl& operator=(const ProvenanceReporterImpl&) = delete;
+ ProvenanceReporterImpl& operator=(ProvenanceReporterImpl&&) = delete;
~ProvenanceReporterImpl() override {
clear();
@@ -273,12 +275,10 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
}
void remove(const std::shared_ptr<ProvenanceEventRecord> &event) override {
- if (events_.find(event) != events_.end()) {
- events_.erase(event);
- }
+ events_.erase(event);
}
- void clear() override {
+ void clear() final {
events_.clear();
}
@@ -308,16 +308,13 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter {
return event;
}
- std::string component_id_;
+ utils::Identifier component_id_;
std::string component_type_;
private:
std::shared_ptr<core::logging::Logger> logger_;
std::set<std::shared_ptr<ProvenanceEventRecord>> events_;
std::shared_ptr<core::Repository> repo_;
-
- ProvenanceReporterImpl(const ProvenanceReporterImpl &parent);
- ProvenanceReporterImpl &operator=(const ProvenanceReporterImpl &parent);
};
} // namespace org::apache::nifi::minifi::provenance
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index d44c71947..31c6a6b29 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -69,7 +69,7 @@ ProcessSessionImpl::ProcessSessionImpl(std::shared_ptr<ProcessContext> processCo
stateManager_(process_context_->getStateManager()) {
logger_->log_trace("ProcessSession created for {}", process_context_->getProcessor().getName());
auto repo = process_context_->getProvenanceRepository();
- provenance_report_ = std::make_shared<provenance::ProvenanceReporterImpl>(repo, process_context_->getProcessor().getUUIDStr(), process_context_->getProcessor().getName());
+ provenance_report_ = std::make_shared<provenance::ProvenanceReporterImpl>(repo, process_context_->getProcessor().getUUID(), process_context_->getProcessor().getName());
content_session_ = process_context_->getContentRepository()->createSession();
if (stateManager_ && !stateManager_->beginTransaction()) {
diff --git a/libminifi/src/provenance/Provenance.cpp b/libminifi/src/provenance/Provenance.cpp
index ed4d0638a..66bcff976 100644
--- a/libminifi/src/provenance/Provenance.cpp
+++ b/libminifi/src/provenance/Provenance.cpp
@@ -40,12 +40,12 @@ std::shared_ptr<core::logging::Logger> ProvenanceEventRecordImpl::logger_ = core
const char *ProvenanceEventRecord::ProvenanceEventTypeStr[REPLAY + 1] = { "CREATE", "RECEIVE", "FETCH", "SEND", "DOWNLOAD", // NOLINT(cppcoreguidelines-avoid-c-arrays)
"DROP", "EXPIRE", "FORK", "JOIN", "CLONE", "CONTENT_MODIFIED", "ATTRIBUTES_MODIFIED", "ROUTE", "ADDINFO", "REPLAY" };
-ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, std::string componentId, std::string componentType)
+ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, utils::Identifier component_id, std::string component_type)
: core::SerializableComponentImpl(core::className<ProvenanceEventRecord>()),
event_type_(event),
event_time_(std::chrono::system_clock::now()),
- component_id_(std::move(componentId)),
- component_type_(std::move(componentType)) {
+ component_id_(std::move(component_id.to_string())),
+ component_type_(std::move(component_type)) {
}
bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr<core::Repository> &repo) {
@@ -80,13 +80,13 @@ bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr<core::R
bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
{
- const auto ret = output_stream.write(this->uuid_);
+ const auto ret = output_stream.write(uuid_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- uint32_t eventType = this->event_type_;
+ uint32_t eventType = event_type_;
const auto ret = output_stream.write(eventType);
if (ret != 4) {
return false;
@@ -107,7 +107,7 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- uint64_t event_duration_ms = this->event_duration_.count();
+ uint64_t event_duration_ms = event_duration_.count();
const auto ret = output_stream.write(event_duration_ms);
if (ret != 8) {
return false;
@@ -121,32 +121,32 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- const auto ret = output_stream.write(this->component_id_);
+ const auto ret = output_stream.write(component_id_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->component_type_);
+ const auto ret = output_stream.write(component_type_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->flow_uuid_);
+ const auto ret = output_stream.write(flow_uuid_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->details_);
+ const auto ret = output_stream.write(details_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
// write flow attributes
{
- const auto numAttributes = gsl::narrow<uint32_t>(this->attributes_.size());
+ const auto numAttributes = gsl::narrow<uint32_t>(attributes_.size());
const auto ret = output_stream.write(numAttributes);
if (ret != 4) {
return false;
@@ -167,33 +167,33 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- const auto ret = output_stream.write(this->content_full_path);
+ const auto ret = output_stream.write(content_full_path);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->size_);
+ const auto ret = output_stream.write(size_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = output_stream.write(this->offset_);
+ const auto ret = output_stream.write(offset_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = output_stream.write(this->source_queue_identifier_);
+ const auto ret = output_stream.write(source_queue_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- if (this->event_type_ == ProvenanceEventRecord::FORK || this->event_type_ == ProvenanceEventRecord::CLONE || this->event_type_ == ProvenanceEventRecord::JOIN) {
+ if (event_type_ == ProvenanceEventRecord::FORK || event_type_ == ProvenanceEventRecord::CLONE || event_type_ == ProvenanceEventRecord::JOIN) {
// write UUIDs
{
- const auto parent_uuids_count = gsl::narrow<uint32_t>(this->parent_uuids_.size());
+ const auto parent_uuids_count = gsl::narrow<uint32_t>(parent_uuids_.size());
const auto ret = output_stream.write(parent_uuids_count);
if (ret != 4) {
return false;
@@ -206,7 +206,7 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
}
}
{
- const auto children_uuids_count = gsl::narrow<uint32_t>(this->children_uuids_.size());
+ const auto children_uuids_count = gsl::narrow<uint32_t>(children_uuids_.size());
const auto ret = output_stream.write(children_uuids_count);
if (ret != 4) {
return false;
@@ -218,20 +218,20 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) {
return false;
}
}
- } else if (this->event_type_ == ProvenanceEventRecord::SEND || this->event_type_ == ProvenanceEventRecord::FETCH) {
- const auto ret = output_stream.write(this->transit_uri_);
+ } else if (event_type_ == ProvenanceEventRecord::SEND || event_type_ == ProvenanceEventRecord::FETCH) {
+ const auto ret = output_stream.write(transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
- } else if (this->event_type_ == ProvenanceEventRecord::RECEIVE) {
+ } else if (event_type_ == ProvenanceEventRecord::RECEIVE) {
{
- const auto ret = output_stream.write(this->transit_uri_);
+ const auto ret = output_stream.write(transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = output_stream.write(this->source_system_flow_file_identifier_);
+ const auto ret = output_stream.write(source_system_flow_file_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -300,28 +300,28 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
}
{
- const auto ret = input_stream.read(this->component_id_);
+ const auto ret = input_stream.read(component_id_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->component_type_);
+ const auto ret = input_stream.read(component_type_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->flow_uuid_);
+ const auto ret = input_stream.read(flow_uuid_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->details_);
+ const auto ret = input_stream.read(details_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -351,38 +351,38 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
return false;
}
}
- this->attributes_[key] = value;
+ attributes_[key] = value;
}
{
- const auto ret = input_stream.read(this->content_full_path);
+ const auto ret = input_stream.read(content_full_path);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->size_);
+ const auto ret = input_stream.read(size_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = input_stream.read(this->offset_);
+ const auto ret = input_stream.read(offset_);
if (ret != 8) {
return false;
}
}
{
- const auto ret = input_stream.read(this->source_queue_identifier_);
+ const auto ret = input_stream.read(source_queue_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- if (this->event_type_ == ProvenanceEventRecord::FORK || this->event_type_ == ProvenanceEventRecord::CLONE || this->event_type_ == ProvenanceEventRecord::JOIN) {
+ if (event_type_ == ProvenanceEventRecord::FORK || event_type_ == ProvenanceEventRecord::CLONE || event_type_ == ProvenanceEventRecord::JOIN) {
// read UUIDs
uint32_t number = 0;
{
@@ -400,7 +400,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
return false;
}
}
- this->addParentUuid(parentUUID);
+ addParentUuid(parentUUID);
}
number = 0;
{
@@ -417,24 +417,24 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) {
return false;
}
}
- this->addChildUuid(childUUID);
+ addChildUuid(childUUID);
}
- } else if (this->event_type_ == ProvenanceEventRecord::SEND || this->event_type_ == ProvenanceEventRecord::FETCH) {
+ } else if (event_type_ == ProvenanceEventRecord::SEND || event_type_ == ProvenanceEventRecord::FETCH) {
{
- const auto ret = input_stream.read(this->transit_uri_);
+ const auto ret = input_stream.read(transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
- } else if (this->event_type_ == ProvenanceEventRecord::RECEIVE) {
+ } else if (event_type_ == ProvenanceEventRecord::RECEIVE) {
{
- const auto ret = input_stream.read(this->transit_uri_);
+ const auto ret = input_stream.read(transit_uri_);
if (ret == 0 || io::isError(ret)) {
return false;
}
}
{
- const auto ret = input_stream.read(this->source_system_flow_file_identifier_);
+ const auto ret = input_stream.read(source_system_flow_file_identifier_);
if (ret == 0 || io::isError(ret)) {
return false;
}
@@ -509,7 +509,6 @@ void ProvenanceReporterImpl::clone(const core::FlowFile& parent, const core::Flo
if (event) {
event->addChildFlowFile(child);
- event->addParentFlowFile(parent);
add(event);
}
}
@@ -577,7 +576,7 @@ void ProvenanceReporterImpl::fetch(const core::FlowFile& flow_file, const std::s
}
std::shared_ptr<ProvenanceEventRecord> ProvenanceEventRecord::create() {
- return std::make_shared<ProvenanceEventRecordImpl>();
+ return std::make_shared<ProvenanceEventRecordImpl>(ProvenanceEventType::CLONE, utils::Identifier{}, "");
}
} // namespace org::apache::nifi::minifi::provenance