szaszm commented on a change in pull request #791:
URL: https://github.com/apache/nifi-minifi-cpp/pull/791#discussion_r433821309
##########
File path: extensions/standard-processors/processors/TailFile.cpp
##########
@@ -204,52 +230,65 @@ void TailFile::parseStateFileLine(char *buf) {
}
std::string value = equal;
- key = trimRight(key);
- value = trimRight(value);
+ key = utils::StringUtils::trimRight(key);
+ value = utils::StringUtils::trimRight(value);
if (key == "FILENAME") {
std::string fileLocation, fileName;
if (utils::file::PathUtils::getFileNameAndPath(value, fileLocation,
fileName)) {
logger_->log_debug("State migration received path %s, file %s",
fileLocation, fileName);
- tail_states_.insert(std::make_pair(fileName, TailState { fileLocation,
fileName, 0, 0 }));
+ state.insert(std::make_pair(fileName, TailState{fileLocation, fileName,
0, 0, 0, 0}));
} else {
- tail_states_.insert(std::make_pair(value, TailState { fileLocation,
value, 0, 0 }));
+ state.insert(std::make_pair(value, TailState{fileLocation, value, 0, 0,
0, 0}));
}
}
if (key == "POSITION") {
// for backwards compatibility
- if (tail_states_.size() != 1) {
+ if (tail_states_.size() != (std::size_t) 1) {
throw minifi::Exception(ExceptionType::PROCESSOR_EXCEPTION,
"Incompatible state file types");
}
const auto position = std::stoull(value);
logger_->log_debug("Received position %d", position);
- tail_states_.begin()->second.currentTailFilePosition_ = position;
+ state.begin()->second.position_ = position;
}
if (key.find(CURRENT_STR) == 0) {
const auto file = key.substr(strlen(CURRENT_STR));
std::string fileLocation, fileName;
if (utils::file::PathUtils::getFileNameAndPath(value, fileLocation,
fileName)) {
- tail_states_[file].path_ = fileLocation;
- tail_states_[file].current_file_name_ = fileName;
+ state[file].path_ = fileLocation;
+ state[file].file_name_ = fileName;
} else {
throw minifi::Exception(ExceptionType::PROCESSOR_EXCEPTION, "State file
contains an invalid file name");
}
}
if (key.find(POSITION_STR) == 0) {
const auto file = key.substr(strlen(POSITION_STR));
- tail_states_[file].currentTailFilePosition_ = std::stoull(value);
+ state[file].position_ = std::stoull(value);
}
}
+bool TailFile::recoverState(const std::shared_ptr<core::ProcessContext>&
context) {
+ std::map<std::string, TailState> new_tail_states;
+ bool state_load_success = getStateFromStateManager(new_tail_states) ||
+ getStateFromLegacyStateFile(new_tail_states);
+ if (!state_load_success) {
+ return false;
+ }
+ logger_->log_debug("load state succeeded");
-bool TailFile::recoverState(const std::shared_ptr<core::ProcessContext>&
context) {
- bool state_load_success = false;
+ tail_states_ = std::move(new_tail_states);
+
+ // Save the state to the state manager
+ storeState(context);
Review comment:
Is this `storeState()` call required to fix the bug? It is not always
cheap, because one can configure the state storage to be durable, i.e. flush
the WAL on every commit in the rocksdb case, causing a blocking IO operation.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
