MINIFICPP-2845 - Monotonic provenance event identifiers, iterations support, refactor by adamdebreceni · Pull Request #2209 · apache/nifi-minifi-cpp

adamdebreceni · 2026-07-06T09:14:22Z

Depends on #2195

Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.

In order to streamline the review of the contribution we ask you to ensure the following steps have been taken:

For all changes:

Is there a JIRA ticket associated with this PR? Is it referenced in the commit message?
Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically main)?
Is your initial contribution a single, squashed commit?

For code changes:

If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
If applicable, have you updated the LICENSE file?
If applicable, have you updated the NOTICE file?

For documentation related changes:

Have you ensured that format looks appropriate for the output in which it is rendered?

Note:

Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.

szaszm · 2026-07-06T09:29:44Z

 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,


I believe the "repo" naming of identifiers that have to be provenance repo was a mistake. I'm glad the type makes it clearer now, but I'd change the parameter and member names too. provenance_repo or prov_repo seem like good options to me.

szaszm · 2026-07-06T09:51:49Z

+  Identifier& operator++();
+  Identifier operator++(int);
+


why do we need these UUID increments? I believe the Identified class is not meant to represent integer auto-increment keys.

currently the RocksDbProvenanceRepository uses the event_uuid for key, we could either bring back the retired SequentialIdGenerator, we could replace the increment operators with a next() method to not cause confusion, or we could use an uint64_t (event ordinal) as key, which one do you prefer?

…efactor

Co-authored-by: Márton Szász <szaszm@apache.org>

Copilot

Pull request overview

Introduces a dedicated provenance repository API with monotonic event identifiers and cursor-based reporting.

Changes:

Adds provenance-specific append and iteration APIs.
Implements persistent monotonic identifiers in RocksDB.
Refactors reporting, runtime wiring, and tests to use the new API.

Reviewed changes

Copilot reviewed 43 out of 43 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
`minifi-api/include/minifi-cpp/provenance/ProvenanceRepository.h`	Adds provenance repository interface.
`minifi-api/include/minifi-cpp/provenance/Provenance.h`	Adds event ordinals and vector-backed reporting.
`minifi-api/include/minifi-cpp/core/Repository.h`	Removes generic provenance operations.
`minifi-api/include/minifi-cpp/core/ProcessContext.h`	Returns typed provenance repositories.
`minifi-api/common/include/minifi-cpp/utils/Id.h`	Declares identifier increment operators.
`minifi_main/MiNiFiMain.cpp`	Casts configured repository to the provenance API.
`libminifi/test/unit/SchedulingAgentTests.cpp`	Updates test repository typing.
`libminifi/test/persistence-tests/PersistenceTests.cpp`	Updates persistence fixtures.
`libminifi/test/libtest/unit/TestControllerWithFlow.cpp`	Updates test flow setup.
`libminifi/test/libtest/unit/TestBase.h`	Updates test plan provenance types.
`libminifi/test/libtest/unit/TestBase.cpp`	Implements updated test plan API.
`libminifi/test/libtest/unit/ProvenanceTestHelper.h`	Implements test provenance repository operations.
`libminifi/test/libtest/integration/IntegrationBase.cpp`	Updates integration repository setup.
`libminifi/test/integration/ControllerServiceIntegrationTests.cpp`	Updates repository inference.
`libminifi/test/integration/C2PauseResumeTest.cpp`	Updates repository inference.
`libminifi/test/flow-tests/SessionTests.cpp`	Casts the no-op repository.
`libminifi/src/provenance/Provenance.cpp`	Uses batched provenance appends.
`libminifi/src/FlowController.cpp`	Accepts the typed repository.
`libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp`	Adds cursor-based reporting and ordinals.
`libminifi/src/core/ProcessContextImpl.cpp`	Stores the typed repository.
`libminifi/src/core/flow/StructuredConfiguration.cpp`	Assigns reporting-task UUIDs.
`libminifi/include/TimerDrivenSchedulingAgent.h`	Updates repository type.
`libminifi/include/ThreadedSchedulingAgent.h`	Updates repository type.
`libminifi/include/SchedulingAgent.h`	Stores the typed repository.
`libminifi/include/provenance/Provenance.h`	Implements ordinals and vector-backed events.
`libminifi/include/FlowController.h`	Exposes the typed repository.
`libminifi/include/EventDrivenSchedulingAgent.h`	Updates repository type.
`libminifi/include/CronDrivenSchedulingAgent.h`	Updates repository type.
`libminifi/include/core/repository/VolatileProvenanceRepository.h`	Implements volatile event appends.
`libminifi/include/core/repository/NoOpThreadedRepository.h`	Implements no-op provenance operations.
`libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h`	Uses typed event records.
`libminifi/include/core/ProcessContextImpl.h`	Updates context repository types.
`extensions/standard-processors/tests/unit/ProcessorTests.cpp`	Adapts reporting tests to `getEvents`.
`extensions/standard-processors/tests/integration/VerifyInvokeHTTP.h`	Updates test repository inference.
`extensions/rocksdb-repos/tests/RepoTests.cpp`	Uses renamed RocksDB repository.
`extensions/rocksdb-repos/tests/ProvenanceTests.cpp`	Tests the append API.
`extensions/rocksdb-repos/tests/DBProvenanceRepositoryTests.cpp`	Adds cursor and monotonicity tests.
`extensions/rocksdb-repos/RocksDbProvenanceRepository.h`	Defines the renamed RocksDB implementation.
`extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp`	Implements counters, cursors, and iteration.
`extensions/rocksdb-repos/ProvenanceRepository.cpp`	Removes the previous implementation.
`core-framework/src/core/Repository.cpp`	Removes generic element storage.
`core-framework/include/core/Repository.h`	Removes generic provenance defaults.
`core-framework/common/src/utils/Id.cpp`	Implements identifier incrementing.

