Add process group id to processor

lordgamez · Dec 10, 2024 · 93203b0 · 93203b0
1 parent 43dae9b
commit 93203b0
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 1 deletion.
diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h
@@ -165,6 +165,14 @@ class Processor : public Connectable, public ConfigurableComponent, public state
     active_tasks_ = 0;
   }
 
+  std::string getProcessGroupUUIDStr() const {
+    return process_group_uuid_;
+  }
+
+  void setProcessGroupUUIDStr(const std::string &uuid) {
+    process_group_uuid_ = uuid;
+  }
+
   void yield() override;
 
   void yield(std::chrono::steady_clock::duration delta_time);
@@ -256,6 +264,8 @@ class Processor : public Connectable, public ConfigurableComponent, public state
 
   // an outgoing connection allows us to reach these nodes
   std::unordered_map<Connection*, std::unordered_set<Processor*>> reachable_processors_;
+
+  std::string process_group_uuid_;
 };
 
 }  // namespace core

diff --git a/libminifi/src/core/ProcessGroup.cpp b/libminifi/src/core/ProcessGroup.cpp
@@ -86,6 +86,7 @@ std::tuple<Processor*, bool> ProcessGroup::addProcessor(std::unique_ptr<Processo
   gsl_Expects(processor);
   const auto name = processor->getName();
   std::lock_guard<std::recursive_mutex> lock(mutex_);
+  processor->setProcessGroupUUIDStr(getUUIDStr());
   const auto [iter, inserted] = processors_.insert(std::move(processor));
   if (inserted) {
     logger_->log_debug("Add processor {} into process group {}", name, name_);

diff --git a/libminifi/src/core/state/nodes/FlowInformation.cpp b/libminifi/src/core/state/nodes/FlowInformation.cpp
@@ -89,7 +89,7 @@ std::vector<SerializedResponseNode> FlowInformation::serialize() {
         .collapsible = false,
         .children = {
           {.name = "id", .value = std::string{processor->getUUIDStr()}},
-          // {.name = "groupId", .value = std::string{processor->getUUIDStr()}},
+          {.name = "groupId", .value = processor->getProcessGroupUUIDStr()},
           {.name = "bytesRead", .value = metrics->bytes_read.load()},
           {.name = "bytesWritten", .value = metrics->bytes_written.load()},
           {.name = "flowFilesIn", .value = metrics->incoming_flow_files.load()},