You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2021/11/18 09:47:50 UTC
[GitHub] [nifi-minifi-cpp] lordgamez commented on a change in pull request #1195: MINIFICPP-1629 Create DeleteAzureDataLakeStorage processor
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org