Skip to content

MINIFICPP-2852 Implement LMDB based flow file repository #2203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lordgamez wants to merge 5 commits into
base: MINIFICPP-2849
Choose a base branch
from
Open

MINIFICPP-2852 Implement LMDB based flow file repository #2203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
81ce624
MINIFICPP-2852 Implement LMDB based flow file repository
lordgamez Jun 17, 2026
2878f27
Review update
lordgamez Aug 13, 2026
c57d649
Mark LmdbFlowFileRepository as experimental in docs
lordgamez Aug 17, 2026
bd87fa7
Review update
lordgamez Aug 25, 2026
0ab4f40
Fix after rebase
lordgamez Aug 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CONFIGURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -620,7 +620,7 @@ Apache MiNiFi C++ uses three repositories similarly to Apache NiFi:

The underlying implementation to use for these repositories can be configured in the minifi.properties file.

The Flow File Repository can be configured with the `nifi.flowfile.repository.class.name` property. If not specified, it uses the `FlowFileRepository` class by default, which stores the flow file metadata in a RocksDB database. Alternatively it can be configured to use a `NoOpRepository` for not keeping any state, flow files are only stored in memory while being transferred between processors.
The Flow File Repository can be configured with the `nifi.flowfile.repository.class.name` property. If not specified, it uses the `FlowFileRepository` class by default, which stores the flow file metadata in a RocksDB database. Alternatively it can be configured to use a `NoOpRepository` for not keeping any state, flow files are only stored in memory while being transferred between processors. Another option is using `LmdbFlowFileRepository` to use LMDB database to store flow file metadata (Note: `LmdbFlowFileRepository` is currently experimental).

# in minifi.properties
nifi.flowfile.repository.class.name=NoOpRepository # VolatileFlowFileRepository can also be used which is an alias for NoOpRepository
Expand Down
175 changes: 20 additions & 155 deletions extensions/lmdb/LmdbContentRepository.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
#include "minifi-cpp/utils/Literals.h"
#include "utils/file/FileUtils.h"

