lordgamez commented on a change in pull request #1195:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1195#discussion_r752069518
##########
File path: PROCESSORS.md
##########
@@ -1278,10 +1303,10 @@ In the list below, the names of required properties
appear in bold. Any other pr
| Name | Default Value | Allowable Values | Description |
| - | - | - | - |
|**Azure Storage Credentials Service**|||Name of the Azure Storage Credentials
Service used to retrieve the connection string from.|
+|**Conflict Resolution Strategy**|fail|fail<br/>replace<br/>ignore|Indicates
what should happen when a file with the same name already exists in the output
directory.|
Review comment:
I think this is a misunderstanding, this documentation change belongs to
the PutAzureDataLakeStorage processor not the DeleteAzureDataLakeStorage. The
reason for the change was to put the properties in alphabetical order.
##########
File path: extensions/azure/processors/AzureDataLakeStorageProcessorBase.cpp
##########
@@ -0,0 +1,80 @@
+/**
+ * @file AzureDataLakeStorageProcessorBase.cpp
+ * AzureDataLakeStorageProcessorBase class implementation
+ *
+ * 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 "AzureDataLakeStorageProcessorBase.h"
+
+#include "utils/ProcessorConfigUtils.h"
+#include "controllerservices/AzureStorageCredentialsService.h"
+
+namespace org::apache::nifi::minifi::azure::processors {
+
+const core::Property AzureDataLakeStorageProcessorBase::FilesystemName(
+ core::PropertyBuilder::createProperty("Filesystem Name")
+ ->withDescription("Name of the Azure Storage File System. It is assumed
to be already existing.")
+ ->supportsExpressionLanguage(true)
+ ->isRequired(true)
+ ->build());
+const core::Property AzureDataLakeStorageProcessorBase::DirectoryName(
+ core::PropertyBuilder::createProperty("Directory Name")
+ ->withDescription("Name of the Azure Storage Directory. The Directory
Name cannot contain a leading '/'. "
+ "If left empty it designates the root directory. The
directory will be created if not already existing.")
+ ->supportsExpressionLanguage(true)
+ ->build());
+const core::Property AzureDataLakeStorageProcessorBase::FileName(
+ core::PropertyBuilder::createProperty("File Name")
+ ->withDescription("The filename in Azure Storage. If left empty the
filename attribute will be used by default.")
+ ->supportsExpressionLanguage(true)
+ ->build());
+
+void AzureDataLakeStorageProcessorBase::onSchedule(const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) {
+ std::optional<storage::AzureStorageCredentials> credentials;
+ std::tie(std::ignore, credentials) =
getCredentialsFromControllerService(context);
+ if (!credentials) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials
Service property missing or invalid");
+ }
+
+ if (!credentials->isValid()) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials
Service properties are not set or invalid");
+ }
Review comment:
There is actually a difference, but the message is wrong. The difference
is if the service property is set to an invalid value or the credentials inside
the controller service. Fixed the message in
3ae91653cec560aece9da318071afef2abee7702
##########
File path: extensions/azure/processors/AzureDataLakeStorageProcessorBase.cpp
##########
@@ -0,0 +1,80 @@
+/**
+ * @file AzureDataLakeStorageProcessorBase.cpp
+ * AzureDataLakeStorageProcessorBase class implementation
+ *
+ * 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 "AzureDataLakeStorageProcessorBase.h"
+
+#include "utils/ProcessorConfigUtils.h"
+#include "controllerservices/AzureStorageCredentialsService.h"
+
+namespace org::apache::nifi::minifi::azure::processors {
+
+const core::Property AzureDataLakeStorageProcessorBase::FilesystemName(
+ core::PropertyBuilder::createProperty("Filesystem Name")
+ ->withDescription("Name of the Azure Storage File System. It is assumed
to be already existing.")
+ ->supportsExpressionLanguage(true)
+ ->isRequired(true)
+ ->build());
+const core::Property AzureDataLakeStorageProcessorBase::DirectoryName(
+ core::PropertyBuilder::createProperty("Directory Name")
+ ->withDescription("Name of the Azure Storage Directory. The Directory
Name cannot contain a leading '/'. "
+ "If left empty it designates the root directory. The
directory will be created if not already existing.")
+ ->supportsExpressionLanguage(true)
+ ->build());
+const core::Property AzureDataLakeStorageProcessorBase::FileName(
+ core::PropertyBuilder::createProperty("File Name")
+ ->withDescription("The filename in Azure Storage. If left empty the
filename attribute will be used by default.")
+ ->supportsExpressionLanguage(true)
+ ->build());
+
+void AzureDataLakeStorageProcessorBase::onSchedule(const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) {
+ std::optional<storage::AzureStorageCredentials> credentials;
+ std::tie(std::ignore, credentials) =
getCredentialsFromControllerService(context);
+ if (!credentials) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials
Service property missing or invalid");
+ }
+
+ if (!credentials->isValid()) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials
Service properties are not set or invalid");
+ }
+
+ credentials_ = *credentials;
+}
+
+bool AzureDataLakeStorageProcessorBase::setCommonParameters(
+ storage::AzureDataLakeStorageParameters& params, const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::FlowFile>& flow_file) {
+ params.credentials = credentials_;
+
+ if (!context->getProperty(FilesystemName, params.file_system_name,
flow_file) || params.file_system_name.empty()) {
+ logger_->log_error("Filesystem Name '%s' is invalid or empty!",
params.file_system_name);
+ return false;
+ }
+
+ context->getProperty(DirectoryName, params.directory_name, flow_file);
+
+ context->getProperty(FileName, params.filename, flow_file);
Review comment:
Updated in 3ae91653cec560aece9da318071afef2abee7702
##########
File path: extensions/azure/processors/AzureDataLakeStorageProcessorBase.cpp
##########
@@ -0,0 +1,80 @@
+/**
+ * @file AzureDataLakeStorageProcessorBase.cpp
+ * AzureDataLakeStorageProcessorBase class implementation
+ *
+ * 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 "AzureDataLakeStorageProcessorBase.h"
+
+#include "utils/ProcessorConfigUtils.h"
+#include "controllerservices/AzureStorageCredentialsService.h"
+
+namespace org::apache::nifi::minifi::azure::processors {
+
+const core::Property AzureDataLakeStorageProcessorBase::FilesystemName(
+ core::PropertyBuilder::createProperty("Filesystem Name")
+ ->withDescription("Name of the Azure Storage File System. It is assumed
to be already existing.")
+ ->supportsExpressionLanguage(true)
+ ->isRequired(true)
+ ->build());
+const core::Property AzureDataLakeStorageProcessorBase::DirectoryName(
+ core::PropertyBuilder::createProperty("Directory Name")
+ ->withDescription("Name of the Azure Storage Directory. The Directory
Name cannot contain a leading '/'. "
+ "If left empty it designates the root directory. The
directory will be created if not already existing.")
+ ->supportsExpressionLanguage(true)
+ ->build());
+const core::Property AzureDataLakeStorageProcessorBase::FileName(
+ core::PropertyBuilder::createProperty("File Name")
+ ->withDescription("The filename in Azure Storage. If left empty the
filename attribute will be used by default.")
+ ->supportsExpressionLanguage(true)
+ ->build());
+
+void AzureDataLakeStorageProcessorBase::onSchedule(const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) {
Review comment:
Updated in 3ae91653cec560aece9da318071afef2abee7702
##########
File path: extensions/azure/processors/DeleteAzureDataLakeStorage.cpp
##########
@@ -0,0 +1,84 @@
+/**
+ * @file DeleteAzureDataLakeStorage.cpp
+ * DeleteAzureDataLakeStorage class implementation
+ *
+ * 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 "DeleteAzureDataLakeStorage.h"
+
+#include "utils/ProcessorConfigUtils.h"
+#include "utils/gsl.h"
+#include "core/Resource.h"
+
+namespace org::apache::nifi::minifi::azure::processors {
+
+const core::Relationship DeleteAzureDataLakeStorage::Success("success", "If
file deletion from Azure storage succeeds the flowfile is transferred to this
relationship");
+const core::Relationship DeleteAzureDataLakeStorage::Failure("failure", "If
file deletion from Azure storage fails the flowfile is transferred to this
relationship");
+
+void DeleteAzureDataLakeStorage::initialize() {
+ // Set the supported properties
+ setSupportedProperties({
+ AzureStorageCredentialsService,
+ FilesystemName,
+ DirectoryName,
+ FileName
+ });
+
+ // Set the supported relationships
+ setSupportedRelationships({
+ Success,
+ Failure
+ });
+}
+
+std::optional<storage::DeleteAzureDataLakeStorageParameters>
DeleteAzureDataLakeStorage::buildDeleteParameters(
+ const std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::FlowFile>& flow_file) {
+ storage::DeleteAzureDataLakeStorageParameters params;
+ if (!setCommonParameters(params, context, flow_file)) {
+ return std::nullopt;
+ }
+
+ return params;
+}
+
+void DeleteAzureDataLakeStorage::onTrigger(const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSession>& session) {
Review comment:
Updated in 3ae91653cec560aece9da318071afef2abee7702
##########
File path: extensions/azure/processors/PutAzureDataLakeStorage.cpp
##########
@@ -71,18 +55,8 @@ void PutAzureDataLakeStorage::initialize() {
});
}
-void PutAzureDataLakeStorage::onSchedule(const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) {
- std::optional<storage::AzureStorageCredentials> credentials;
- std::tie(std::ignore, credentials) =
getCredentialsFromControllerService(context);
- if (!credentials) {
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials
Service property missing or invalid");
- }
-
- if (!credentials->isValid()) {
- throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials
Service properties are not set or invalid");
- }
-
- credentials_ = *credentials;
+void PutAzureDataLakeStorage::onSchedule(const
std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) {
+ AzureDataLakeStorageProcessorBase::onSchedule(context, sessionFactory);
Review comment:
Updated in 3ae91653cec560aece9da318071afef2abee7702
##########
File path: libminifi/test/azure-tests/AzureDataLakeStorageTestsFixture.h
##########
@@ -0,0 +1,123 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include <utility>
+#include <vector>
+#include <memory>
+#include <string>
+
+#include "MockDataLakeStorageClient.h"
+#include "../TestBase.h"
+#include "utils/TestUtils.h"
+#include "utils/IntegrationTestUtils.h"
+#include "core/Processor.h"
+#include "processors/GetFile.h"
+#include "processors/PutFile.h"
+#include "processors/LogAttribute.h"
+#include "processors/UpdateAttribute.h"
+#include "utils/file/FileUtils.h"
+#include "controllerservices/AzureStorageCredentialsService.h"
+
+const std::string FILESYSTEM_NAME = "testfilesystem";
+const std::string DIRECTORY_NAME = "testdir";
+const std::string FILE_NAME = "testfile.txt";
+const std::string CONNECTION_STRING = "test-connectionstring";
+const std::string TEST_DATA = "data123";
+const std::string GETFILE_FILE_NAME = "input_data.log";
+
+template<typename AzureDataLakeStorageProcessor>
+class AzureDataLakeStorageTestsFixture {
+ public:
+ AzureDataLakeStorageTestsFixture() {
+ LogTestController::getInstance().setDebug<TestPlan>();
+ LogTestController::getInstance().setDebug<minifi::core::Processor>();
+ LogTestController::getInstance().setTrace<minifi::core::ProcessSession>();
+ LogTestController::getInstance().setTrace<minifi::processors::GetFile>();
+ LogTestController::getInstance().setTrace<minifi::processors::PutFile>();
+
LogTestController::getInstance().setDebug<minifi::processors::UpdateAttribute>();
+
LogTestController::getInstance().setDebug<minifi::processors::LogAttribute>();
+ LogTestController::getInstance().setTrace<AzureDataLakeStorageProcessor>();
+
+ // Build MiNiFi processing graph
+ plan_ = test_controller_.createPlan();
+ auto mock_data_lake_storage_client =
std::make_unique<MockDataLakeStorageClient>();
+ mock_data_lake_storage_client_ptr_ = mock_data_lake_storage_client.get();
+ azure_data_lake_storage_ = std::shared_ptr<AzureDataLakeStorageProcessor>(
+ new AzureDataLakeStorageProcessor("AzureDataLakeStorageProcessor",
utils::Identifier(), std::move(mock_data_lake_storage_client)));
+ auto input_dir = test_controller_.createTempDirectory();
+ utils::putFileToDir(input_dir, GETFILE_FILE_NAME, TEST_DATA);
+
+ get_file_ = plan_->addProcessor("GetFile", "GetFile");
+ plan_->setProperty(get_file_,
minifi::processors::GetFile::Directory.getName(), input_dir);
+ plan_->setProperty(get_file_,
minifi::processors::GetFile::KeepSourceFile.getName(), "false");
+
+ update_attribute_ = plan_->addProcessor("UpdateAttribute",
"UpdateAttribute", { {"success", "d"} }, true);
+ plan_->addProcessor(azure_data_lake_storage_,
"AzureDataLakeStorageProcessor", { {"success", "d"}, {"failure", "d"} }, true);
+ auto logattribute = plan_->addProcessor("LogAttribute", "LogAttribute", {
{"success", "d"} }, true);
+ logattribute->setAutoTerminatedRelationships({{"success", "d"}});
+
+ putfile_ = plan_->addProcessor("PutFile", "PutFile", { {"success", "d"} },
false);
+ plan_->addConnection(azure_data_lake_storage_, {"failure", "d"}, putfile_);
+ putfile_->setAutoTerminatedRelationships({{"success", "d"}, {"failure",
"d"}});
+ output_dir_ = test_controller_.createTempDirectory();
+ plan_->setProperty(putfile_,
org::apache::nifi::minifi::processors::PutFile::Directory.getName(),
output_dir_);
+
+ azure_storage_cred_service_ =
plan_->addController("AzureStorageCredentialsService",
"AzureStorageCredentialsService");
+ setDefaultProperties();
+ }
+
+ std::vector<std::string> getFailedFlowFileContents() {
+ std::vector<std::string> file_contents;
+
+ auto lambda = [&file_contents](const std::string& path, const std::string&
filename) -> bool {
+ std::ifstream is(path + utils::file::FileUtils::get_separator() +
filename, std::ifstream::binary);
+ std::string file_content((std::istreambuf_iterator<char>(is)),
std::istreambuf_iterator<char>());
+ file_contents.push_back(file_content);
Review comment:
Updated in 3ae91653cec560aece9da318071afef2abee7702
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
