lordgamez commented on code in PR #2203: URL: https://github.com/apache/nifi-minifi-cpp/pull/2203#discussion_r3853935091
########## extensions/lmdb/tests/LmdbFlowFileRepositoryTests.cpp: ########## @@ -0,0 +1,470 @@ +/** + * + * 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 "LmdbFlowFileRepository.h" +#include "LmdbContentRepository.h" +#include "Connection.h" +#include "FlowFileRecord.h" +#include "ResourceClaim.h" +#include "io/BufferStream.h" +#include "properties/Configure.h" +#include "utils/Id.h" +#include "unit/Catch.h" +#include "unit/TestBase.h" + +namespace org::apache::nifi::minifi::test { + +namespace { +std::shared_ptr<minifi::FlowFileRecord> createFlowFileWithContent(core::ContentRepository& content_repo, std::string_view content) { + auto flow_file = std::make_shared<minifi::FlowFileRecordImpl>(); + const auto content_session = content_repo.createSession(); + const auto claim = content_session->create(); + const auto stream = content_session->write(claim); + stream->write(std::as_bytes(std::span(content))); + flow_file->setResourceClaim(claim); + flow_file->setSize(stream->size()); + flow_file->setOffset(0); + stream->close(); + content_session->commit(); + return flow_file; +} +} // namespace + +class LmdbFlowFileRepositoryTests : TestController { + public: + LmdbFlowFileRepositoryTests() { + db_path_ = createTempDirectory(); + auto configuration = std::make_shared<minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_flowfile_repository_directory_default, db_path_.string()); + REQUIRE(flow_file_repo_->initialize(configuration)); + content_db_path_ = createTempDirectory(); + configuration->set(minifi::Configure::nifi_dbcontent_repository_directory_default, content_db_path_.string()); + REQUIRE(content_repo_->initialize(configuration)); + flow_file_repo_->loadComponent(content_repo_); + } + + protected: + std::filesystem::path db_path_; + std::filesystem::path content_db_path_; + std::shared_ptr<extensions::lmdb::LmdbFlowFileRepository> flow_file_repo_ = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + std::shared_ptr<extensions::lmdb::LmdbContentRepository> content_repo_ = std::make_shared<extensions::lmdb::LmdbContentRepository>(); +}; + +TEST_CASE("Initialize LmdbFlowFileRepository", "[lmdb]") { + TestController controller; + + std::filesystem::path db_path; + SECTION("initialize succeeds when the directory does not exist") { + db_path = controller.createTempDirectory() / "does_not_exist_yet"; + REQUIRE_FALSE(std::filesystem::exists(db_path)); + } + + SECTION("initialize succeeds when the directory already exists") { + db_path = controller.createTempDirectory(); + REQUIRE(std::filesystem::exists(db_path)); + } + + auto configuration = std::make_shared<minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_flowfile_repository_directory_default, db_path.string()); + + auto flow_file_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE(flow_file_repo->initialize(configuration)); + REQUIRE(std::filesystem::exists(db_path)); +} + +TEST_CASE("initialize honors a valid max db size and persists data", "[lmdb]") { + TestController controller; + LogTestController::getInstance().setDebug<extensions::lmdb::LmdbWrapper>(); + const auto db_path = controller.createTempDirectory(); + auto configuration = std::make_shared<minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_flowfile_repository_directory_default, db_path.string()); + configuration->set(minifi::Configure::nifi_flowfile_repository_lmdb_max_db_size, "32 MB"); + + auto flow_file_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE(flow_file_repo->initialize(configuration)); + REQUIRE(LogTestController::getInstance().contains("Setting LMDB max DB size to 33554432 bytes")); +} + +TEST_CASE("initialize throws on an invalid max db size", "[lmdb]") { + TestController controller; + const auto db_path = controller.createTempDirectory(); + auto configuration = std::make_shared<minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_flowfile_repository_directory_default, db_path.string()); + configuration->set(minifi::Configure::nifi_flowfile_repository_lmdb_max_db_size, "not-a-size"); + + auto flow_file_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE_THROWS(flow_file_repo->initialize(configuration)); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Put on empty repository increases entry count", "[lmdb]") { + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 0); + static constexpr std::string_view payload = "hello flowfile"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 1); + REQUIRE(flow_file_repo_->getRepositorySize() > 0); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Put with same key overwrites and keeps entry count at 1", "[lmdb]") { + static constexpr std::string_view first = "first"; + static constexpr std::string_view second = "second value, longer than first"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(first.data()), first.size())); + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(second.data()), second.size())); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 1); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Put of multiple distinct keys", "[lmdb]") { + static constexpr std::string_view payload = "data"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + REQUIRE(flow_file_repo_->Put("key2", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + REQUIRE(flow_file_repo_->Put("key3", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 3); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Put of empty value succeeds", "[lmdb]") { + REQUIRE(flow_file_repo_->Put("key1", nullptr, 0)); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 1); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Get on empty repository returns false", "[lmdb]") { + std::string value = "untouched"; + REQUIRE_FALSE(flow_file_repo_->Get("missing", value)); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Put then Get round-trips the value", "[lmdb]") { + static constexpr std::string_view payload = "hello flowfile"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + std::string value; + REQUIRE(flow_file_repo_->Get("key1", value)); + REQUIRE(value == payload); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Get returns false for nonexistent key after other Puts", "[lmdb]") { + static constexpr std::string_view payload = "data"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + std::string value; + REQUIRE_FALSE(flow_file_repo_->Get("key2", value)); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Get reflects the latest Put for a key", "[lmdb]") { + static constexpr std::string_view first = "first"; + static constexpr std::string_view second = "second value"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(first.data()), first.size())); + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(second.data()), second.size())); + std::string value; + REQUIRE(flow_file_repo_->Get("key1", value)); + REQUIRE(value == second); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "MultiPut with empty vector succeeds and adds nothing", "[lmdb]") { + REQUIRE(flow_file_repo_->MultiPut({})); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 0); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "MultiPut writes all entries", "[lmdb]") { + std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>> data; + data.emplace_back("key1", std::make_unique<minifi::io::BufferStream>(std::string{"value-one"})); + data.emplace_back("key2", std::make_unique<minifi::io::BufferStream>(std::string{"value-two"})); + data.emplace_back("key3", std::make_unique<minifi::io::BufferStream>(std::string{"value-three"})); + REQUIRE(flow_file_repo_->MultiPut(data)); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 3); + std::string value; + REQUIRE(flow_file_repo_->Get("key1", value)); + REQUIRE(value == "value-one"); + REQUIRE(flow_file_repo_->Get("key2", value)); + REQUIRE(value == "value-two"); + REQUIRE(flow_file_repo_->Get("key3", value)); + REQUIRE(value == "value-three"); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "MultiPut overwrites existing keys", "[lmdb]") { + static constexpr std::string_view old_value = "old"; + REQUIRE(flow_file_repo_->Put("key1", reinterpret_cast<const uint8_t*>(old_value.data()), old_value.size())); + + std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>> data; + data.emplace_back("key1", std::make_unique<minifi::io::BufferStream>(std::string{"new-one"})); + data.emplace_back("key2", std::make_unique<minifi::io::BufferStream>(std::string{"new-two"})); + REQUIRE(flow_file_repo_->MultiPut(data)); + + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 2); + std::string value; + REQUIRE(flow_file_repo_->Get("key1", value)); + REQUIRE(value == "new-one"); + REQUIRE(flow_file_repo_->Get("key2", value)); + REQUIRE(value == "new-two"); +} + +TEST_CASE("MultiPut entries persist across LmdbFlowFileRepository re-open", "[lmdb]") { + TestController controller; + auto db_path = controller.createTempDirectory(); + auto configuration = std::make_shared<minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_flowfile_repository_directory_default, db_path.string()); + + { + auto flow_file_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE(flow_file_repo->initialize(configuration)); + std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>> data; + data.emplace_back("key1", std::make_unique<minifi::io::BufferStream>(std::string{"persisted-one"})); + data.emplace_back("key2", std::make_unique<minifi::io::BufferStream>(std::string{"persisted-two"})); + REQUIRE(flow_file_repo->MultiPut(data)); + } + + auto reopened_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE(reopened_repo->initialize(configuration)); + REQUIRE(reopened_repo->getRepositoryEntryCount() == 2); + std::string value; + REQUIRE(reopened_repo->Get("key1", value)); + REQUIRE(value == "persisted-one"); + REQUIRE(reopened_repo->Get("key2", value)); + REQUIRE(value == "persisted-two"); +} + +TEST_CASE("Put persists across LmdbFlowFileRepository re-open", "[lmdb]") { + TestController controller; + auto db_path = controller.createTempDirectory(); + auto configuration = std::make_shared<minifi::ConfigureImpl>(); + configuration->set(minifi::Configure::nifi_flowfile_repository_directory_default, db_path.string()); + + static constexpr std::string_view payload = "persisted flowfile"; + { + auto flow_file_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE(flow_file_repo->initialize(configuration)); + REQUIRE(flow_file_repo->Put("key1", reinterpret_cast<const uint8_t*>(payload.data()), payload.size())); + REQUIRE(flow_file_repo->getRepositoryEntryCount() == 1); + } + + auto reopened_repo = std::make_shared<extensions::lmdb::LmdbFlowFileRepository>(); + REQUIRE(reopened_repo->initialize(configuration)); + REQUIRE(reopened_repo->getRepositoryEntryCount() == 1); +} + +TEST_CASE_METHOD(LmdbFlowFileRepositoryTests, "Deleting keys is done in batches after flush", "[lmdb]") { + std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>> data; + data.emplace_back("key1", std::make_unique<minifi::io::BufferStream>(std::string{"value-one"})); + data.emplace_back("key2", std::make_unique<minifi::io::BufferStream>(std::string{"value-two"})); + data.emplace_back("key3", std::make_unique<minifi::io::BufferStream>(std::string{"value-three"})); + REQUIRE(flow_file_repo_->MultiPut(data)); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 3); + REQUIRE(flow_file_repo_->Delete("key1")); + REQUIRE(flow_file_repo_->Delete("key2")); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 3); + flow_file_repo_->flush(); + REQUIRE(flow_file_repo_->getRepositoryEntryCount() == 1); + std::string value; + REQUIRE_FALSE(flow_file_repo_->Get("key1", value)); + REQUIRE_FALSE(flow_file_repo_->Get("key1", value)); Review Comment: Good catch, fixed in https://github.com/apache/nifi-minifi-cpp/pull/2203/commits/37dc0d2ab2abf7f0d0c2213be5b76cec50c28f75 ########## extensions/lmdb/LmdbWrapper.cpp: ########## @@ -0,0 +1,269 @@ +/** + * + * 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 "LmdbWrapper.h" + +#include <filesystem> + +#include "minifi-cpp/utils/gsl.h" + +namespace org::apache::nifi::minifi::extensions::lmdb { + +bool LmdbWrapper::initialize(const std::string& directory, size_t max_db_size) { + if (const auto rc = mdb_env_create(&lmdb_env_); rc != MDB_SUCCESS) { + logger_->log_error("Failed to create LMDB environment: {}", mdb_strerror(rc)); + 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_, 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; + } + + if (std::filesystem::exists(directory)) { + logger_->log_info("Using existing LMDB Repository directory at {}", directory); + } else { + logger_->log_info("Creating LMDB Repository directory at {}", directory); + if (!std::filesystem::create_directories(directory)) { + logger_->log_error("Failed to create LMDB Repository directory at {}", directory); + mdb_env_close(lmdb_env_); + lmdb_env_ = nullptr; + return false; + } + } + + if (const auto rc = mdb_env_open(lmdb_env_, directory.c_str(), MDB_NOTLS, 0664)) { Review Comment: Updated in https://github.com/apache/nifi-minifi-cpp/pull/2203/commits/37dc0d2ab2abf7f0d0c2213be5b76cec50c28f75 ########## extensions/lmdb/LmdbFlowFileRepository.cpp: ########## @@ -0,0 +1,251 @@ +/** + * 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 "LmdbFlowFileRepository.h" +#include "core/Resource.h" +#include "minifi-cpp/FlowFileRecord.h" + +using namespace std::literals::chrono_literals; + +namespace org::apache::nifi::minifi::extensions::lmdb { + +namespace { +bool getRepositoryCheckHealth(const Configure& configure) { + std::string check_health_str; + configure.get(Configure::nifi_flow_file_repository_check_health, check_health_str); + return utils::string::toBool(check_health_str).value_or(true); +} +} // namespace + +bool LmdbFlowFileRepository::initialize(const std::shared_ptr<Configure> &configure) { + std::string value; + + if (configure->get(Configure::nifi_flowfile_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + check_flowfile_content_size_ = getRepositoryCheckHealth(*configure); + logger_->log_debug("NiFi LMDB FlowFile Repository Directory {}", directory_); Review Comment: Updated logging in https://github.com/apache/nifi-minifi-cpp/pull/2203/commits/37dc0d2ab2abf7f0d0c2213be5b76cec50c28f75 ########## extensions/lmdb/LmdbFlowFileRepository.cpp: ########## @@ -0,0 +1,251 @@ +/** + * 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 "LmdbFlowFileRepository.h" +#include "core/Resource.h" +#include "minifi-cpp/FlowFileRecord.h" + +using namespace std::literals::chrono_literals; + +namespace org::apache::nifi::minifi::extensions::lmdb { + +namespace { +bool getRepositoryCheckHealth(const Configure& configure) { + std::string check_health_str; + configure.get(Configure::nifi_flow_file_repository_check_health, check_health_str); + return utils::string::toBool(check_health_str).value_or(true); +} +} // namespace + +bool LmdbFlowFileRepository::initialize(const std::shared_ptr<Configure> &configure) { + std::string value; + + if (configure->get(Configure::nifi_flowfile_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + check_flowfile_content_size_ = getRepositoryCheckHealth(*configure); + logger_->log_debug("NiFi LMDB FlowFile Repository Directory {}", directory_); + + // Reserve virtual address space for the DB file (max size it can grow to) + const auto max_db_size = configure->get(Configure::nifi_flowfile_repository_lmdb_max_db_size) | utils::andThen([](auto max_db_size_str) -> std::optional<uint64_t> { + if (max_db_size_str.empty()) { return std::nullopt; } + return parsing::parseDataSize(max_db_size_str) | utils::orThrow(fmt::format("{} was set to invalid value: '{}'", Configure::nifi_flowfile_repository_lmdb_max_db_size, max_db_size_str)); + }) | utils::orElse([] { + return std::make_optional<uint64_t>(MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE); + }); + + if (!max_db_size) { + logger_->log_error("Invalid max DB size configuration for LMDB FlowFile Repository"); + return false; + } Review Comment: Updated in https://github.com/apache/nifi-minifi-cpp/pull/2203/commits/37dc0d2ab2abf7f0d0c2213be5b76cec50c28f75 ########## extensions/lmdb/LmdbFlowFileRepository.cpp: ########## @@ -0,0 +1,251 @@ +/** + * 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 "LmdbFlowFileRepository.h" +#include "core/Resource.h" +#include "minifi-cpp/FlowFileRecord.h" + +using namespace std::literals::chrono_literals; + +namespace org::apache::nifi::minifi::extensions::lmdb { + +namespace { +bool getRepositoryCheckHealth(const Configure& configure) { + std::string check_health_str; + configure.get(Configure::nifi_flow_file_repository_check_health, check_health_str); + return utils::string::toBool(check_health_str).value_or(true); +} +} // namespace + +bool LmdbFlowFileRepository::initialize(const std::shared_ptr<Configure> &configure) { + std::string value; + + if (configure->get(Configure::nifi_flowfile_repository_directory_default, value) && !value.empty()) { + directory_ = value; + } + check_flowfile_content_size_ = getRepositoryCheckHealth(*configure); + logger_->log_debug("NiFi LMDB FlowFile Repository Directory {}", directory_); + + // Reserve virtual address space for the DB file (max size it can grow to) + const auto max_db_size = configure->get(Configure::nifi_flowfile_repository_lmdb_max_db_size) | utils::andThen([](auto max_db_size_str) -> std::optional<uint64_t> { + if (max_db_size_str.empty()) { return std::nullopt; } + return parsing::parseDataSize(max_db_size_str) | utils::orThrow(fmt::format("{} was set to invalid value: '{}'", Configure::nifi_flowfile_repository_lmdb_max_db_size, max_db_size_str)); + }) | utils::orElse([] { + return std::make_optional<uint64_t>(MAX_FLOWFILE_REPOSITORY_STORAGE_SIZE); + }); + + if (!max_db_size) { + logger_->log_error("Invalid max DB size configuration for LMDB FlowFile Repository"); + return false; + } + + logger_->log_info("Using LMDB FlowFile Repository directory '{}'", directory_); + return lmdb_wrapper_.initialize(directory_, *max_db_size); +} + +bool LmdbFlowFileRepository::Delete(const std::string& key) { + keys_to_delete_.enqueue({.key = key}); + return true; +} + +bool LmdbFlowFileRepository::Delete(const std::shared_ptr<core::CoreComponent>& item) { + if (auto ff = std::dynamic_pointer_cast<core::FlowFile>(item)) { + keys_to_delete_.enqueue({.key = item->getUUIDStr(), .content = ff->getResourceClaim()}); + } else { + keys_to_delete_.enqueue({.key = item->getUUIDStr()}); + } + return true; +} + +bool LmdbFlowFileRepository::Put(const std::string& key, const uint8_t* buf, size_t bufLen) { + if (buf == nullptr) { + return bufLen == 0 && lmdb_wrapper_.putValue(key, {}); + } + return lmdb_wrapper_.putValue(key, std::string(reinterpret_cast<const char*>(buf), bufLen)); +} + +bool LmdbFlowFileRepository::MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) { + return lmdb_wrapper_.putValues(data); +} + +bool LmdbFlowFileRepository::Get(const std::string& key, std::string& value) { + auto result = lmdb_wrapper_.getValue(key); + if (result) { + value = std::move(*result); + return true; + } + return false; +} + +uint64_t LmdbFlowFileRepository::getRepositorySize() const { + const auto stat = lmdb_wrapper_.getDbStat(); + return stat.ms_psize * (stat.ms_branch_pages + stat.ms_leaf_pages + stat.ms_overflow_pages); +} + +uint64_t LmdbFlowFileRepository::getRepositoryEntryCount() const { + return lmdb_wrapper_.getDbStat().ms_entries; +} + +void LmdbFlowFileRepository::flush() { + std::list<ExpiredFlowFileInfo> flow_files; + + while (keys_to_delete_.size_approx() > 0) { + ExpiredFlowFileInfo info; + if (keys_to_delete_.try_dequeue(info)) { + flow_files.push_back(std::move(info)); + } + } + + deserializeFlowFilesWithNoContentClaim(flow_files); + + std::vector<std::string> flow_file_keys; + for (auto& ff : flow_files) { + flow_file_keys.push_back(ff.key); + logger_->log_debug("Issuing batch delete, including {}, Content path {}", ff.key, ff.content ? ff.content->getContentFullPath() : "null"); + } + + if (!lmdb_wrapper_.removeKeys(flow_file_keys)) { + for (auto&& ff : flow_files) { + keys_to_delete_.enqueue(std::move(ff)); + } + return; // Stop here - don't delete from content repo while we have records in FF repo + } + + if (content_repo_) { + for (auto& ff : flow_files) { + if (ff.content) { + ff.content->decreaseFlowFileRecordOwnedCount(); + } + } + } +} + +void LmdbFlowFileRepository::deserializeFlowFilesWithNoContentClaim(std::list<ExpiredFlowFileInfo>& flow_files) { + std::vector<std::string> keys; + std::vector<std::list<ExpiredFlowFileInfo>::iterator> key_positions; + for (auto it = flow_files.begin(); it != flow_files.end(); ++it) { + if (!it->content) { + keys.push_back(it->key); + key_positions.push_back(it); + } + } + if (keys.empty()) { + return; + } + std::vector<std::optional<std::string>> values; + values.reserve(keys.size()); + for (const auto& key : keys) { + values.push_back(lmdb_wrapper_.getValue(key)); + } + + gsl_Expects(keys.size() == values.size()); + + for (size_t i = 0; i < keys.size(); ++i) { + if (!values[i]) { + logger_->log_error("Failed to read key from LMDB: {}! DB is most probably in an inconsistent state!", keys[i].data()); Review Comment: Fixed in https://github.com/apache/nifi-minifi-cpp/pull/2203/commits/37dc0d2ab2abf7f0d0c2213be5b76cec50c28f75 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
