You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ab...@apache.org on 2020/07/16 22:30:33 UTC
[nifi-minifi-cpp] branch main updated: MINIFICPP-1272 - Wait for
non-source processors during FlowController shutdown
This is an automated email from the ASF dual-hosted git repository.
aboda pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new f11d398 MINIFICPP-1272 - Wait for non-source processors during FlowController shutdown
f11d398 is described below
commit f11d3986124741322abe7fe93cc2db55707859b8
Author: Adam Debreceni <ad...@protonmail.com>
AuthorDate: Mon Jul 6 16:10:23 2020 +0200
MINIFICPP-1272 - Wait for non-source processors during FlowController shutdown
Signed-off-by: Arpad Boda <ab...@apache.org>
This closes #833
---
libminifi/include/FlowController.h | 2 +
libminifi/include/core/ProcessGroup.h | 4 +-
libminifi/include/core/Property.h | 104 +-----------
libminifi/include/core/TypedValues.h | 75 +++-----
libminifi/include/properties/Configure.h | 1 +
libminifi/include/utils/StringUtils.h | 8 -
libminifi/include/utils/ValueParser.h | 60 +++++++
libminifi/src/Configure.cpp | 1 +
libminifi/src/FlowController.cpp | 32 +++-
libminifi/src/ThreadedSchedulingAgent.cpp | 12 +-
libminifi/src/core/ProcessGroup.cpp | 22 ++-
libminifi/test/flow-tests/CustomProcessors.h | 7 +
libminifi/test/flow-tests/CycleTest.cpp | 23 +--
libminifi/test/flow-tests/FlowBuilder.h | 64 -------
libminifi/test/flow-tests/FlowControllerTests.cpp | 188 +++++++++++++++++----
libminifi/test/flow-tests/LoopTest.cpp | 21 +--
libminifi/test/flow-tests/MultiLoopTest.cpp | 21 +--
libminifi/test/flow-tests/TestControllerWithFlow.h | 67 ++++++++
18 files changed, 397 insertions(+), 315 deletions(-)
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index a2a41c7..86e504c 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -414,6 +414,8 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
std::chrono::steady_clock::time_point last_metrics_capture_;
private:
+ std::chrono::milliseconds shutdown_check_interval_{1000};
+
std::shared_ptr<logging::Logger> logger_;
std::string serial_number_;
diff --git a/libminifi/include/core/ProcessGroup.h b/libminifi/include/core/ProcessGroup.h
index 9944658..370426a 100644
--- a/libminifi/include/core/ProcessGroup.h
+++ b/libminifi/include/core/ProcessGroup.h
@@ -176,7 +176,7 @@ class ProcessGroup {
// Start Processing
void startProcessing(const std::shared_ptr<TimerDrivenSchedulingAgent> timeScheduler, const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler, const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler); // NOLINT
// Stop Processing
- void stopProcessing(const std::shared_ptr<TimerDrivenSchedulingAgent> timeScheduler, const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler, const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler); // NOLINT
+ void stopProcessing(const std::shared_ptr<TimerDrivenSchedulingAgent> timeScheduler, const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler, const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler, const std::function<bool(const std::shared_ptr<Processor>&)>& filter = [] (const std::shared_ptr<Processor>&) {return true;}); // NOLINT
// Whether it is root process group
bool isRootProcessGroup();
// set parent process group
@@ -230,6 +230,8 @@ class ProcessGroup {
void drainConnections();
+ std::size_t getTotalFlowFileCount() const;
+
protected:
void startProcessingProcessors(const std::shared_ptr<TimerDrivenSchedulingAgent> timeScheduler, const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler, const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler); // NOLINT
diff --git a/libminifi/include/core/Property.h b/libminifi/include/core/Property.h
index 01937e7..6e505fb 100644
--- a/libminifi/include/core/Property.h
+++ b/libminifi/include/core/Property.h
@@ -249,112 +249,12 @@ class Property {
// Convert String
static bool StringToTime(std::string input, uint64_t &output, TimeUnit &timeunit) {
- if (input.size() == 0) {
- return false;
- }
-
- const char *cvalue = input.c_str();
- char *pEnd;
- auto ival = std::strtoll(cvalue, &pEnd, 0);
-
- if (pEnd[0] == '\0') {
- return false;
- }
-
- while (*pEnd == ' ') {
- // Skip the space
- pEnd++;
- }
-
- std::string unit(pEnd);
- std::transform(unit.begin(), unit.end(), unit.begin(), ::tolower);
-
- if (unit == "sec" || unit == "s" || unit == "second" || unit == "seconds" || unit == "secs") {
- timeunit = SECOND;
- output = ival;
- return true;
- } else if (unit == "msec" || unit == "ms" || unit == "millisecond" || unit == "milliseconds" || unit == "msecs") {
- timeunit = MILLISECOND;
- output = ival;
- return true;
- } else if (unit == "min" || unit == "m" || unit == "mins" || unit == "minute" || unit == "minutes") {
- timeunit = MINUTE;
- output = ival;
- return true;
- } else if (unit == "ns" || unit == "nano" || unit == "nanos" || unit == "nanoseconds") {
- timeunit = NANOSECOND;
- output = ival;
- return true;
- } else if (unit == "ms" || unit == "milli" || unit == "millis" || unit == "milliseconds") {
- timeunit = MILLISECOND;
- output = ival;
- return true;
- } else if (unit == "h" || unit == "hr" || unit == "hour" || unit == "hrs" || unit == "hours") {
- timeunit = HOUR;
- output = ival;
- return true;
- } else if (unit == "d" || unit == "day" || unit == "days") {
- timeunit = DAY;
- output = ival;
- return true;
- } else {
- return false;
- }
+ return utils::internal::StringToTime(input, output, timeunit);
}
// Convert String
static bool StringToTime(const std::string& input, int64_t &output, TimeUnit &timeunit) {
- if (input.size() == 0) {
- return false;
- }
-
- const char *cvalue = input.c_str();
- char *pEnd;
- auto ival = std::strtoll(cvalue, &pEnd, 0);
-
- if (pEnd[0] == '\0') {
- return false;
- }
-
- while (*pEnd == ' ') {
- // Skip the space
- pEnd++;
- }
-
- std::string unit(pEnd);
- std::transform(unit.begin(), unit.end(), unit.begin(), ::tolower);
-
- if (unit == "sec" || unit == "s" || unit == "second" || unit == "seconds" || unit == "secs") {
- timeunit = SECOND;
- output = ival;
- return true;
- } else if (unit == "msec" || unit == "ms" || unit == "millisecond" || unit == "milliseconds" || unit == "msecs") {
- timeunit = MILLISECOND;
- output = ival;
- return true;
- } else if (unit == "min" || unit == "m" || unit == "mins" || unit == "minute" || unit == "minutes") {
- timeunit = MINUTE;
- output = ival;
- return true;
- } else if (unit == "ns" || unit == "nano" || unit == "nanos" || unit == "nanoseconds") {
- timeunit = NANOSECOND;
- output = ival;
- return true;
- } else if (unit == "ms" || unit == "milli" || unit == "millis" || unit == "milliseconds") {
- timeunit = MILLISECOND;
- output = ival;
- return true;
- } else if (unit == "h" || unit == "hr" || unit == "hour" || unit == "hrs" || unit == "hours") {
- timeunit = HOUR;
- output = ival;
- return true;
- } else if (unit == "d" || unit == "day" || unit == "days") {
- timeunit = DAY;
- output = ival;
- return true;
- } else {
- return false;
- }
+ return utils::internal::StringToTime(input, output, timeunit);
}
static bool StringToDateTime(const std::string& input, int64_t& output) {
diff --git a/libminifi/include/core/TypedValues.h b/libminifi/include/core/TypedValues.h
index c22ed79..29c585f 100644
--- a/libminifi/include/core/TypedValues.h
+++ b/libminifi/include/core/TypedValues.h
@@ -24,6 +24,9 @@
#include "state/Value.h"
#include "utils/StringUtils.h"
+#include "utils/ValueParser.h"
+#include "utils/PropertyErrors.h"
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -49,15 +52,31 @@ class TimePeriodValue : public TransformableValue, public state::response::UInt6
explicit TimePeriodValue(const std::string &timeString)
: state::response::UInt64Value(0) {
TimeUnit units;
- StringToTime(timeString, value, units);
+ if (!StringToTime(timeString, value, units)) {
+ throw utils::internal::ParseException("Couldn't parse TimePeriodValue");
+ }
string_value = timeString;
- ConvertTimeUnitToMS<uint64_t>(value, units, value);
+ if (!ConvertTimeUnitToMS<uint64_t>(value, units, value)) {
+ throw utils::internal::ConversionException("Couldn't convert TimePeriodValue to milliseconds");
+ }
}
explicit TimePeriodValue(uint64_t value)
: state::response::UInt64Value(value) {
}
+ uint64_t getMilliseconds() const {
+ return getValue();
+ }
+
+ static utils::optional<TimePeriodValue> fromString(const std::string& str) {
+ try {
+ return TimePeriodValue(str);
+ } catch (const utils::internal::ValueException&) {
+ return utils::nullopt;
+ }
+ }
+
// Convert TimeUnit to MilliSecond
template<typename T>
static bool ConvertTimeUnitToMS(T input, TimeUnit unit, T &out) {
@@ -85,57 +104,7 @@ class TimePeriodValue : public TransformableValue, public state::response::UInt6
}
static bool StringToTime(std::string input, uint64_t &output, TimeUnit &timeunit) {
- if (input.size() == 0) {
- return false;
- }
-
- const char *cvalue = input.c_str();
- char *pEnd;
- auto ival = std::strtoll(cvalue, &pEnd, 0);
-
- if (pEnd[0] == '\0') {
- return false;
- }
-
- while (*pEnd == ' ') {
- // Skip the space
- pEnd++;
- }
-
- std::string unit(pEnd);
- std::transform(unit.begin(), unit.end(), unit.begin(), ::tolower);
-
- if (unit == "sec" || unit == "s" || unit == "second" || unit == "seconds" || unit == "secs") {
- timeunit = SECOND;
- output = ival;
- return true;
- } else if (unit == "msec" || unit == "ms" || unit == "millisecond" || unit == "milliseconds" || unit == "msecs") {
- timeunit = MILLISECOND;
- output = ival;
- return true;
- } else if (unit == "min" || unit == "m" || unit == "mins" || unit == "minute" || unit == "minutes") {
- timeunit = MINUTE;
- output = ival;
- return true;
- } else if (unit == "ns" || unit == "nano" || unit == "nanos" || unit == "nanoseconds") {
- timeunit = NANOSECOND;
- output = ival;
- return true;
- } else if (unit == "ms" || unit == "milli" || unit == "millis" || unit == "milliseconds") {
- timeunit = MILLISECOND;
- output = ival;
- return true;
- } else if (unit == "h" || unit == "hr" || unit == "hour" || unit == "hrs" || unit == "hours") {
- timeunit = HOUR;
- output = ival;
- return true;
- } else if (unit == "d" || unit == "day" || unit == "days") {
- timeunit = DAY;
- output = ival;
- return true;
- } else {
- return false;
- }
+ return utils::internal::StringToTime(input, output, timeunit);
}
};
diff --git a/libminifi/include/properties/Configure.h b/libminifi/include/properties/Configure.h
index f1b6bba..7a298f2 100644
--- a/libminifi/include/properties/Configure.h
+++ b/libminifi/include/properties/Configure.h
@@ -63,6 +63,7 @@ class Configure : public Properties {
static const char *nifi_administrative_yield_duration;
static const char *nifi_bored_yield_duration;
static const char *nifi_graceful_shutdown_seconds;
+ static const char *nifi_flowcontroller_drain_timeout;
static const char *nifi_log_level;
static const char *nifi_server_name;
static const char *nifi_configuration_class_name;
diff --git a/libminifi/include/utils/StringUtils.h b/libminifi/include/utils/StringUtils.h
index ff675a1..08847a4 100644
--- a/libminifi/include/utils/StringUtils.h
+++ b/libminifi/include/utils/StringUtils.h
@@ -34,14 +34,6 @@
#include "utils/FailurePolicy.h"
#include "utils/GeneralUtils.h"
-enum TimeUnit {
- DAY,
- HOUR,
- MINUTE,
- SECOND,
- MILLISECOND,
- NANOSECOND
-};
#if defined(WIN32) || (__cplusplus >= 201103L && (!defined(__GLIBCXX__) || (__cplusplus >= 201402L) || (defined(_GLIBCXX_RELEASE) && _GLIBCXX_RELEASE > 4)))
#define HAVE_REGEX_CPP 1
#else
diff --git a/libminifi/include/utils/ValueParser.h b/libminifi/include/utils/ValueParser.h
index dfca9bc..00b48bd 100644
--- a/libminifi/include/utils/ValueParser.h
+++ b/libminifi/include/utils/ValueParser.h
@@ -26,6 +26,7 @@
#include <cstdlib>
#include <type_traits>
#include <limits>
+#include <algorithm>
#include "PropertyErrors.h"
@@ -156,6 +157,65 @@ class ValueParser {
std::size_t offset;
};
+template<typename Out>
+bool StringToTime(const std::string& input, Out& output, core::TimeUnit& timeunit) {
+ if (input.size() == 0) {
+ return false;
+ }
+
+ const char* begin = input.c_str();
+ char *end;
+ errno = 0;
+ auto ival = std::strtoll(begin, &end, 0);
+ if (end == begin || errno == ERANGE) {
+ return false;
+ }
+
+ if (end[0] == '\0') {
+ return false;
+ }
+
+ while (*end == ' ') {
+ // Skip the space
+ end++;
+ }
+
+ std::string unit(end);
+ std::transform(unit.begin(), unit.end(), unit.begin(), ::tolower);
+
+ if (unit == "sec" || unit == "s" || unit == "second" || unit == "seconds" || unit == "secs") {
+ timeunit = core::TimeUnit::SECOND;
+ output = ival;
+ return true;
+ } else if (unit == "msec" || unit == "ms" || unit == "millisecond" || unit == "milliseconds" || unit == "msecs") {
+ timeunit = core::TimeUnit::MILLISECOND;
+ output = ival;
+ return true;
+ } else if (unit == "min" || unit == "m" || unit == "mins" || unit == "minute" || unit == "minutes") {
+ timeunit = core::TimeUnit::MINUTE;
+ output = ival;
+ return true;
+ } else if (unit == "ns" || unit == "nano" || unit == "nanos" || unit == "nanoseconds") {
+ timeunit = core::TimeUnit::NANOSECOND;
+ output = ival;
+ return true;
+ } else if (unit == "ms" || unit == "milli" || unit == "millis" || unit == "milliseconds") {
+ timeunit = core::TimeUnit::MILLISECOND;
+ output = ival;
+ return true;
+ } else if (unit == "h" || unit == "hr" || unit == "hour" || unit == "hrs" || unit == "hours") {
+ timeunit = core::TimeUnit::HOUR;
+ output = ival;
+ return true;
+ } else if (unit == "d" || unit == "day" || unit == "days") {
+ timeunit = core::TimeUnit::DAY;
+ output = ival;
+ return true;
+ } else {
+ return false;
+ }
+}
+
} /* namespace internal */
} /* namespace utils */
} /* namespace minifi */
diff --git a/libminifi/src/Configure.cpp b/libminifi/src/Configure.cpp
index 04f585a..467524b 100644
--- a/libminifi/src/Configure.cpp
+++ b/libminifi/src/Configure.cpp
@@ -33,6 +33,7 @@ const char *Configure::nifi_flow_engine_event_driven_time_slice = "nifi.flow.eng
const char *Configure::nifi_administrative_yield_duration = "nifi.administrative.yield.duration";
const char *Configure::nifi_bored_yield_duration = "nifi.bored.yield.duration";
const char *Configure::nifi_graceful_shutdown_seconds = "nifi.flowcontroller.graceful.shutdown.period";
+const char *Configure::nifi_flowcontroller_drain_timeout = "nifi.flowcontroller.drain.timeout";
const char *Configure::nifi_log_level = "nifi.log.level";
const char *Configure::nifi_server_name = "nifi.server.name";
const char *Configure::nifi_configuration_class_name = "nifi.flow.configuration.class.name";
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index 669285b..80a0102 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -201,7 +201,7 @@ bool FlowController::applyConfiguration(const std::string &source, const std::st
std::lock_guard<std::recursive_mutex> flow_lock(mutex_);
stop(true);
- waitUnload(30000);
+ unload();
controller_map_->clear();
auto prevRoot = std::move(this->root_);
this->root_ = std::move(newRoot);
@@ -239,8 +239,36 @@ int16_t FlowController::stop(bool force, uint64_t timeToWait) {
if (running_) {
// immediately indicate that we are not running
logger_->log_info("Stop Flow Controller");
- if (this->root_)
+ if (this->root_) {
+ // stop source processors first
+ this->root_->stopProcessing(timer_scheduler_, event_scheduler_, cron_scheduler_, [] (const std::shared_ptr<core::Processor>& proc) -> bool {
+ return !proc->hasIncomingConnections();
+ });
+ auto shutdown_start = std::chrono::steady_clock::now();
+ // we enable C2 to progressively increase the timeout
+ // in case it sees that waiting for a little longer could
+ // allow the FlowFiles to be processed
+ auto shutdown_timeout = [&]() -> std::chrono::milliseconds {
+ if (timeToWait != 0) {
+ return std::chrono::milliseconds{timeToWait};
+ }
+ utils::optional<core::TimePeriodValue> shutdown_timeout;
+ std::string shutdown_timeout_str;
+ if (configuration_->get(minifi::Configure::nifi_flowcontroller_drain_timeout, shutdown_timeout_str)) {
+ shutdown_timeout = core::TimePeriodValue::fromString(shutdown_timeout_str);
+ }
+ if (shutdown_timeout) {
+ return std::chrono::milliseconds{shutdown_timeout->getMilliseconds()};
+ }
+ return std::chrono::milliseconds{0};
+ };
+ std::size_t count;
+ while ((std::chrono::steady_clock::now() - shutdown_start) < shutdown_timeout() && (count = this->root_->getTotalFlowFileCount()) != 0) {
+ std::this_thread::sleep_for(shutdown_check_interval_);
+ }
+ // shutdown all other processors as well
this->root_->stopProcessing(timer_scheduler_, event_scheduler_, cron_scheduler_);
+ }
// stop after we've attempted to stop the processors.
this->timer_scheduler_->stop();
this->event_scheduler_->stop();
diff --git a/libminifi/src/ThreadedSchedulingAgent.cpp b/libminifi/src/ThreadedSchedulingAgent.cpp
index c096fbb..3790865 100644
--- a/libminifi/src/ThreadedSchedulingAgent.cpp
+++ b/libminifi/src/ThreadedSchedulingAgent.cpp
@@ -36,6 +36,8 @@
#include "core/ProcessSession.h"
#include "core/ProcessSessionFactory.h"
#include "utils/GeneralUtils.h"
+#include "utils/ValueParser.h"
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -49,16 +51,18 @@ void ThreadedSchedulingAgent::schedule(std::shared_ptr<core::Processor> processo
std::string yieldValue;
if (configure_->get(Configure::nifi_administrative_yield_duration, yieldValue)) {
- core::TimeUnit unit;
- if (core::Property::StringToTime(yieldValue, admin_yield_duration_, unit) && core::Property::ConvertTimeUnitToMS(admin_yield_duration_, unit, admin_yield_duration_)) {
+ utils::optional<core::TimePeriodValue> value = core::TimePeriodValue::fromString(yieldValue);
+ if (value) {
+ admin_yield_duration_ = value->getMilliseconds();
logger_->log_debug("nifi_administrative_yield_duration: [%" PRId64 "] ms", admin_yield_duration_);
}
}
bored_yield_duration_ = 0;
if (configure_->get(Configure::nifi_bored_yield_duration, yieldValue)) {
- core::TimeUnit unit;
- if (core::Property::StringToTime(yieldValue, bored_yield_duration_, unit) && core::Property::ConvertTimeUnitToMS(bored_yield_duration_, unit, bored_yield_duration_)) {
+ utils::optional<core::TimePeriodValue> value = core::TimePeriodValue::fromString(yieldValue);
+ if (value) {
+ bored_yield_duration_ = value->getMilliseconds();
logger_->log_debug("nifi_bored_yield_duration: [%" PRId64 "] ms", bored_yield_duration_);
}
}
diff --git a/libminifi/src/core/ProcessGroup.cpp b/libminifi/src/core/ProcessGroup.cpp
index ee88d79..01e403e 100644
--- a/libminifi/src/core/ProcessGroup.cpp
+++ b/libminifi/src/core/ProcessGroup.cpp
@@ -222,7 +222,7 @@ void ProcessGroup::startProcessing(const std::shared_ptr<TimerDrivenSchedulingAg
}
void ProcessGroup::stopProcessing(const std::shared_ptr<TimerDrivenSchedulingAgent> timeScheduler, const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler,
- const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler) {
+ const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler, const std::function<bool(const std::shared_ptr<Processor>&)>& filter) {
std::lock_guard<std::recursive_mutex> lock(mutex_);
if (onScheduleTimer_) {
@@ -234,6 +234,9 @@ void ProcessGroup::stopProcessing(const std::shared_ptr<TimerDrivenSchedulingAge
try {
// Stop all the processor node, input and output ports
for (const auto &processor : processors_) {
+ if (!filter(processor)) {
+ continue;
+ }
logger_->log_debug("Stopping %s", processor->getName());
switch (processor->getSchedulingStrategy()) {
case TIMER_DRIVEN:
@@ -248,9 +251,8 @@ void ProcessGroup::stopProcessing(const std::shared_ptr<TimerDrivenSchedulingAge
}
}
- for (std::set<ProcessGroup *>::iterator it = child_process_groups_.begin(); it != child_process_groups_.end(); ++it) {
- ProcessGroup *processGroup(*it);
- processGroup->stopProcessing(timeScheduler, eventScheduler, cronScheduler);
+ for (ProcessGroup* childGroup : child_process_groups_) {
+ childGroup->stopProcessing(timeScheduler, eventScheduler, cronScheduler, filter);
}
} catch (std::exception &exception) {
logger_->log_debug("Caught Exception %s", exception.what());
@@ -412,6 +414,18 @@ void ProcessGroup::drainConnections() {
}
}
+std::size_t ProcessGroup::getTotalFlowFileCount() const {
+ std::size_t sum = 0;
+ for (auto& conn : connections_) {
+ sum += conn->getQueueSize();
+ }
+
+ for (const ProcessGroup* childGroup : child_process_groups_) {
+ sum += childGroup->getTotalFlowFileCount();
+ }
+ return sum;
+}
+
} /* namespace core */
} /* namespace minifi */
} /* namespace nifi */
diff --git a/libminifi/test/flow-tests/CustomProcessors.h b/libminifi/test/flow-tests/CustomProcessors.h
index eee987e..6b8605b 100644
--- a/libminifi/test/flow-tests/CustomProcessors.h
+++ b/libminifi/test/flow-tests/CustomProcessors.h
@@ -43,6 +43,7 @@ static core::Property BananaProbability = core::PropertyBuilder::createProperty(
class ProcessorWithStatistics {
public:
std::atomic<int> trigger_count{0};
+ std::function<void()> onTriggerCb_;
};
class TestProcessor : public core::Processor, public ProcessorWithStatistics {
@@ -55,6 +56,9 @@ class TestProcessor : public core::Processor, public ProcessorWithStatistics {
}
void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) override {
++trigger_count;
+ if (onTriggerCb_) {
+ onTriggerCb_();
+ }
auto flowFile = session->get();
if (!flowFile) return;
std::random_device rd{};
@@ -92,6 +96,9 @@ class TestFlowFileGenerator : public processors::GenerateFlowFile, public Proces
void onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) override {
++trigger_count;
+ if (onTriggerCb_) {
+ onTriggerCb_();
+ }
GenerateFlowFile::onTrigger(context.get(), session.get());
}
};
diff --git a/libminifi/test/flow-tests/CycleTest.cpp b/libminifi/test/flow-tests/CycleTest.cpp
index 8f76ebd..93aa312 100644
--- a/libminifi/test/flow-tests/CycleTest.cpp
+++ b/libminifi/test/flow-tests/CycleTest.cpp
@@ -18,7 +18,7 @@
#undef NDEBUG
#include "CustomProcessors.h"
-#include "FlowBuilder.h"
+#include "TestControllerWithFlow.h"
// A flow with structure:
//
@@ -97,22 +97,15 @@ Remote Processing Groups:
)";
TEST_CASE("Flow with two long cycle", "[FlowWithCycle]") {
- TestController testController;
+ TestControllerWithFlow testController(flowConfigurationYaml);
+ testController.startFlow();
- LogTestController::getInstance().setTrace<core::Processor>();
- LogTestController::getInstance().setTrace<minifi::Connection>();
- LogTestController::getInstance().setTrace<core::ProcessSession>();
+ auto controller = testController.controller_;
+ auto root = testController.root_;
- char format[] = "/tmp/flow.XXXXXX";
- std::string dir = testController.createTempDirectory(format);
- std::string yamlPath = utils::file::FileUtils::concat_path(dir, "config.yml");
- std::ofstream{yamlPath} << flowConfigurationYaml;
-
- Flow flow = createFlow(yamlPath);
-
- auto procGenerator = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestFlowFileGenerator>(flow.root_->findProcessor("Generator"));
- auto procA = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(flow.root_->findProcessor("A"));
- auto procB = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(flow.root_->findProcessor("B"));
+ auto procGenerator = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestFlowFileGenerator>(root->findProcessor("Generator"));
+ auto procA = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(root->findProcessor("A"));
+ auto procB = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(root->findProcessor("B"));
int tryCount = 0;
while (tryCount++ < 10 && !(procA->trigger_count.load() >= 10 && procB->trigger_count.load() >= 10)) {
diff --git a/libminifi/test/flow-tests/FlowBuilder.h b/libminifi/test/flow-tests/FlowBuilder.h
deleted file mode 100644
index 83ff2bd..0000000
--- a/libminifi/test/flow-tests/FlowBuilder.h
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- *
- * 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.
- */
-#ifndef NIFI_MINIFI_CPP_FLOWCREATOR_H
-#define NIFI_MINIFI_CPP_FLOWCREATOR_H
-
-#include <unordered_map>
-#include <string>
-#include <random>
-#include <YamlConfiguration.h>
-#include "core/Processor.h"
-#include "TestBase.h"
-
-struct Flow{
- Flow(std::shared_ptr<minifi::FlowController>&& controller, std::shared_ptr<core::ProcessGroup>&& root)
- : controller_(std::move(controller)), root_(std::move(root)) {
- controller_->load(root_);
- controller_->start();
- }
- ~Flow() {
- controller_->stop(true);
- controller_->unload();
- }
- std::shared_ptr<minifi::FlowController> controller_;
- std::shared_ptr<core::ProcessGroup> root_;
-};
-
-Flow createFlow(const std::string& yamlPath) {
- std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>();
- std::shared_ptr<core::Repository> prov_repo = std::make_shared<core::repository::VolatileContentRepository>();
- std::shared_ptr<core::Repository> ff_repo = std::make_shared<core::repository::VolatileContentRepository>();
- std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
-
- configuration->set(minifi::Configure::nifi_flow_configuration_file, yamlPath);
-
- std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
- content_repo->initialize(configuration);
-
- std::unique_ptr<core::FlowConfiguration> flow = utils::make_unique<core::YamlConfiguration>(prov_repo, ff_repo, content_repo, stream_factory, configuration, yamlPath);
- std::shared_ptr<core::ProcessGroup> root = flow->getRoot();
-
- std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(
- prov_repo, ff_repo, configuration,
- std::move(flow),
- content_repo, DEFAULT_ROOT_GROUP_NAME, true);
-
- return Flow{std::move(controller), std::move(root)};
-}
-
-#endif //NIFI_MINIFI_CPP_FLOWCREATOR_H
diff --git a/libminifi/test/flow-tests/FlowControllerTests.cpp b/libminifi/test/flow-tests/FlowControllerTests.cpp
index eb444f6..6bd6699 100644
--- a/libminifi/test/flow-tests/FlowControllerTests.cpp
+++ b/libminifi/test/flow-tests/FlowControllerTests.cpp
@@ -32,16 +32,18 @@
#include "../unit/ProvenanceTestHelper.h"
#include "../TestBase.h"
#include "YamlConfiguration.h"
+#include "CustomProcessors.h"
+#include "TestControllerWithFlow.h"
const char* yamlConfig =
-R"(
+ R"(
Flow Controller:
name: MiNiFi Flow
id: 2438e3c8-015a-1000-79ca-83af40ec1990
Processors:
- name: Generator
id: 2438e3c8-015a-1000-79ca-83af40ec1991
- class: org.apache.nifi.processors.standard.GenerateFlowFile
+ class: org.apache.nifi.processors.standard.TestFlowFileGenerator
max concurrent tasks: 1
scheduling strategy: TIMER_DRIVEN
scheduling period: 100 ms
@@ -50,67 +52,53 @@ Processors:
run duration nanos: 0
auto-terminated relationships list:
Properties:
- Batch Size: 10
- - name: LogAttribute
+ Batch Size: 3
+ - name: TestProcessor
id: 2438e3c8-015a-1000-79ca-83af40ec1992
- class: org.apache.nifi.processors.standard.LogAttribute
+ class: org.apache.nifi.processors.standard.TestProcessor
max concurrent tasks: 1
scheduling strategy: TIMER_DRIVEN
- scheduling period: 1000 sec
- penalization period: 30 sec
+ scheduling period: 100 ms
+ penalization period: 3 sec
yield period: 1 sec
run duration nanos: 0
auto-terminated relationships list:
-
+ - apple
+ - banana
Connections:
- name: Gen
id: 2438e3c8-015a-1000-79ca-83af40ec1997
source name: Generator
source id: 2438e3c8-015a-1000-79ca-83af40ec1991
source relationship name: success
- destination name: LogAttribute
+ destination name: TestProcessor
destination id: 2438e3c8-015a-1000-79ca-83af40ec1992
max work queue size: 0
max work queue data size: 1 MB
flowfile expiration: 60 sec
-
Remote Processing Groups:
-
)";
TEST_CASE("Flow shutdown drains connections", "[TestFlow1]") {
- TestController testController;
- char format[] = "/tmp/flowTest.XXXXXX";
- std::string dir = testController.createTempDirectory(format);
+ TestControllerWithFlow testController(yamlConfig);
+ auto controller = testController.controller_;
+ auto root = testController.root_;
- std::string yamlPath = utils::file::FileUtils::concat_path(dir, "config.yml");
- std::ofstream{yamlPath} << yamlConfig;
+ testController.configuration_->set(minifi::Configure::nifi_flowcontroller_drain_timeout, "100 ms");
- std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>();
- std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestRepository>();
- std::shared_ptr<core::Repository> ff_repo = std::make_shared<TestFlowRepository>();
- std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
-
- configuration->set(minifi::Configure::nifi_flow_configuration_file, yamlPath);
-
- REQUIRE(content_repo->initialize(configuration));
- std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
+ auto sinkProc = std::static_pointer_cast<minifi::processors::TestProcessor>(root->findProcessor("TestProcessor"));
+ // prevent execution of the consumer processor
+ sinkProc->yield(10000);
std::map<std::string, std::shared_ptr<minifi::Connection>> connectionMap;
- std::unique_ptr<core::FlowConfiguration> flow = utils::make_unique<core::YamlConfiguration>(prov_repo, ff_repo, content_repo, stream_factory, configuration, yamlPath);
- std::shared_ptr<core::ProcessGroup> root = flow->getRoot();
- std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(
- prov_repo, ff_repo, configuration,
- std::move(flow),
- content_repo, DEFAULT_ROOT_GROUP_NAME, true);
-
root->getConnections(connectionMap);
// adds the single connection to the map both by name and id
REQUIRE(connectionMap.size() == 2);
- controller->load(root);
- controller->start();
+ testController.startFlow();
+
+ // wait for the generator to create some files
std::this_thread::sleep_for(std::chrono::milliseconds{1000});
for (auto& it : connectionMap) {
@@ -119,7 +107,139 @@ TEST_CASE("Flow shutdown drains connections", "[TestFlow1]") {
controller->stop(true);
+ REQUIRE(sinkProc->trigger_count == 0);
+
for (auto& it : connectionMap) {
REQUIRE(it.second->isEmpty());
}
}
+
+TEST_CASE("Flow shutdown waits for a while", "[TestFlow2]") {
+ TestControllerWithFlow testController(yamlConfig);
+ auto controller = testController.controller_;
+ auto root = testController.root_;
+
+ testController.configuration_->set(minifi::Configure::nifi_flowcontroller_drain_timeout, "10 s");
+
+ auto sourceProc = std::static_pointer_cast<minifi::processors::TestFlowFileGenerator>(root->findProcessor("Generator"));
+ auto sinkProc = std::static_pointer_cast<minifi::processors::TestProcessor>(root->findProcessor("TestProcessor"));
+
+ // prevent the initial trigger
+ // in case the source got triggered
+ // and the scheduler triggers the sink
+ // before we could initiate the shutdown
+ sinkProc->yield(100);
+
+ testController.startFlow();
+
+ // wait for the source processor to enqueue its flowFiles
+ int tryCount = 0;
+ while (tryCount++ < 10 && root->getTotalFlowFileCount() != 3) {
+ std::this_thread::sleep_for(std::chrono::milliseconds{20});
+ }
+
+ REQUIRE(root->getTotalFlowFileCount() == 3);
+ REQUIRE(sourceProc->trigger_count.load() == 1);
+ REQUIRE(sinkProc->trigger_count.load() == 0);
+
+ controller->stop(true);
+
+ REQUIRE(sourceProc->trigger_count.load() == 1);
+ REQUIRE(sinkProc->trigger_count.load() == 3);
+}
+
+TEST_CASE("Flow stopped after grace period", "[TestFlow3]") {
+ TestControllerWithFlow testController(yamlConfig);
+ auto controller = testController.controller_;
+ auto root = testController.root_;
+
+ testController.configuration_->set(minifi::Configure::nifi_flowcontroller_drain_timeout, "1000 ms");
+
+ auto sourceProc = std::static_pointer_cast<minifi::processors::TestFlowFileGenerator>(root->findProcessor("Generator"));
+ auto sinkProc = std::static_pointer_cast<minifi::processors::TestProcessor>(root->findProcessor("TestProcessor"));
+
+ // prevent the initial trigger
+ // in case the source got triggered
+ // and the scheduler triggers the sink
+ sinkProc->yield(100);
+
+ sinkProc->onTriggerCb_ = [&]{
+ static std::atomic<bool> first_onTrigger{true};
+ bool isFirst = true;
+ // sleep only on the first trigger
+ if (first_onTrigger.compare_exchange_strong(isFirst, false)) {
+ std::this_thread::sleep_for(std::chrono::milliseconds{1500});
+ }
+ };
+
+ testController.startFlow();
+
+ // wait for the source processor to enqueue its flowFiles
+ int tryCount = 0;
+ while (tryCount++ < 10 && root->getTotalFlowFileCount() != 3) {
+ std::this_thread::sleep_for(std::chrono::milliseconds{20});
+ }
+
+ REQUIRE(root->getTotalFlowFileCount() == 3);
+ REQUIRE(sourceProc->trigger_count.load() == 1);
+ REQUIRE(sinkProc->trigger_count.load() == 0);
+
+ controller->stop(true);
+
+ REQUIRE(sourceProc->trigger_count.load() == 1);
+ REQUIRE(sinkProc->trigger_count.load() == 1);
+}
+
+TEST_CASE("Extend the waiting period during shutdown", "[TestFlow4]") {
+ TestControllerWithFlow testController(yamlConfig);
+ auto controller = testController.controller_;
+ auto root = testController.root_;
+
+ int timeout_ms = 1000;
+
+ testController.configuration_->set(minifi::Configure::nifi_flowcontroller_drain_timeout, std::to_string(timeout_ms) + " ms");
+
+ auto sourceProc = std::static_pointer_cast<minifi::processors::TestFlowFileGenerator>(root->findProcessor("Generator"));
+ auto sinkProc = std::static_pointer_cast<minifi::processors::TestProcessor>(root->findProcessor("TestProcessor"));
+
+ // prevent the initial trigger
+ // in case the source got triggered
+ // and the scheduler triggers the sink
+ sinkProc->yield(100);
+
+ sinkProc->onTriggerCb_ = [&]{
+ static std::atomic<bool> first_onTrigger{true};
+ bool isFirst = true;
+ // sleep only on the first trigger
+ if (first_onTrigger.compare_exchange_strong(isFirst, false)) {
+ std::this_thread::sleep_for(std::chrono::milliseconds{1500});
+ }
+ };
+
+ testController.startFlow();
+
+ // wait for the source processor to enqueue its flowFiles
+ int tryCount = 0;
+ while (tryCount++ < 10 && root->getTotalFlowFileCount() != 3) {
+ std::this_thread::sleep_for(std::chrono::milliseconds{20});
+ }
+
+ REQUIRE(root->getTotalFlowFileCount() == 3);
+ REQUIRE(sourceProc->trigger_count.load() == 1);
+ REQUIRE(sinkProc->trigger_count.load() == 0);
+
+ std::thread shutdownThread([&]{
+ controller->stop(true);
+ });
+
+ while (controller->isRunning()) {
+ std::this_thread::sleep_for(std::chrono::milliseconds{500});
+ timeout_ms += 500;
+ testController.configuration_->set(minifi::Configure::nifi_flowcontroller_drain_timeout, std::to_string(timeout_ms) + " ms");
+ }
+
+ shutdownThread.join();
+
+ REQUIRE(sourceProc->trigger_count.load() == 1);
+ REQUIRE(sinkProc->trigger_count.load() == 3);
+}
diff --git a/libminifi/test/flow-tests/LoopTest.cpp b/libminifi/test/flow-tests/LoopTest.cpp
index c05ddd4..f6e7230 100644
--- a/libminifi/test/flow-tests/LoopTest.cpp
+++ b/libminifi/test/flow-tests/LoopTest.cpp
@@ -18,7 +18,7 @@
#undef NDEBUG
#include "CustomProcessors.h"
-#include "FlowBuilder.h"
+#include "TestControllerWithFlow.h"
// A flow with structure:
// [Generator] ---> [A] ---|
@@ -75,21 +75,14 @@ Remote Processing Groups:
)";
TEST_CASE("Flow with a single loop", "[SingleLoopFlow]") {
- TestController testController;
+ TestControllerWithFlow testController(flowConfigurationYaml);
+ testController.startFlow();
- LogTestController::getInstance().setTrace<core::Processor>();
- LogTestController::getInstance().setTrace<minifi::Connection>();
- LogTestController::getInstance().setTrace<core::ProcessSession>();
+ auto controller = testController.controller_;
+ auto root = testController.root_;
- char format[] = "/tmp/flow.XXXXXX";
- std::string dir = testController.createTempDirectory(format);
- std::string yamlPath = utils::file::FileUtils::concat_path(dir, "config.yml");
- std::ofstream{yamlPath} << flowConfigurationYaml;
-
- Flow flow = createFlow(yamlPath);
-
- auto procGenerator = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestFlowFileGenerator>(flow.root_->findProcessor("Generator"));
- auto procA = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(flow.root_->findProcessor("A"));
+ auto procGenerator = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestFlowFileGenerator>(root->findProcessor("Generator"));
+ auto procA = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(root->findProcessor("A"));
int tryCount = 0;
// wait for the procA to get triggered 15 times
diff --git a/libminifi/test/flow-tests/MultiLoopTest.cpp b/libminifi/test/flow-tests/MultiLoopTest.cpp
index 6bf68b7..2049d51 100644
--- a/libminifi/test/flow-tests/MultiLoopTest.cpp
+++ b/libminifi/test/flow-tests/MultiLoopTest.cpp
@@ -18,7 +18,7 @@
#undef NDEBUG
#include "CustomProcessors.h"
-#include "FlowBuilder.h"
+#include "TestControllerWithFlow.h"
// A flow with structure:
// v-----|
@@ -84,21 +84,14 @@ Remote Processing Groups:
)";
TEST_CASE("Flow with two loops", "[MultiLoopFlow]") {
- TestController testController;
+ TestControllerWithFlow testController(flowConfigurationYaml);
+ testController.startFlow();
- LogTestController::getInstance().setTrace<core::Processor>();
- LogTestController::getInstance().setTrace<minifi::Connection>();
- LogTestController::getInstance().setTrace<core::ProcessSession>();
+ auto controller = testController.controller_;
+ auto root = testController.root_;
- char format[] = "/tmp/flow.XXXXXX";
- std::string dir = testController.createTempDirectory(format);
- std::string yamlPath = utils::file::FileUtils::concat_path(dir, "config.yml");
- std::ofstream{yamlPath} << flowConfigurationYaml;
-
- Flow flow = createFlow(yamlPath);
-
- auto procGenerator = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestFlowFileGenerator>(flow.root_->findProcessor("Generator"));
- auto procA = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(flow.root_->findProcessor("A"));
+ auto procGenerator = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestFlowFileGenerator>(root->findProcessor("Generator"));
+ auto procA = std::static_pointer_cast<org::apache::nifi::minifi::processors::TestProcessor>(root->findProcessor("A"));
int tryCount = 0;
while (tryCount++ < 10 && !(procA->trigger_count.load() > 15)) {
diff --git a/libminifi/test/flow-tests/TestControllerWithFlow.h b/libminifi/test/flow-tests/TestControllerWithFlow.h
new file mode 100644
index 0000000..aee610d
--- /dev/null
+++ b/libminifi/test/flow-tests/TestControllerWithFlow.h
@@ -0,0 +1,67 @@
+/**
+ *
+ * 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.
+ */
+
+#ifndef NIFI_MINIFI_CPP_TESTCONTROLLERWITHFLOW_H
+#define NIFI_MINIFI_CPP_TESTCONTROLLERWITHFLOW_H
+
+class TestControllerWithFlow: public TestController{
+ public:
+ TestControllerWithFlow(const char* yamlConfigContent) {
+ LogTestController::getInstance().setTrace<processors::TestProcessor>();
+ LogTestController::getInstance().setTrace<processors::TestFlowFileGenerator>();
+ LogTestController::getInstance().setTrace<minifi::Connection>();
+
+ char format[] = "/tmp/flowTest.XXXXXX";
+ std::string dir = createTempDirectory(format);
+
+ std::string yamlPath = utils::file::FileUtils::concat_path(dir, "config.yml");
+ std::ofstream{yamlPath} << yamlConfigContent;
+
+ configuration_ = std::make_shared<minifi::Configure>();
+ std::shared_ptr<core::Repository> prov_repo = std::make_shared<TestRepository>();
+ std::shared_ptr<core::Repository> ff_repo = std::make_shared<TestFlowRepository>();
+ std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
+
+ configuration_->set(minifi::Configure::nifi_flow_configuration_file, yamlPath);
+
+ REQUIRE(content_repo->initialize(configuration_));
+ std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration_);
+
+ std::unique_ptr<core::FlowConfiguration> flow = utils::make_unique<core::YamlConfiguration>(prov_repo, ff_repo, content_repo, stream_factory, configuration_, yamlPath);
+ root_ = flow->getRoot();
+ controller_ = std::make_shared<minifi::FlowController>(
+ prov_repo, ff_repo, configuration_,
+ std::move(flow),
+ content_repo, DEFAULT_ROOT_GROUP_NAME, true);
+ }
+
+ void startFlow() {
+ controller_->load(root_);
+ controller_->start();
+ }
+
+ ~TestControllerWithFlow() {
+ LogTestController::getInstance().reset();
+ }
+
+ std::shared_ptr<minifi::Configure> configuration_;
+ std::shared_ptr<minifi::FlowController> controller_;
+ std::shared_ptr<core::ProcessGroup> root_;
+};
+
+#endif //NIFI_MINIFI_CPP_TESTCONTROLLERWITHFLOW_H