Suppressed comments (4)

extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp:131

A persisted cursor is expected to be parsed so the reporting task can fall back when it is invalid, but every string is currently accepted. A malformed/high-valued state such as zzzz will seek past every UUID and make provenance reporting return no records forever; validate it as an Identifier and return nullptr on failure.

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>("");
}

extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp:174

The iterator status is never checked after traversal, so a RocksDB read error is returned as a successful partial result and the reporting task can persist its cursor. Check it->status() before updating the cursor and return std::unexpected on failure, consistent with RocksDbStateStorage iteration.

  if (event_cursor) {
    event_cursor->event_id_ = last_event_id;
  }
  return records;

libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp:224

A false return from transmitPayload only yields and then reaches this cursor update, marking records as consumed even though they were not transmitted. Return from the trigger on the false path, just as the exception path does, so the same cursor batch is retried.

  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)) {

extensions/rocksdb-repos/RocksDbProvenanceRepository.cpp:169

Failed deserialization is silently skipped while last_event_id still advances. Once a later valid record is returned, the reporting task persists that advanced cursor, permanently losing the malformed record from the reporting stream. Return an error instead of advancing past an event that could not be decoded.

    if (eventRead->deserialize(stream)) {
      records.push_back(eventRead);
      if (--max_size == 0) {
        break;
      }
    }

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

+    MultiPut(data);
+
+    return {};


+    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();


+    if (auto event_ordinal = record->getEventOrdinal()) {
+      recordJson.AddMember("eventOrdinal", event_ordinal.value(), alloc);
+    }


lordgamez · 2026-08-14T08:26:54Z

+  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)


nitpick: the parameters are not indented correctly.

lordgamez · 2026-08-14T08:37:07Z

+    if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value))
+      max_partition_millis_ = *max_partition;


nitpick: we should put braces around this line to be consistent with other one line conditional statements

lordgamez · 2026-08-14T08:58:40Z

-  testRepository->storeElement(record1);
+  testRepository->appendEvents({record1});
+
+  utils::Identifier eventId = record1->getEventId();


This variable is redeclared and does not compile

lordgamez · 2026-08-14T08:58:46Z

-  testRepository->storeElement(record1);
+  testRepository->appendEvents({record1});
+
+  utils::Identifier eventId = record1->getEventId();


This variable is redeclared and does not compile

lordgamez · 2026-08-14T09:02:25Z

+    int64_t max_buffer_size = 16 << 20;
+    cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_));