namespace org::apache::nifi::minifi::core::repository {
namespace org::apache::nifi::minifi::extensions::lmdb {

LmdbContentRepository::Session::Session(std::shared_ptr<LmdbContentRepository> repository) : BufferedContentSession(std::move(repository)) {}

Expand All @@ -49,7 +49,7 @@ void LmdbContentRepository::Session::commit() {
throw Exception(REPOSITORY_EXCEPTION, "Couldn't open the underlying resource for write: " + resource_claim->getContentFullPath());
}
const auto size = stream->size();
auto lmdb_out_stream = std::dynamic_pointer_cast<io::LmdbStream>(outStream);
auto lmdb_out_stream = std::dynamic_pointer_cast<LmdbStream>(outStream);
if (lmdb_out_stream == nullptr) {
throw Exception(REPOSITORY_EXCEPTION, "Couldn't cast output stream to LmdbStream for commit: " + resource_claim->getContentFullPath());
}
Expand All @@ -74,11 +74,6 @@ void LmdbContentRepository::Session::commit() {
}

bool LmdbContentRepository::initialize(const std::shared_ptr<minifi::Configure>& configuration) {
if (const int rc = mdb_env_create(&lmdb_env_); rc != MDB_SUCCESS) {
logger_->log_error("Failed to create LMDB environment: {}", mdb_strerror(rc));
return false;
}

// Reserve virtual address space for the DB file (max size it can grow to)
std::expected<uint64_t, std::error_code> max_db_size;
auto max_db_size_str_opt = configuration->get(Configure::nifi_content_repository_lmdb_max_db_size);
Expand All @@ -90,16 +85,6 @@ bool LmdbContentRepository::initialize(const std::shared_ptr<minifi::Configure>&

if (!max_db_size) {
logger_->log_error("Invalid max DB size configuration for LMDB Content Repository: {}", *max_db_size_str_opt);
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}

logger_->log_info("Setting LMDB max DB size to {} bytes", *max_db_size);
if (const auto rc = mdb_env_set_mapsize(lmdb_env_, gsl::narrow<size_t>(*max_db_size)); rc != MDB_SUCCESS) {
logger_->log_error("Failed to set LMDB map size: {}", mdb_strerror(rc));
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}

Expand All @@ -111,152 +96,46 @@ bool LmdbContentRepository::initialize(const std::shared_ptr<minifi::Configure>&
} else {
directory_ = (working_dir / "lmdbcontentrepository").string();
}
logger_->log_info("Using LMDB Content Repository directory '{}'", directory_);

if (std::filesystem::exists(directory_)) {
logger_->log_info("Using existing LMDB Content Repository directory at {}", directory_);
} else {
logger_->log_info("Creating LMDB Content Repository directory at {}", directory_);
if (!std::filesystem::create_directories(directory_)) {
logger_->log_error("Failed to create LMDB Content Repository directory at {}", directory_);
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}
}

if (const int rc = mdb_env_open(lmdb_env_, directory_.c_str(), MDB_NOTLS, 0664)) {
logger_->log_error("Failed to open LMDB environment: {}", mdb_strerror(rc));
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}

MDB_txn* init_txn = nullptr;
if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, 0, &init_txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to begin LMDB transaction during initialize: {}", mdb_strerror(rc));
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}
if (const int rc = mdb_dbi_open(init_txn, nullptr, 0, &lmdb_handle_); rc != MDB_SUCCESS) {
logger_->log_error("Failed to open LMDB database: {}", mdb_strerror(rc));
mdb_txn_abort(init_txn);
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}

if (const int rc = mdb_txn_commit(init_txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to commit LMDB transaction during initialize: {}", mdb_strerror(rc));
mdb_env_close(lmdb_env_);
lmdb_env_ = nullptr;
return false;
}

return true;
return lmdb_wrapper_.initialize(directory_, *max_db_size);
}

void LmdbContentRepository::start() {}
void LmdbContentRepository::stop() {}

std::shared_ptr<ContentSession> LmdbContentRepository::createSession() {
std::shared_ptr<core::ContentSession> LmdbContentRepository::createSession() {
return std::make_shared<Session>(sharedFromThis<LmdbContentRepository>());
}

std::shared_ptr<io::BaseStream> LmdbContentRepository::write(const minifi::ResourceClaim& claim, bool append) {
return std::make_shared<io::LmdbStream>(claim.getContentFullPath(), lmdb_env_, &lmdb_handle_, true, append);
return std::make_shared<LmdbStream>(claim.getContentFullPath(), lmdb_wrapper_, true, append);
}

std::shared_ptr<io::BaseStream> LmdbContentRepository::read(const minifi::ResourceClaim& claim) {
return std::make_shared<io::LmdbStream>(claim.getContentFullPath(), lmdb_env_, &lmdb_handle_, false);
return std::make_shared<LmdbStream>(claim.getContentFullPath(), lmdb_wrapper_, false);
}

bool LmdbContentRepository::exists(const minifi::ResourceClaim& streamId) {
const auto path = streamId.getContentFullPath();
MDB_val key{path.size(), const_cast<char*>(path.data())};
MDB_val value{};

MDB_txn* txn = nullptr;
if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, MDB_RDONLY, &txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to begin LMDB read transaction in exists: {}", mdb_strerror(rc));
return false;
}
auto guard = gsl::finally([txn] { mdb_txn_abort(txn); });

const auto rc = mdb_get(txn, lmdb_handle_, &key, &value);
if (rc != MDB_SUCCESS && rc != MDB_NOTFOUND) {
logger_->log_error("Failed to get value from LMDB database: {}", mdb_strerror(rc));
}
return rc == MDB_SUCCESS;
return lmdb_wrapper_.exists(path);
}

bool LmdbContentRepository::removeKey(const std::string& content_path) {
MDB_val key{content_path.size(), const_cast<char*>(content_path.data())};

MDB_txn* txn = nullptr;
if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, 0, &txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to begin LMDB write transaction in removeKey: {}", mdb_strerror(rc));
return false;
}
int rc = mdb_del(txn, lmdb_handle_, &key, nullptr);

if (rc == MDB_SUCCESS) {
if (const int rc = mdb_txn_commit(txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to commit LMDB transaction during delete: {}", mdb_strerror(rc));
return false;
}
return true;
} else if (rc == MDB_NOTFOUND) {
logger_->log_debug("Key {} not found in LMDB database during delete", content_path);
mdb_txn_abort(txn);
return true;
} else {
logger_->log_error("Failed to delete key '{}' from LMDB database: {}", content_path, mdb_strerror(rc));
mdb_txn_abort(txn);
return false;
}
return lmdb_wrapper_.removeKey(content_path);
}

void LmdbContentRepository::clearOrphans() {
std::vector<std::string> keys_to_be_deleted;

{
MDB_txn* txn = nullptr;
if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, MDB_RDONLY, &txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to begin LMDB read transaction in clearOrphans: {}", mdb_strerror(rc));
return;
}
auto txn_guard = gsl::finally([txn] { mdb_txn_abort(txn); });

MDB_cursor* cursor = nullptr;
if (const int rc = mdb_cursor_open(txn, lmdb_handle_, &cursor); rc != MDB_SUCCESS) {
logger_->log_error("Failed to open LMDB cursor in clearOrphans: {}", mdb_strerror(rc));
return;
}
auto cursor_guard = gsl::finally([cursor] { mdb_cursor_close(cursor); });

MDB_val key{};
MDB_val val{};
int rc = mdb_cursor_get(cursor, &key, &val, MDB_FIRST);

{
std::lock_guard<std::mutex> lock(count_map_mutex_);
while (rc == MDB_SUCCESS) {
std::string key_string = std::string(static_cast<char*>(key.mv_data), key.mv_size);

auto claim_it = count_map_.find(key_string);
if (claim_it == count_map_.end() || claim_it->second == 0) {
logger_->log_debug("Deleting orphan resource {}", key_string);
keys_to_be_deleted.push_back(key_string);
}
rc = mdb_cursor_get(cursor, &key, &val, MDB_NEXT);
std::lock_guard<std::mutex> lock(count_map_mutex_);
lmdb_wrapper_.forEach([this, &keys_to_be_deleted](const MDB_val& key, const MDB_val&) {
std::string key_string = std::string(static_cast<char*>(key.mv_data), key.mv_size);

auto claim_it = count_map_.find(key_string);
if (claim_it == count_map_.end() || claim_it->second == 0) {
logger_->log_debug("Deleting orphan resource {}", key_string);
keys_to_be_deleted.push_back(key_string);
}
}

if (rc != MDB_NOTFOUND) {
logger_->log_error("Failed to iterate over LMDB database: {}", mdb_strerror(rc));
return;
}
});
}

std::vector<std::string> failed_deletions;
Expand All @@ -272,28 +151,14 @@ void LmdbContentRepository::clearOrphans() {
purge_list_.insert(purge_list_.end(), std::make_move_iterator(failed_deletions.begin()), std::make_move_iterator(failed_deletions.end()));
}

MDB_stat LmdbContentRepository::getDbStat() const {
MDB_stat stat{};
MDB_txn* txn = nullptr;
if (const int rc = mdb_txn_begin(lmdb_env_, nullptr, MDB_RDONLY, &txn); rc != MDB_SUCCESS) {
logger_->log_error("Failed to begin LMDB read transaction in getDbStat: {}", mdb_strerror(rc));
return stat;
}
if (const int rc = mdb_stat(txn, lmdb_handle_, &stat); rc != MDB_SUCCESS) {
logger_->log_error("Failed to read LMDB database stats: {}", mdb_strerror(rc));
}
mdb_txn_abort(txn);
return stat;
}

uint64_t LmdbContentRepository::getRepositorySize() const {
return utils::file::path_size(directory_);
}

uint64_t LmdbContentRepository::getRepositoryEntryCount() const {
return getDbStat().ms_entries;
return lmdb_wrapper_.getDbStat().ms_entries;
}

REGISTER_RESOURCE_AS(LmdbContentRepository, InternalResource, ("LmdbContentRepository", "lmdbcontentrepository"));

} // namespace org::apache::nifi::minifi::core::repository
} // namespace org::apache::nifi::minifi::extensions::lmdb
32 changes: 12 additions & 20 deletions extensions/lmdb/LmdbContentRepository.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,40 +28,35 @@
#include "minifi-cpp/core/PropertyDefinition.h"
#include "minifi-cpp/utils/Export.h"
#include "minifi-cpp/utils/Id.h"
#include "lmdb.h"
#include "LmdbWrapper.h"

namespace org::apache::nifi::minifi::core::repository {
namespace org::apache::nifi::minifi::extensions::lmdb {

class LmdbContentRepository : public core::ContentRepositoryImpl {
public:
explicit LmdbContentRepository(std::string_view name = className<LmdbContentRepository>(), const utils::Identifier& uuid = {})
explicit LmdbContentRepository(std::string_view name = core::className<LmdbContentRepository>(), const utils::Identifier& uuid = {})
: core::ContentRepositoryImpl(name, uuid) {}

LmdbContentRepository(const LmdbContentRepository&) = delete;
LmdbContentRepository(LmdbContentRepository&&) = delete;
LmdbContentRepository& operator=(const LmdbContentRepository&) = delete;
LmdbContentRepository& operator=(LmdbContentRepository&&) = delete;

~LmdbContentRepository() noexcept override {
LmdbContentRepository::stop();
if (lmdb_env_) {
mdb_dbi_close(lmdb_env_, lmdb_handle_);
mdb_env_close(lmdb_env_);
}
~LmdbContentRepository() override {
stop();
}

EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 0>{};
EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;

class Session : public BufferedContentSession {
class Session : public core::BufferedContentSession {
public:
explicit Session(std::shared_ptr<LmdbContentRepository> repository);

void commit() override;
};

std::shared_ptr<ContentSession> createSession() override;
std::shared_ptr<core::ContentSession> createSession() override;
bool initialize(const std::shared_ptr<minifi::Configure>& configuration) override;
std::shared_ptr<io::BaseStream> write(const minifi::ResourceClaim& claim, bool append) override;
std::shared_ptr<io::BaseStream> read(const minifi::ResourceClaim& claim) override;
Expand All @@ -72,8 +67,8 @@ class LmdbContentRepository : public core::ContentRepositoryImpl {

void clearOrphans() override;

void start() override;
void stop() override;
void start() override {}
void stop() override {}

uint64_t getRepositorySize() const override;
uint64_t getRepositoryEntryCount() const override;
Expand All @@ -82,11 +77,8 @@ class LmdbContentRepository : public core::ContentRepositoryImpl {
bool removeKey(const std::string& content_path) override;

private:
MDB_stat getDbStat() const;

MDB_env* lmdb_env_{nullptr};
MDB_dbi lmdb_handle_{};
std::shared_ptr<logging::Logger> logger_{logging::LoggerFactory<LmdbContentRepository>::getLogger()};
LmdbWrapper lmdb_wrapper_;
std::shared_ptr<core::logging::Logger> logger_{core::logging::LoggerFactory<LmdbContentRepository>::getLogger()};
};

} // namespace org::apache::nifi::minifi::core::repository
} // namespace org::apache::nifi::minifi::extensions::lmdb
Loading
Loading