libminifi/test/unit/FlowFileQueueSwapTests.cpp - nifi-minifi-cpp - Git at Google

 /**
  * 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 <span>
 #include <string>
 #include <vector>
 #include <utility>
 #include <memory>

 #include "Connection.h"
 #include "unit/TestBase.h"
 #include "unit/TestUtils.h"
 #include "unit/Catch.h"
 #include "unit/ProvenanceTestHelper.h"
 #include "utils/gsl.h"

 namespace org::apache::nifi::minifi::test {

 using Timepoint = std::chrono::time_point<std::chrono::steady_clock>;

 enum EventKind {
   Store, Load
 };

 struct SwapEvent {
   EventKind kind;
   std::vector<minifi::SwappedFlowFile> flow_files;

   void verifyTimes(std::initializer_list<unsigned> seconds) {
     REQUIRE(flow_files.size() == seconds.size());
     size_t idx = 0;
     for (auto& second : seconds) {
       REQUIRE(flow_files[idx].to_be_processed_after == Timepoint{std::chrono::seconds{second}});
       ++idx;
     }
   }
 };

 class SwappingFlowFileTestRepo : public TestFlowRepository, public minifi::SwapManager {
  public:
   void store(std::vector<std::shared_ptr<core::FlowFile>> flow_files) override {
     std::vector<minifi::SwappedFlowFile> ids;
     for (const auto& ff : flow_files) {
       ids.push_back(minifi::SwappedFlowFile{ff->getUUID(), ff->getPenaltyExpiration()});
       minifi::io::BufferStream output;
       std::dynamic_pointer_cast<minifi::FlowFileRecord>(ff)->Serialize(output);
       Put(ff->getUUIDStr().c_str(), reinterpret_cast<const uint8_t*>(output.getBuffer().data()), output.size());
     }
     swap_events_.push_back({Store, ids});
   }

   std::future<std::vector<std::shared_ptr<core::FlowFile>>> load(std::vector<minifi::SwappedFlowFile> flow_files) override {
     swap_events_.push_back({Load, flow_files});
     LoadTask load_task;
     auto future = load_task.promise.get_future();
     load_task.result.reserve(flow_files.size());
     for (const auto& ff_id : flow_files) {
       std::string value;
       Get(ff_id.id.to_string().c_str(), value);
       minifi::utils::Identifier container_id;
       auto ff = minifi::FlowFileRecord::DeSerialize(gsl::make_span(value).as_span<const std::byte>(), content_repo_, container_id);
       ff->setPenaltyExpiration(ff_id.to_be_processed_after);
       load_task.result.push_back(std::move(ff));
     }
     load_tasks_.push_back(std::move(load_task));
     return future;
   }

   struct LoadTask {
     std::promise<std::vector<std::shared_ptr<core::FlowFile>>> promise;
     std::vector<std::shared_ptr<core::FlowFile>> result;

     void complete() {
       promise.set_value(result);
     }
   };

   std::vector<LoadTask> load_tasks_;
   std::vector<SwapEvent> swap_events_;
 };

 using FlowFilePtr = std::shared_ptr<core::FlowFile>;
 using FlowFilePtrVec = std::vector<FlowFilePtr>;

 struct FlowFileComparator {
   bool operator()(const FlowFilePtr& left, const FlowFilePtr& right) const {
     return left->getPenaltyExpiration() < right->getPenaltyExpiration();
   }
 };

 struct VerifiedQueue {
   void push(FlowFilePtr ff) {
     size();
     impl.push(ff);
     ref_.insert(std::lower_bound(ref_.begin(), ref_.end(), ff, FlowFileComparator{}), ff);
     size();
   }

   FlowFilePtr poll() {
     size();
     FlowFilePtr ff = impl.pop();
     REQUIRE(!ref_.empty());
     // the order when flow files have the same penalty is not fixed
     REQUIRE(ff->getPenaltyExpiration() == ref_.front()->getPenaltyExpiration());
     ref_.erase(ref_.begin());
     size();
     return ff;
   }

   void verify(std::initializer_list<unsigned> live, std::optional<std::initializer_list<unsigned>> inter, std::initializer_list<unsigned> swapped) const {
     // check live ffs
     auto live_copy = utils::FlowFileQueueTestAccessor::get_queue_(impl);
     REQUIRE(live_copy.size() == live.size());
     for (auto sec : live) {
       auto min = live_copy.popMin();
       REQUIRE(min->getPenaltyExpiration() == Timepoint{std::chrono::seconds{sec}});
     }

     // check inter ffs
     if (!inter) {
       REQUIRE_FALSE(utils::FlowFileQueueTestAccessor::get_load_task_(impl).has_value());
     } else {
       auto& intermediate = utils::FlowFileQueueTestAccessor::get_load_task_(impl)->intermediate_items;
       REQUIRE(intermediate.size() == inter->size());
       size_t idx = 0;
       for (auto sec : inter.value()) {
         REQUIRE(intermediate[idx]->getPenaltyExpiration() == Timepoint{std::chrono::seconds{sec}});
         ++idx;
       }
     }

     // check swapped ffs
     auto swapped_copy = utils::FlowFileQueueTestAccessor::get_swapped_flow_files_(impl);
     REQUIRE(swapped_copy.size() == swapped.size());
     for (auto sec : swapped) {
       auto min = swapped_copy.popMin();
       REQUIRE(min.to_be_processed_after == Timepoint{std::chrono::seconds{sec}});
     }
   }

   bool isWorkAvailable() const {
     return impl.isWorkAvailable();
   }

   size_t size() const {
     size_t result = impl.size();
     REQUIRE(result == ref_.size());
     return result;
   }

   explicit VerifiedQueue(std::shared_ptr<minifi::SwapManager> swap_manager)
     : impl(std::move(swap_manager)) {}

   minifi::utils::FlowFileQueue impl;
   FlowFilePtrVec ref_;
 };

 class SwapTestController : public TestController {
  public:
   SwapTestController() {
     content_repo_ = std::make_shared<core::repository::VolatileContentRepository>();
     flow_repo_ = std::make_shared<SwappingFlowFileTestRepo>();
     flow_repo_->loadComponent(content_repo_);
     clock_ = std::make_shared<minifi::test::utils::ManualClock>();
     minifi::utils::timeutils::setClock(clock_);
     queue_ = std::make_shared<VerifiedQueue>(std::static_pointer_cast<minifi::SwapManager>(flow_repo_));
   }

   void setLimits(size_t min_size, size_t target_size, size_t max_size) {
     queue_->impl.setMinSize(min_size);
     queue_->impl.setTargetSize(target_size);
     queue_->impl.setMaxSize(max_size);
   }

   struct SwapEventPattern {
     EventKind kind;
     std::initializer_list<unsigned > seconds;
   };

   void verifySwapEvents(std::vector<SwapEventPattern> events) {
     REQUIRE(flow_repo_->swap_events_.size() == events.size());
     size_t idx = 0;
     for (auto& pattern : events) {
       REQUIRE(pattern.kind == flow_repo_->swap_events_[idx].kind);
       flow_repo_->swap_events_[idx].verifyTimes(pattern.seconds);
     }
   }

   void clearSwapEvents() {
     flow_repo_->swap_events_.clear();
   }

   void verifyQueue(std::initializer_list<unsigned> live, std::optional<std::initializer_list<unsigned>> inter, std::initializer_list<unsigned> swapped) {
     queue_->verify(live, inter, swapped);
   }

   void pushAll(std::initializer_list<unsigned> seconds) {
     for (auto sec : seconds) {
       auto ff = std::static_pointer_cast<core::FlowFile>(std::make_shared<minifi::FlowFileRecordImpl>());
       ff->setPenaltyExpiration(Timepoint{std::chrono::seconds{sec}});
       queue_->push(std::move(ff));
     }
   }

   void popAll(std::initializer_list<unsigned> seconds, bool check_is_work_available = false) {
     for (auto sec : seconds) {
       if (check_is_work_available) {
         REQUIRE(queue_->isWorkAvailable());
       }
       auto ff = queue_->poll();
       REQUIRE(ff->getPenaltyExpiration() == Timepoint{std::chrono::seconds{sec}});
     }
   }

   std::shared_ptr<SwappingFlowFileTestRepo> flow_repo_;
   std::shared_ptr<core::repository::VolatileContentRepository> content_repo_;
   std::shared_ptr<VerifiedQueue> queue_;
   std::shared_ptr<minifi::test::utils::ManualClock> clock_;
 };

 TEST_CASE("Setting swap threshold sets underlying queue limits", "[SwapTest1]") {
   const size_t target_size = 4;
   const size_t min_size = target_size / 2;
   const size_t max_size = target_size * 3 / 2;

   minifi::ConnectionImpl conn(nullptr, nullptr, "");
   conn.setSwapThreshold(target_size);
   REQUIRE(utils::FlowFileQueueTestAccessor::get_min_size_(utils::ConnectionTestAccessor::get_queue_(conn)) == min_size);
   REQUIRE(utils::FlowFileQueueTestAccessor::get_target_size_(utils::ConnectionTestAccessor::get_queue_(conn)) == target_size);
   REQUIRE(utils::FlowFileQueueTestAccessor::get_max_size_(utils::ConnectionTestAccessor::get_queue_(conn)) == max_size);
 }

 TEST_CASE_METHOD(SwapTestController, "Default constructed FlowFileQueue won't swap", "[SwapTest2]") {
   for (unsigned i = 0; i < 100; ++i) {
     pushAll({i});
   }

   REQUIRE(queue_->impl.size() == 100);

   clock_->advance(std::chrono::seconds{200});
   for (size_t i = 0; i < 100; ++i) {
     queue_->poll();
   }

   REQUIRE(queue_->impl.empty());
   verifySwapEvents({});
 }

 TEST_CASE_METHOD(SwapTestController, "Up to max no swap-out is triggered", "[SwapTest3]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40});

   REQUIRE_FALSE(queue_->isWorkAvailable());
   verifySwapEvents({});
 }

 TEST_CASE_METHOD(SwapTestController, "Pushing beyond max triggers a swap-out", "[SwapTest4]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40});

   pushAll({28});
   // size goes from 7 to 4, 3 largest must have been swapped out
   verifySwapEvents({{Store, {60, 50, 40}}});
   verifyQueue({10, 20, 28, 30}, {}, {40, 50, 60});
 }

 TEST_CASE_METHOD(SwapTestController, "Popping until min size does not trigger swap-in", "[SwapTest5]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40, 28});
   clearSwapEvents();
   clock_->advance(std::chrono::seconds{35});
   REQUIRE(queue_->isWorkAvailable());
   popAll({10, 20});
   verifyQueue({28, 30}, {}, {40, 50, 60});
   verifySwapEvents({});
 }

 TEST_CASE_METHOD(SwapTestController, "Popping beyond min size triggers swap-in", "[SwapTest6]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40, 28});
   clearSwapEvents();
   clock_->advance(std::chrono::seconds{35});
   popAll({10, 20, 28});

   // trying to swap-in all three swapped flow files
   verifyQueue({30}, {{}}, {});
   verifySwapEvents({{Load, {40, 50, 60}}});
 }

 TEST_CASE_METHOD(SwapTestController, "Pushing while a swap-in is pending", "[SwapTest7]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40, 28});
   clock_->advance(std::chrono::seconds{35});
   popAll({10, 20, 28});
   verifyQueue({30}, {{}}, {});
   clearSwapEvents();

   SECTION("Pushing into the pending swap-in range") {
     pushAll({45});
     verifyQueue({30}, {{45}}, {});
     verifySwapEvents({});
   }

   SECTION("Pushing before the pending swap-in range") {
     pushAll({35});
     verifyQueue({30, 35}, {{}}, {});
     verifySwapEvents({});
   }

   SECTION("Pushing after the pending swap-in range") {
     pushAll({65});
     verifyQueue({30}, {{}}, {65});
     verifySwapEvents({{Store, {65}}});
   }
 }

 TEST_CASE_METHOD(SwapTestController, "isWorkAvailable depends on the swap-in task", "[SwapTest8]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40, 28});
   popAll({10, 20, 28});
   verifyQueue({30}, {{}}, {});

   REQUIRE_FALSE(queue_->isWorkAvailable());

   clock_->advance(std::chrono::seconds{35});

   REQUIRE(queue_->isWorkAvailable());
   popAll({30});
   REQUIRE_FALSE(queue_->isWorkAvailable());

   SECTION("Load is completed but the minimum of those files is still not viable") {
     flow_repo_->load_tasks_[0].complete();
     REQUIRE_FALSE(queue_->isWorkAvailable());
   }

   SECTION("The minimum of the load task is viable but not yet completed") {
     clock_->advance(std::chrono::seconds{35});
     REQUIRE_FALSE(queue_->isWorkAvailable());
     // completing the task renders it viable
     flow_repo_->load_tasks_[0].complete();
     REQUIRE(queue_->isWorkAvailable());
   }
 }

 TEST_CASE_METHOD(SwapTestController, "Polling from load task", "[SwapTest8]") {
   setLimits(2, 4, 6);
   pushAll({50, 20, 30, 60, 10, 40, 28});
   popAll({10, 20, 28});
   pushAll({45});
   verifyQueue({30}, {{45}}, {});

   flow_repo_->load_tasks_[0].complete();

   clock_->advance(std::chrono::seconds{100});

   popAll({30, 40, 45, 50, 60}, true);
   verifyQueue({}, {}, {});
 }

 TEST_CASE_METHOD(SwapTestController, "Popping below min checks if the pending load is finished", "[SwapTest8]") {
   setLimits(6, 8, 10);
   pushAll({10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120});
   verifyQueue({10, 20, 30, 40, 50, 60, 70, 80}, {}, {90, 100, 110, 120});
   clock_->advance(std::chrono::seconds{200});
   clearSwapEvents();
   popAll({10, 20, 30});
   verifySwapEvents({{Load, {90, 100, 110}}});
   verifyQueue({40, 50, 60, 70, 80}, {{}}, {120});
   clearSwapEvents();

   popAll({40, 50});
   verifyQueue({60, 70, 80}, {{}}, {120});
   flow_repo_->load_tasks_[0].complete();
   popAll({60});
   // even though the live queue is not empty we check if
   // the load_task is finished and initiate a load if need be
   verifySwapEvents({{Load, {120}}});
   verifyQueue({70, 80, 90, 100, 110}, {{}}, {});
 }

 }  // namespace org::apache::nifi::minifi::test
	/**
	* 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 <span>
	#include <string>
	#include <vector>
	#include <utility>
	#include <memory>

	#include "Connection.h"
	#include "unit/TestBase.h"
	#include "unit/TestUtils.h"
	#include "unit/Catch.h"
	#include "unit/ProvenanceTestHelper.h"
	#include "utils/gsl.h"

	namespace org::apache::nifi::minifi::test {

	using Timepoint = std::chrono::time_point<std::chrono::steady_clock>;

	enum EventKind {
	Store, Load
	};

	struct SwapEvent {
	EventKind kind;
	std::vector<minifi::SwappedFlowFile> flow_files;

	void verifyTimes(std::initializer_list<unsigned> seconds) {
	REQUIRE(flow_files.size() == seconds.size());
	size_t idx = 0;
	for (auto& second : seconds) {
	REQUIRE(flow_files[idx].to_be_processed_after == Timepoint{std::chrono::seconds{second}});
	++idx;
	}
	}
	};

	class SwappingFlowFileTestRepo : public TestFlowRepository, public minifi::SwapManager {
	public:
	void store(std::vector<std::shared_ptr<core::FlowFile>> flow_files) override {
	std::vector<minifi::SwappedFlowFile> ids;
	for (const auto& ff : flow_files) {
	ids.push_back(minifi::SwappedFlowFile{ff->getUUID(), ff->getPenaltyExpiration()});
	minifi::io::BufferStream output;
	std::dynamic_pointer_cast<minifi::FlowFileRecord>(ff)->Serialize(output);
	Put(ff->getUUIDStr().c_str(), reinterpret_cast<const uint8_t*>(output.getBuffer().data()), output.size());
	}
	swap_events_.push_back({Store, ids});
	}

	std::future<std::vector<std::shared_ptr<core::FlowFile>>> load(std::vector<minifi::SwappedFlowFile> flow_files) override {
	swap_events_.push_back({Load, flow_files});
	LoadTask load_task;
	auto future = load_task.promise.get_future();
	load_task.result.reserve(flow_files.size());
	for (const auto& ff_id : flow_files) {
	std::string value;
	Get(ff_id.id.to_string().c_str(), value);
	minifi::utils::Identifier container_id;
	auto ff = minifi::FlowFileRecord::DeSerialize(gsl::make_span(value).as_span<const std::byte>(), content_repo_, container_id);
	ff->setPenaltyExpiration(ff_id.to_be_processed_after);
	load_task.result.push_back(std::move(ff));
	}
	load_tasks_.push_back(std::move(load_task));
	return future;
	}

	struct LoadTask {
	std::promise<std::vector<std::shared_ptr<core::FlowFile>>> promise;
	std::vector<std::shared_ptr<core::FlowFile>> result;

	void complete() {
	promise.set_value(result);
	}
	};

	std::vector<LoadTask> load_tasks_;
	std::vector<SwapEvent> swap_events_;
	};

	using FlowFilePtr = std::shared_ptr<core::FlowFile>;
	using FlowFilePtrVec = std::vector<FlowFilePtr>;

	struct FlowFileComparator {
	bool operator()(const FlowFilePtr& left, const FlowFilePtr& right) const {
	return left->getPenaltyExpiration() < right->getPenaltyExpiration();
	}
	};

	struct VerifiedQueue {
	void push(FlowFilePtr ff) {
	size();
	impl.push(ff);
	ref_.insert(std::lower_bound(ref_.begin(), ref_.end(), ff, FlowFileComparator{}), ff);
	size();
	}

	FlowFilePtr poll() {
	size();
	FlowFilePtr ff = impl.pop();
	REQUIRE(!ref_.empty());
	// the order when flow files have the same penalty is not fixed
	REQUIRE(ff->getPenaltyExpiration() == ref_.front()->getPenaltyExpiration());
	ref_.erase(ref_.begin());
	size();
	return ff;
	}

	void verify(std::initializer_list<unsigned> live, std::optional<std::initializer_list<unsigned>> inter, std::initializer_list<unsigned> swapped) const {
	// check live ffs
	auto live_copy = utils::FlowFileQueueTestAccessor::get_queue_(impl);
	REQUIRE(live_copy.size() == live.size());
	for (auto sec : live) {
	auto min = live_copy.popMin();
	REQUIRE(min->getPenaltyExpiration() == Timepoint{std::chrono::seconds{sec}});
	}

	// check inter ffs
	if (!inter) {
	REQUIRE_FALSE(utils::FlowFileQueueTestAccessor::get_load_task_(impl).has_value());
	} else {
	auto& intermediate = utils::FlowFileQueueTestAccessor::get_load_task_(impl)->intermediate_items;
	REQUIRE(intermediate.size() == inter->size());
	size_t idx = 0;
	for (auto sec : inter.value()) {
	REQUIRE(intermediate[idx]->getPenaltyExpiration() == Timepoint{std::chrono::seconds{sec}});
	++idx;
	}
	}

	// check swapped ffs
	auto swapped_copy = utils::FlowFileQueueTestAccessor::get_swapped_flow_files_(impl);
	REQUIRE(swapped_copy.size() == swapped.size());
	for (auto sec : swapped) {
	auto min = swapped_copy.popMin();
	REQUIRE(min.to_be_processed_after == Timepoint{std::chrono::seconds{sec}});
	}
	}

	bool isWorkAvailable() const {
	return impl.isWorkAvailable();
	}

	size_t size() const {
	size_t result = impl.size();
	REQUIRE(result == ref_.size());
	return result;
	}

	explicit VerifiedQueue(std::shared_ptr<minifi::SwapManager> swap_manager)
	: impl(std::move(swap_manager)) {}

	minifi::utils::FlowFileQueue impl;
	FlowFilePtrVec ref_;
	};

	class SwapTestController : public TestController {
	public:
	SwapTestController() {
	content_repo_ = std::make_shared<core::repository::VolatileContentRepository>();
	flow_repo_ = std::make_shared<SwappingFlowFileTestRepo>();
	flow_repo_->loadComponent(content_repo_);
	clock_ = std::make_shared<minifi::test::utils::ManualClock>();
	minifi::utils::timeutils::setClock(clock_);
	queue_ = std::make_shared<VerifiedQueue>(std::static_pointer_cast<minifi::SwapManager>(flow_repo_));
	}

	void setLimits(size_t min_size, size_t target_size, size_t max_size) {
	queue_->impl.setMinSize(min_size);
	queue_->impl.setTargetSize(target_size);
	queue_->impl.setMaxSize(max_size);
	}

	struct SwapEventPattern {
	EventKind kind;
	std::initializer_list<unsigned > seconds;
	};

	void verifySwapEvents(std::vector<SwapEventPattern> events) {
	REQUIRE(flow_repo_->swap_events_.size() == events.size());
	size_t idx = 0;
	for (auto& pattern : events) {
	REQUIRE(pattern.kind == flow_repo_->swap_events_[idx].kind);
	flow_repo_->swap_events_[idx].verifyTimes(pattern.seconds);
	}
	}

	void clearSwapEvents() {
	flow_repo_->swap_events_.clear();
	}

	void verifyQueue(std::initializer_list<unsigned> live, std::optional<std::initializer_list<unsigned>> inter, std::initializer_list<unsigned> swapped) {
	queue_->verify(live, inter, swapped);
	}

	void pushAll(std::initializer_list<unsigned> seconds) {
	for (auto sec : seconds) {
	auto ff = std::static_pointer_cast<core::FlowFile>(std::make_shared<minifi::FlowFileRecordImpl>());
	ff->setPenaltyExpiration(Timepoint{std::chrono::seconds{sec}});
	queue_->push(std::move(ff));
	}
	}

	void popAll(std::initializer_list<unsigned> seconds, bool check_is_work_available = false) {
	for (auto sec : seconds) {
	if (check_is_work_available) {
	REQUIRE(queue_->isWorkAvailable());
	}
	auto ff = queue_->poll();
	REQUIRE(ff->getPenaltyExpiration() == Timepoint{std::chrono::seconds{sec}});
	}
	}

	std::shared_ptr<SwappingFlowFileTestRepo> flow_repo_;
	std::shared_ptr<core::repository::VolatileContentRepository> content_repo_;
	std::shared_ptr<VerifiedQueue> queue_;
	std::shared_ptr<minifi::test::utils::ManualClock> clock_;
	};

	TEST_CASE("Setting swap threshold sets underlying queue limits", "[SwapTest1]") {
	const size_t target_size = 4;
	const size_t min_size = target_size / 2;
	const size_t max_size = target_size * 3 / 2;

	minifi::ConnectionImpl conn(nullptr, nullptr, "");
	conn.setSwapThreshold(target_size);
	REQUIRE(utils::FlowFileQueueTestAccessor::get_min_size_(utils::ConnectionTestAccessor::get_queue_(conn)) == min_size);
	REQUIRE(utils::FlowFileQueueTestAccessor::get_target_size_(utils::ConnectionTestAccessor::get_queue_(conn)) == target_size);
	REQUIRE(utils::FlowFileQueueTestAccessor::get_max_size_(utils::ConnectionTestAccessor::get_queue_(conn)) == max_size);
	}

	TEST_CASE_METHOD(SwapTestController, "Default constructed FlowFileQueue won't swap", "[SwapTest2]") {
	for (unsigned i = 0; i < 100; ++i) {
	pushAll({i});
	}

	REQUIRE(queue_->impl.size() == 100);

	clock_->advance(std::chrono::seconds{200});
	for (size_t i = 0; i < 100; ++i) {
	queue_->poll();
	}

	REQUIRE(queue_->impl.empty());
	verifySwapEvents({});
	}

	TEST_CASE_METHOD(SwapTestController, "Up to max no swap-out is triggered", "[SwapTest3]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40});

	REQUIRE_FALSE(queue_->isWorkAvailable());
	verifySwapEvents({});
	}

	TEST_CASE_METHOD(SwapTestController, "Pushing beyond max triggers a swap-out", "[SwapTest4]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40});

	pushAll({28});
	// size goes from 7 to 4, 3 largest must have been swapped out
	verifySwapEvents({{Store, {60, 50, 40}}});
	verifyQueue({10, 20, 28, 30}, {}, {40, 50, 60});
	}

	TEST_CASE_METHOD(SwapTestController, "Popping until min size does not trigger swap-in", "[SwapTest5]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40, 28});
	clearSwapEvents();
	clock_->advance(std::chrono::seconds{35});
	REQUIRE(queue_->isWorkAvailable());
	popAll({10, 20});
	verifyQueue({28, 30}, {}, {40, 50, 60});
	verifySwapEvents({});
	}

	TEST_CASE_METHOD(SwapTestController, "Popping beyond min size triggers swap-in", "[SwapTest6]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40, 28});
	clearSwapEvents();
	clock_->advance(std::chrono::seconds{35});
	popAll({10, 20, 28});

	// trying to swap-in all three swapped flow files
	verifyQueue({30}, {{}}, {});
	verifySwapEvents({{Load, {40, 50, 60}}});
	}

	TEST_CASE_METHOD(SwapTestController, "Pushing while a swap-in is pending", "[SwapTest7]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40, 28});
	clock_->advance(std::chrono::seconds{35});
	popAll({10, 20, 28});
	verifyQueue({30}, {{}}, {});
	clearSwapEvents();

	SECTION("Pushing into the pending swap-in range") {
	pushAll({45});
	verifyQueue({30}, {{45}}, {});
	verifySwapEvents({});
	}

	SECTION("Pushing before the pending swap-in range") {
	pushAll({35});
	verifyQueue({30, 35}, {{}}, {});
	verifySwapEvents({});
	}

	SECTION("Pushing after the pending swap-in range") {
	pushAll({65});
	verifyQueue({30}, {{}}, {65});
	verifySwapEvents({{Store, {65}}});
	}
	}

	TEST_CASE_METHOD(SwapTestController, "isWorkAvailable depends on the swap-in task", "[SwapTest8]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40, 28});
	popAll({10, 20, 28});
	verifyQueue({30}, {{}}, {});

	REQUIRE_FALSE(queue_->isWorkAvailable());

	clock_->advance(std::chrono::seconds{35});

	REQUIRE(queue_->isWorkAvailable());
	popAll({30});
	REQUIRE_FALSE(queue_->isWorkAvailable());

	SECTION("Load is completed but the minimum of those files is still not viable") {
	flow_repo_->load_tasks_[0].complete();
	REQUIRE_FALSE(queue_->isWorkAvailable());
	}

	SECTION("The minimum of the load task is viable but not yet completed") {
	clock_->advance(std::chrono::seconds{35});
	REQUIRE_FALSE(queue_->isWorkAvailable());
	// completing the task renders it viable
	flow_repo_->load_tasks_[0].complete();
	REQUIRE(queue_->isWorkAvailable());
	}
	}

	TEST_CASE_METHOD(SwapTestController, "Polling from load task", "[SwapTest8]") {
	setLimits(2, 4, 6);
	pushAll({50, 20, 30, 60, 10, 40, 28});
	popAll({10, 20, 28});
	pushAll({45});
	verifyQueue({30}, {{45}}, {});

	flow_repo_->load_tasks_[0].complete();

	clock_->advance(std::chrono::seconds{100});

	popAll({30, 40, 45, 50, 60}, true);
	verifyQueue({}, {}, {});
	}

	TEST_CASE_METHOD(SwapTestController, "Popping below min checks if the pending load is finished", "[SwapTest8]") {
	setLimits(6, 8, 10);
	pushAll({10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120});
	verifyQueue({10, 20, 30, 40, 50, 60, 70, 80}, {}, {90, 100, 110, 120});
	clock_->advance(std::chrono::seconds{200});
	clearSwapEvents();
	popAll({10, 20, 30});
	verifySwapEvents({{Load, {90, 100, 110}}});
	verifyQueue({40, 50, 60, 70, 80}, {{}}, {120});
	clearSwapEvents();

	popAll({40, 50});
	verifyQueue({60, 70, 80}, {{}}, {120});
	flow_repo_->load_tasks_[0].complete();
	popAll({60});
	// even though the live queue is not empty we check if
	// the load_task is finished and initiate a load if need be
	verifySwapEvents({{Load, {120}}});
	verifyQueue({70, 80, 90, 100, 110}, {{}}, {});
	}

	} // namespace org::apache::nifi::minifi::test