Can we use 16_MB from Literals.h here and omit the max_buffer_size variable?

lordgamez · 2026-08-14T09:08:46Z

+  if (cursor_str.has_value()) {
+    return std::make_unique<EventCursor>(cursor_str.value());
+  }
+  return std::make_unique<EventCursor>("");


If the std::nullopt behavior is the same as the empty string, do we need to make this optional or can we just use empty string when we want a cursor from the start? Or is it possible in the future that an empty string and nullopt cases would differ?

lordgamez · 2026-08-14T09:14:21Z

+}
+
+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;


This type is kind of long and repeated quite a lot in the repository domian. It should be named with an alias because its variable is usually also just called "data" which is not really descriptive.

lordgamez · 2026-08-14T09:20:42Z

+  }
+  for (auto& event : events) {
+    data.emplace_back(event->getUUIDStr(), std::make_unique<io::BufferStream>());
+    event->serialize(*data.back().second);


Should we also handle if the event serialization fails?

lordgamez · 2026-08-14T09:40:35Z

+  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);
+  }


I would extract getting the cursor to a separate function in anonymous namespace.

lordgamez · 2026-08-14T10:16:47Z

  try {
    std::map<std::string, std::string> attributes;
    if (!protocol_->transmitPayload(context, jsonStr, attributes)) {
      context.yield();


With the added code should we return here after the yield?

lordgamez · 2026-08-14T11:38:58Z

+  return std::unexpected{"Failed to append provenance events"};
+}
+
+REGISTER_RESOURCE_AS(RocksDbProvenanceRepository, InternalResource, ("RocksDbProvenanceRepository", "ProvenanceRepository", "provenancerepository"));


With this registration the repository does not work when I set the provenance repo name in minifi.properties to "RocksDbProvenanceRepository", only with "ProvenanceRepository", I think the "RocksDbProvenanceRepository" as an option should be added.

lordgamez · 2026-08-14T11:40:03Z

+    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);


This error occurs on first startup when the id is not yet stored, that should not be displayed as an error in that case, maybe just as a debug log.

adamdebreceni mentioned this pull request Jul 6, 2026

MINIFICPP-2858 - SQLite provenance repository backend #2210

Draft

8 tasks

szaszm reviewed Jul 6, 2026

View reviewed changes

adamdebreceni force-pushed the MINIFICPP-2845 branch from a5b3ee2 to af3f98a Compare July 23, 2026 13:34

adamdebreceni changed the base branch from main to MINIFICPP-2840 July 23, 2026 13:34

adamdebreceni marked this pull request as ready for review August 3, 2026 10:48

adamdebreceni closed this Aug 5, 2026

adamdebreceni reopened this Aug 5, 2026

adamdebreceni force-pushed the MINIFICPP-2840 branch from 7797653 to 82815c6 Compare August 5, 2026 15:36

adamdebreceni and others added 3 commits August 5, 2026 17:40

MINIFICPP-2845 - Monotonic provenance event ids, iteration support, r…

a0bc7ca

…efactor

MINIFICPP-2845 - Remove extra file

cd32c74

Update core-framework/common/src/utils/Id.cpp

f1b2dcd

Co-authored-by: Márton Szász <szaszm@apache.org>

adamdebreceni force-pushed the MINIFICPP-2845 branch from af3f98a to f1b2dcd Compare August 5, 2026 15:41

martinzink requested a balanced review from Copilot August 12, 2026 13:32

Copilot started reviewing on behalf of martinzink August 12, 2026 13:33 View session

Copilot AI reviewed Aug 12, 2026

View reviewed changes

lordgamez reviewed Aug 14, 2026

View reviewed changes

		if (auto max_partition = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value))
		max_partition_millis_ = *max_partition;

		int64_t max_buffer_size = 16 << 20;
		cf_opts.write_buffer_size = gsl::narrow<size_t>(std::min(max_buffer_size, max_partition_bytes_));

Uh oh!

Conversation

adamdebreceni commented Jul 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

For all changes:

For code changes:

For documentation related changes:

Note:

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adamdebreceni Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

adamdebreceni commented Jul 6, 2026 •

edited

Loading

adamdebreceni Jul 23, 2026 •

edited

Loading