You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by lo...@apache.org on 2023/04/28 10:14:15 UTC
[nifi-minifi-cpp] branch main updated (0f6cabc75 -> 72e1c3516)
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 0f6cabc75 MINIFICPP-2037 Build SSLContext from minifi.properties for SSL communication on C2
new 4fb71dee2 MINIFICPP-2040 - Store content claim for flow file deletion
new 72e1c3516 MINIFICPP-2103 JNI extension compile fix (libc++)
The 2 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.
Summary of changes:
extensions/jni/jvm/JniReferenceObjects.h | 13 +--
extensions/rocksdb-repos/FlowFileRepository.cpp | 106 +++++++++++++++---------
extensions/rocksdb-repos/FlowFileRepository.h | 11 ++-
libminifi/include/core/Repository.h | 4 +
4 files changed, 89 insertions(+), 45 deletions(-)
[nifi-minifi-cpp] 02/02: MINIFICPP-2103 JNI extension compile fix (libc++)
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 72e1c35164ab19137fc19396f902536c625584a5
Author: Martin Zink <ma...@apache.org>
AuthorDate: Fri Apr 14 14:34:58 2023 +0200
MINIFICPP-2103 JNI extension compile fix (libc++)
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1561
---
extensions/jni/jvm/JniReferenceObjects.h | 13 ++++++++-----
1 file changed, 8 insertions(+), 5 deletions(-)
diff --git a/extensions/jni/jvm/JniReferenceObjects.h b/extensions/jni/jvm/JniReferenceObjects.h
index 558f68561..028089863 100644
--- a/extensions/jni/jvm/JniReferenceObjects.h
+++ b/extensions/jni/jvm/JniReferenceObjects.h
@@ -104,7 +104,7 @@ class JniByteInputStream {
}
// seek to offset
- auto remaining = gsl::narrow<uint64_t>(size);
+ auto remaining = gsl::narrow<size_t>(size);
int writtenOffset = 0;
int read = 0;
do {
@@ -128,8 +128,11 @@ class JniByteInputStream {
return read;
}
- int64_t read(uint8_t &arr) {
- return stream_->read(arr);
+ int64_t read(uint8_t& value) const {
+ auto read_result = stream_->read(value);
+ if (io::isError(read_result))
+ return -1;
+ return gsl::narrow<int64_t>(read_result);
}
std::shared_ptr<minifi::io::InputStream> stream_;
@@ -153,9 +156,9 @@ class JniInputStream : public core::WeakReference {
}
}
- int64_t read(uint8_t &arr) {
+ int64_t read(uint8_t& value) {
if (!removed_) {
- return jbi_->read(arr);
+ return jbi_->read(value);
}
return -1;
}
[nifi-minifi-cpp] 01/02: MINIFICPP-2040 - Store content claim for flow file deletion
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 4fb71dee2cf2401659df95d8a7f2a226c3ab2568
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Wed Feb 8 07:36:23 2023 +0100
MINIFICPP-2040 - Store content claim for flow file deletion
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1508
---
extensions/rocksdb-repos/FlowFileRepository.cpp | 106 +++++++++++++++---------
extensions/rocksdb-repos/FlowFileRepository.h | 11 ++-
libminifi/include/core/Repository.h | 4 +
3 files changed, 81 insertions(+), 40 deletions(-)
diff --git a/extensions/rocksdb-repos/FlowFileRepository.cpp b/extensions/rocksdb-repos/FlowFileRepository.cpp
index 7568201e3..36999c4e0 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.cpp
+++ b/extensions/rocksdb-repos/FlowFileRepository.cpp
@@ -41,52 +41,71 @@ void FlowFileRepository::flush() {
return;
}
auto batch = opendb->createWriteBatch();
- rocksdb::ReadOptions options;
- std::vector<std::shared_ptr<FlowFile>> purgeList;
+ std::list<ExpiredFlowFileInfo> flow_files;
- std::vector<rocksdb::Slice> keys;
- std::list<std::string> keystrings;
- std::vector<std::string> values;
-
- while (keys_to_delete.size_approx() > 0) {
- std::string key;
- if (keys_to_delete.try_dequeue(key)) {
- keystrings.push_back(std::move(key)); // rocksdb::Slice doesn't copy the string, only grabs ptrs. Hacky, but have to ensure the required lifetime of the strings.
- keys.push_back(keystrings.back());
+ while (keys_to_delete_.size_approx() > 0) {
+ ExpiredFlowFileInfo info;
+ if (keys_to_delete_.try_dequeue(info)) {
+ flow_files.push_back(std::move(info));
}
}
- auto multistatus = opendb->MultiGet(options, keys, &values);
- for (size_t i = 0; i < keys.size() && i < values.size() && i < multistatus.size(); ++i) {
- if (!multistatus[i].ok()) {
- logger_->log_error("Failed to read key from rocksdb: %s! DB is most probably in an inconsistent state!", keys[i].data());
- keystrings.remove(keys[i].data());
- continue;
- }
+ deserializeFlowFilesWithNoContentClaim(opendb.value(), flow_files);
- utils::Identifier containerId;
- auto eventRead = FlowFileRecord::DeSerialize(gsl::make_span(values[i]).as_span<const std::byte>(), content_repo_, containerId);
- if (eventRead) {
- purgeList.push_back(eventRead);
- }
- logger_->log_debug("Issuing batch delete, including %s, Content path %s", eventRead->getUUIDStr(), eventRead->getContentFullPath());
- batch.Delete(keys[i]);
+ for (auto& ff : flow_files) {
+ batch.Delete(ff.key);
+ logger_->log_debug("Issuing batch delete, including %s, Content path %s", ff.key, ff.content ? ff.content->getContentFullPath() : "null");
}
auto operation = [&batch, &opendb]() { return opendb->Write(rocksdb::WriteOptions(), &batch); };
if (!ExecuteWithRetry(operation)) {
- for (const auto& key : keystrings) {
- keys_to_delete.enqueue(key); // Push back the values that we could get but couldn't delete
+ 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 (const auto &ffr : purgeList) {
- auto claim = ffr->getResourceClaim();
- if (claim) claim->decreaseFlowFileRecordOwnedCount();
+ for (auto& ff : flow_files) {
+ if (ff.content) {
+ ff.content->decreaseFlowFileRecordOwnedCount();
+ }
+ }
+ }
+}
+
+void FlowFileRepository::deserializeFlowFilesWithNoContentClaim(minifi::internal::OpenRocksDb& opendb, std::list<ExpiredFlowFileInfo>& flow_files) {
+ std::vector<rocksdb::Slice> 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::string> values;
+ auto multistatus = opendb.MultiGet(rocksdb::ReadOptions{}, keys, &values);
+ gsl_Expects(keys.size() == values.size() && values.size() == multistatus.size());
+
+ for (size_t i = 0; i < keys.size(); ++i) {
+ if (!multistatus[i].ok()) {
+ logger_->log_error("Failed to read key from rocksdb: %s! DB is most probably in an inconsistent state!", keys[i].data());
+ flow_files.erase(key_positions.at(i));
+ continue;
+ }
+
+ utils::Identifier container_id;
+ auto flow_file = FlowFileRecord::DeSerialize(gsl::make_span(values[i]).as_span<const std::byte>(), content_repo_, container_id);
+ if (flow_file) {
+ gsl_Expects(flow_file->getUUIDStr() == key_positions.at(i)->key);
+ key_positions.at(i)->content = flow_file->getResourceClaim();
+ } else {
+ logger_->log_error("Could not deserialize flow file %s", key_positions.at(i)->key);
}
}
}
@@ -149,34 +168,34 @@ void FlowFileRepository::initialize_repository() {
auto it = opendb->NewIterator(rocksdb::ReadOptions());
for (it->SeekToFirst(); it->Valid(); it->Next()) {
- utils::Identifier containerId;
- auto eventRead = FlowFileRecord::DeSerialize(gsl::make_span(it->value()).as_span<const std::byte>(), content_repo_, containerId);
+ utils::Identifier container_id;
+ auto eventRead = FlowFileRecord::DeSerialize(gsl::make_span(it->value()).as_span<const std::byte>(), content_repo_, container_id);
std::string key = it->key().ToString();
if (eventRead) {
// on behalf of the just resurrected persisted instance
auto claim = eventRead->getResourceClaim();
if (claim) claim->increaseFlowFileRecordOwnedCount();
bool found = false;
- auto search = containers_.find(containerId.to_string());
+ auto search = containers_.find(container_id.to_string());
found = (search != containers_.end());
if (!found) {
// for backward compatibility
- search = connection_map_.find(containerId.to_string());
+ search = connection_map_.find(container_id.to_string());
found = (search != connection_map_.end());
}
if (found) {
- logger_->log_debug("Found connection for %s, path %s ", containerId.to_string(), eventRead->getContentFullPath());
+ logger_->log_debug("Found connection for %s, path %s ", container_id.to_string(), eventRead->getContentFullPath());
eventRead->setStoredToRepository(true);
// we found the connection for the persistent flowFile
// even if a processor immediately marks it for deletion, flush only happens after prune_stored_flowfiles
search->second->restore(eventRead);
} else {
- logger_->log_warn("Could not find connection for %s, path %s ", containerId.to_string(), eventRead->getContentFullPath());
- keys_to_delete.enqueue(key);
+ logger_->log_warn("Could not find connection for %s, path %s ", container_id.to_string(), eventRead->getContentFullPath());
+ keys_to_delete_.enqueue({.key = key, .content = eventRead->getResourceClaim()});
}
} else {
// failed to deserialize FlowFile, cannot clear claim
- keys_to_delete.enqueue(key);
+ keys_to_delete_.enqueue({.key = key});
}
}
flush();
@@ -287,7 +306,7 @@ bool FlowFileRepository::MultiPut(const std::vector<std::pair<std::string, std::
}
bool FlowFileRepository::Delete(const std::string& key) {
- keys_to_delete.enqueue(key);
+ keys_to_delete_.enqueue({.key = key});
return true;
}
@@ -340,6 +359,15 @@ std::future<std::vector<std::shared_ptr<core::FlowFile>>> FlowFileRepository::lo
return swap_loader_->load(std::move(flow_files));
}
+bool FlowFileRepository::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;
+}
+
REGISTER_RESOURCE_AS(FlowFileRepository, InternalResource, ("FlowFileRepository", "flowfilerepository"));
} // namespace org::apache::nifi::minifi::core::repository
diff --git a/extensions/rocksdb-repos/FlowFileRepository.h b/extensions/rocksdb-repos/FlowFileRepository.h
index e90081221..68959ce88 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.h
+++ b/extensions/rocksdb-repos/FlowFileRepository.h
@@ -20,6 +20,7 @@
#include <vector>
#include <string>
#include <memory>
+#include <list>
#include "utils/file/FileUtils.h"
#include "rocksdb/db.h"
@@ -61,6 +62,11 @@ constexpr auto FLOWFILE_REPOSITORY_RETRY_INTERVAL_INCREMENTS = std::chrono::mill
class FlowFileRepository : public ThreadedRepository, public SwapManager {
static constexpr std::chrono::milliseconds DEFAULT_COMPACTION_PERIOD = std::chrono::minutes{2};
+ struct ExpiredFlowFileInfo {
+ std::string key;
+ std::shared_ptr<ResourceClaim> content{};
+ };
+
public:
static constexpr const char* ENCRYPTION_KEY_NAME = "nifi.flowfile.repository.encryption.key";
@@ -98,6 +104,7 @@ class FlowFileRepository : public ThreadedRepository, public SwapManager {
bool Put(const std::string& key, const uint8_t *buf, size_t bufLen) override;
bool MultiPut(const std::vector<std::pair<std::string, std::unique_ptr<minifi::io::BufferStream>>>& data) override;
bool Delete(const std::string& key) override;
+ bool Delete(const std::shared_ptr<core::CoreComponent>& item) override;
bool Get(const std::string &key, std::string &value) override;
void loadComponent(const std::shared_ptr<core::ContentRepository> &content_repo) override;
@@ -116,11 +123,13 @@ class FlowFileRepository : public ThreadedRepository, public SwapManager {
void initialize_repository();
+ void deserializeFlowFilesWithNoContentClaim(minifi::internal::OpenRocksDb& opendb, std::list<ExpiredFlowFileInfo>& flow_files);
+
std::thread& getThread() override {
return thread_;
}
- moodycamel::ConcurrentQueue<std::string> keys_to_delete;
+ moodycamel::ConcurrentQueue<ExpiredFlowFileInfo> keys_to_delete_;
std::shared_ptr<core::ContentRepository> content_repo_;
std::unique_ptr<minifi::internal::RocksDatabase> db_;
std::unique_ptr<FlowFileLoader> swap_loader_;
diff --git a/libminifi/include/core/Repository.h b/libminifi/include/core/Repository.h
index 2bcf70e5d..77fdfbec3 100644
--- a/libminifi/include/core/Repository.h
+++ b/libminifi/include/core/Repository.h
@@ -98,6 +98,10 @@ class Repository : public core::CoreComponent {
return true;
}
+ virtual bool Delete(const std::shared_ptr<core::CoreComponent>& item) {
+ return Delete(item->getUUIDStr());
+ }
+
virtual bool Delete(std::vector<std::shared_ptr<core::SerializableComponent>> &storedValues);
void setConnectionMap(std::map<std::string, core::Connectable*> connectionMap) {