Method rollback

libminifi/src/core/ProcessSession.cpp:783–831

}

void ProcessSession::rollback() {
  // new FlowFiles are only persisted during commit
  // no need to delete them here
  std::map<std::shared_ptr<Connectable>, std::vector<std::shared_ptr<FlowFile>>> connectionQueues;

  try {
    // Requeue the snapshot of the flowfile back
    for (const auto &it : _updatedFlowFiles) {
      auto flowFile = it.second.modified;
      // restore flowFile to original state
      *flowFile = *it.second.snapshot;
      logger_->log_debug("ProcessSession rollback for %s, record %s, to connection %s",
          process_context_->getProcessorNode()->getName(),
          flowFile->getUUIDStr(),
          flowFile->getConnection()->getName());
      connectionQueues[flowFile->getConnection()].push_back(flowFile);
    }

    for (const auto& record : _deletedFlowFiles) {
      record->setDeleted(false);
    }

    // put everything back where it came from
    for (auto& cq : connectionQueues) {
      auto connection = std::dynamic_pointer_cast<Connection>(cq.first);
      if (connection) {
        connection->multiPut(cq.second);
      } else {
        for (auto& flow : cq.second) {
          cq.first->put(flow);
        }
      }
    }

    content_session_->rollback();

    _clonedFlowFiles.clear();
    _addedFlowFiles.clear();
    _updatedFlowFiles.clear();
    _deletedFlowFiles.clear();
    logger_->log_warn("ProcessSession rollback for %s executed", process_context_->getProcessorNode()->getName());
  } catch (std::exception &exception) {
    logger_->log_warn("Caught Exception during process session rollback: %s", exception.what());
    throw;
  } catch (...) {
    logger_->log_warn("Caught Exception during process session rollback");
    throw;
  }
}

void ProcessSession::persistFlowFilesBeforeTransfer(
    std::map<std::shared_ptr<Connectable>, std::vector<std::shared_ptr<core::FlowFile> > >& transactionMap,