You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by milanchandna <gi...@git.apache.org> on 2017/09/18 08:25:44 UTC
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
GitHub user milanchandna opened a pull request:
https://github.com/apache/nifi/pull/2158
NIFI-4360 Adding support for ADLS Processors. Feature includes List, …
…Get, Put processors. Till now ADLS interaction was possible using HDFS processors. Now users can ingress and egress their data directly using ADLS processors. It is much simpler and easy. And one lesser layer to go through as well. This will also help users who are not familiar with Hadoop configurations.
Thank you for submitting a contribution to Apache NiFi.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [x] Has your PR been rebased against the latest commit within the target branch (typically master)?
- [x] Is your initial contribution a single, squashed commit?
### For code changes:
- [ ] Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder? - There were failures in other modules testRecoverFileThatHasTrailingNULBytesAndTruncation(org.wali.TestMinimalLockingWriteAheadLog)
- [x] Have you written or updated unit tests to verify your changes?
- [x] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [x] If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
- [ ] If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
- [ ] If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [x] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/milanchandna/nifi NIFI-4360
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/2158.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2158
----
commit 4f449c55c99868bd8b96bc370186503f4928c11f
Author: Milan Chandna <mi...@gmail.com>
Date: 2017-09-18T07:00:17Z
NIFI-4360 Adding support for ADLS Processors. Feature includes List, Get, Put processors.
----
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140447052
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ADLSConstants.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+public class ADLSConstants {
+
+ public static final int CHUNK_SIZE_IN_BYTES = 4000000;
+
+ public static final PropertyDescriptor PATH_NAME = new PropertyDescriptor.Builder()
+ .name("Path")
+ .description("Path for file in Azure Data Lake, e.g. /adlshome/")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor RETRY_ON_FAIL = new PropertyDescriptor.Builder()
+ .name("Overwrite policy")
+ .description("How many times to retry if read fails per chunk read, defaults to 3")
+ .required(true)
+ //TODO add Integer validator
+ .defaultValue("3")
+ .build();
+
+ public static final PropertyDescriptor ACCOUNT_NAME = new PropertyDescriptor.Builder()
+ .name("Account FQDN")
+ .description("Azure account fully qualified domain name eg: accountname.azuredatalakestore.net")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLIENT_ID = new PropertyDescriptor.Builder()
+ .name("Client ID")
+ .description("Azure client ID")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLIENT_SECRET = new PropertyDescriptor.Builder()
+ .name("Client secret")
+ .description("Azure client secret")
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor AUTH_TOKEN_ENDPOINT = new PropertyDescriptor.Builder()
+ .name("Auth token endpoint")
+ .description("Azure client secret")
--- End diff --
copy/paste description
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140503820
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/adls/TestPutADLSFile.java ---
@@ -0,0 +1,522 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.ADLStoreOptions;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.QueueDispatcher;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+
+public class TestPutADLSFile {
+
+ private MockWebServer server;
+ private ADLStoreClient client;
+ private Processor processor;
+ private TestRunner runner;
+ Map<String, String> fileAttributes;
+
+ @Before
+ public void init() throws IOException, InitializationException {
+ server = new MockWebServer();
+ QueueDispatcher dispatcher = new QueueDispatcher();
+ dispatcher.setFailFast(new MockResponse().setResponseCode(400));
+ server.setDispatcher(dispatcher);
+ server.start();
+ String accountFQDN = server.getHostName() + ":" + server.getPort();
+ String dummyToken = "testDummyAadToken";
+
+ client = ADLStoreClient.createClient(accountFQDN, dummyToken);
+ client.setOptions(new ADLStoreOptions().setInsecureTransport());
+
+ processor = new PutADLSWithMockClient(client);
+
+ runner = TestRunners.newTestRunner(processor);
+
+ runner.setProperty(ADLSConstants.ACCOUNT_NAME, accountFQDN);
+ runner.setProperty(ADLSConstants.CLIENT_ID, "foobar");
+ runner.setProperty(ADLSConstants.CLIENT_SECRET, "foobar");
+ runner.setProperty(ADLSConstants.AUTH_TOKEN_ENDPOINT, "foobar");
+ runner.setProperty(PutADLSFile.DIRECTORY, "/sample/");
+ runner.setProperty(PutADLSFile.CONFLICT_RESOLUTION, PutADLSFile.FAIL_RESOLUTION_AV);
+ runner.removeProperty(PutADLSFile.UMASK);
+ runner.removeProperty(PutADLSFile.ACL);
+
+ fileAttributes = new HashMap<>();
+ fileAttributes.put("filename", "sample.txt");
+ fileAttributes.put("path", "/root/");
+ }
+
+ @Test
+ public void testPutConflictFail() throws IOException, InterruptedException {
+
+ //for create call
+ server.enqueue(new MockResponse().setResponseCode(200));
+ //for append call(internal call while writing to stream)
+ server.enqueue(new MockResponse().setResponseCode(200));
+ //for rename call(since its fail conflict)
+ server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"boolean\":true}"));
+ //for delete call(removing temp file)
+ server.enqueue(new MockResponse().setResponseCode(200));
+
+// String bodySimpleFileContent = new String(Files.readAllBytes(Paths.get(bodySimpleFileName)));
--- End diff --
yes, removed.
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on the issue:
https://github.com/apache/nifi/pull/2158
Thanks @joewitt
These test resource files contains Lorem Ipsum text.
But anyways I will create my own if required but dont want to delete as they are required for an important test case.
Thanks.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140446942
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ADLSConstants.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+public class ADLSConstants {
+
+ public static final int CHUNK_SIZE_IN_BYTES = 4000000;
+
+ public static final PropertyDescriptor PATH_NAME = new PropertyDescriptor.Builder()
+ .name("Path")
--- End diff --
Could you use ``.name()`` and ``.displayName()`` ?
Name cannot be changed once we release the processor because it'd break backward compatibility on existing workflow. But we can change the display name as it's only used in logs/UI. We usually use a script-friendly naming for the name field (no whitespace, lower case, etc). For instance: adls-overwrite-policy.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r143404785
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java ---
@@ -0,0 +1,289 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.DirectoryEntry;
+import com.microsoft.azure.datalake.store.DirectoryEntryType;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ACCOUNT_NAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@TriggerSerially
+@TriggerWhenEmpty
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"hadoop", "ADLS", "get", "list", "ingest", "ingress", "source", "filesystem"})
--- End diff --
done.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r143422216
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java ---
@@ -0,0 +1,289 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.DirectoryEntry;
+import com.microsoft.azure.datalake.store.DirectoryEntryType;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ACCOUNT_NAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@TriggerSerially
+@TriggerWhenEmpty
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"hadoop", "ADLS", "get", "list", "ingest", "ingress", "source", "filesystem"})
--- End diff --
done.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140448507
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java ---
@@ -0,0 +1,267 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.acl.AclEntry;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.*;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"hadoop", "ADLS", "put", "copy", "filesystem", "restricted"})
--- End diff --
Same comment here
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on the issue:
https://github.com/apache/nifi/pull/2158
Hey @pvillard31 Thanks for comments, looking into them.
Currently when I am running full build, I am getting failures in TestMinimalLockingWriteAheadLog.
testRecoverFileThatHasTrailingNULBytesAndTruncation(org.wali.TestMinimalLockingWriteAheadLog)
And because of this not able to reliably detect failures in my own module.
Any suggestion to skip the above failures?
-Milan.
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on the issue:
https://github.com/apache/nifi/pull/2158
Can I please get some feedback here. Thanks.
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on the issue:
https://github.com/apache/nifi/pull/2158
@joewitt, @jzonthemtn @pvillard31 - Hey, did you guys got a chance to look at this?
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140449455
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/adls/TestPutADLSFile.java ---
@@ -0,0 +1,522 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.ADLStoreOptions;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.QueueDispatcher;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+
+public class TestPutADLSFile {
+
+ private MockWebServer server;
+ private ADLStoreClient client;
+ private Processor processor;
+ private TestRunner runner;
+ Map<String, String> fileAttributes;
+
+ @Before
+ public void init() throws IOException, InitializationException {
+ server = new MockWebServer();
+ QueueDispatcher dispatcher = new QueueDispatcher();
+ dispatcher.setFailFast(new MockResponse().setResponseCode(400));
+ server.setDispatcher(dispatcher);
+ server.start();
+ String accountFQDN = server.getHostName() + ":" + server.getPort();
+ String dummyToken = "testDummyAadToken";
+
+ client = ADLStoreClient.createClient(accountFQDN, dummyToken);
+ client.setOptions(new ADLStoreOptions().setInsecureTransport());
+
+ processor = new PutADLSWithMockClient(client);
+
+ runner = TestRunners.newTestRunner(processor);
+
+ runner.setProperty(ADLSConstants.ACCOUNT_NAME, accountFQDN);
+ runner.setProperty(ADLSConstants.CLIENT_ID, "foobar");
+ runner.setProperty(ADLSConstants.CLIENT_SECRET, "foobar");
+ runner.setProperty(ADLSConstants.AUTH_TOKEN_ENDPOINT, "foobar");
+ runner.setProperty(PutADLSFile.DIRECTORY, "/sample/");
+ runner.setProperty(PutADLSFile.CONFLICT_RESOLUTION, PutADLSFile.FAIL_RESOLUTION_AV);
+ runner.removeProperty(PutADLSFile.UMASK);
+ runner.removeProperty(PutADLSFile.ACL);
+
+ fileAttributes = new HashMap<>();
+ fileAttributes.put("filename", "sample.txt");
+ fileAttributes.put("path", "/root/");
+ }
+
+ @Test
+ public void testPutConflictFail() throws IOException, InterruptedException {
+
+ //for create call
+ server.enqueue(new MockResponse().setResponseCode(200));
+ //for append call(internal call while writing to stream)
+ server.enqueue(new MockResponse().setResponseCode(200));
+ //for rename call(since its fail conflict)
+ server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"boolean\":true}"));
+ //for delete call(removing temp file)
+ server.enqueue(new MockResponse().setResponseCode(200));
+
+// String bodySimpleFileContent = new String(Files.readAllBytes(Paths.get(bodySimpleFileName)));
--- End diff --
line to be removed?
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140502906
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ADLSConstants.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+public class ADLSConstants {
+
+ public static final int CHUNK_SIZE_IN_BYTES = 4000000;
+
+ public static final PropertyDescriptor PATH_NAME = new PropertyDescriptor.Builder()
+ .name("Path")
--- End diff --
Used name and displayname properly wherever applicable.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140448400
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java ---
@@ -0,0 +1,289 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.DirectoryEntry;
+import com.microsoft.azure.datalake.store.DirectoryEntryType;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ACCOUNT_NAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@TriggerSerially
+@TriggerWhenEmpty
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"hadoop", "ADLS", "get", "list", "ingest", "ingress", "source", "filesystem"})
+@CapabilityDescription("Retrieves a listing of files from ADLS. For each file that is "
+ + "listed in ADLS, this processor creates a FlowFile that represents the ADLS file to be fetched in conjunction with FetchADLSFile. This Processor is "
+ + "designed to run on Primary Node only in a cluster. If the primary node changes, the new Primary Node will pick up where the previous node left "
+ + "off without duplicating all of the data.")
+@WritesAttributes({
+ @WritesAttribute(attribute="filename", description="The name of the file that was read from ADLS"),
+ @WritesAttribute(attribute="path", description="The path is set to the path of the file on ADLS"),
+ @WritesAttribute(attribute="adls.owner", description="The user that owns the file"),
+ @WritesAttribute(attribute="adls.group", description="The group that owns the file"),
+ @WritesAttribute(attribute="adls.lastModified", description="The timestamp of when the file was last modified, as milliseconds since midnight Jan 1, 1970 UTC"),
+ @WritesAttribute(attribute="adls.length", description="The number of bytes in the file"),
+ @WritesAttribute(attribute="adls.expirytime", description="The number of bytes in the file"),
+ @WritesAttribute(attribute="adls.permissions", description="The permissions for the file in ADLS. This is formatted as 3 characters for the owner, "
+ + "3 for the group, and 3 for other users. For example rw-rw-r--")
+})
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a listing of ADLS files, the latest timestamp of all the files transferred is stored. "
+ + "This allows the Processor to list only files that have been added or modified after "
+ + "this date the next time that the Processor is run, without having to store all of the actual filenames/paths which could lead to performance "
+ + "problems. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary "
+ + "Node is selected, the new node can pick up where the previous node left off, without duplicating the data.")
+public class ListADLSFile extends ADLSAbstractProcessor {
+
+ public static final PropertyDescriptor DISTRIBUTED_CACHE_SERVICE = new PropertyDescriptor.Builder()
+ .name("Distributed Cache Service")
+ .description("Specifies the Controller Service that should be used to maintain state about what has been pulled from ADLS so that if a new node "
+ + "begins pulling data, it won't duplicate all of the work that has been done.")
+ .required(false)
+ .identifiesControllerService(DistributedMapCacheClient.class)
+ .build();
+
+ public static final PropertyDescriptor RECURSE_SUBDIRS = new PropertyDescriptor.Builder()
+ .name("Recurse Subdirectories")
+ .description("Indicates whether to list files from subdirectories of the ADLS directory")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+
+ public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
+ .name("File Filter")
+ .description("Only files whose names match the given regular expression will be picked up")
+ .required(true)
+ .defaultValue("[^\\.].*")
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The ADLS directory from which files should be listed")
+ .required(true)
+ .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ // increase this factor from test case to test minimum duration between runs functionality
+ protected long testingMinimumDurationMultiplicationFactor = 1;
+
+ private volatile long latestTimestampListed = 0;
+ private volatile long lastRunTimestamp = 0;
+
+ protected static final String LISTING_TIMESTAMP_KEY = "adls.listing.timestamp";
+
+ protected static final long MINIMUM_DURATON_BW_RUNS_NANOS = TimeUnit.MILLISECONDS.toNanos(100L);
+
+ @Override
+ protected void init(ProcessorInitializationContext context) {
+ super.init(context);
+
+ super.descriptors.add(DISTRIBUTED_CACHE_SERVICE);
+ super.descriptors.add(RECURSE_SUBDIRS);
+ super.descriptors.add(FILE_FILTER);
+ super.descriptors.add(DIRECTORY);
+ }
+
+ @Override
+ public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) {
+ super.onPropertyModified(descriptor, oldValue, newValue);
+ if (isConfigurationRestored()
+ && (descriptor.equals(DIRECTORY)
+ || descriptor.equals(FILE_FILTER)
+ || descriptor.equals(ACCOUNT_NAME))) {
+ latestTimestampListed = 0;
+ }
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession) throws ProcessException {
+ final long now = System.nanoTime();
+ if (now - lastRunTimestamp < MINIMUM_DURATON_BW_RUNS_NANOS * testingMinimumDurationMultiplicationFactor) {
+ processContext.yield();
+ return;
+ }
+ lastRunTimestamp = now;
+
+ final String fileFilter = processContext.getProperty(FILE_FILTER).getValue();
+ final String directoryValue = processContext.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue();
+ final String directory = Paths.get(directoryValue).normalize().toString();
+ final Date latestTimestampListedDate;
+ try {
+ latestTimestampListed = getLatestTimestampListedFromState(processContext);
+ latestTimestampListedDate = new Date(latestTimestampListed);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve timestamp of last listing from Distributed Cache Service. Will not perform listing until this is accomplished.");
+ processContext.yield();
+ return;
+ }
+
+ ADLStoreClient adlStoreClient = getAdlStoreClient();
+ if(adlStoreClient == null) {
+ createADLSClient(processContext);
+ adlStoreClient = getAdlStoreClient();
+ }
+ final boolean recursive = processContext.getProperty(RECURSE_SUBDIRS).asBoolean();
+
+ List<DirectoryEntry> directoryEntries = null;
+ try {
+ if(directory == null || directory.isEmpty())
+ throw new IOException("The ADLS directory from which files should be listed is empty");
+ directoryEntries = getDirectoryListing(adlStoreClient, directory, recursive);
+ } catch (IOException e) {
+ getLogger().error("Failed to perform listing of ADLS due to {}", new Object[] {e});
+ return;
+ }
+
+ if(directoryEntries.size() == 0) {
+ getLogger().debug("There is no data to list. Yielding.");
+ processContext.yield();
+ return;
+ }
+
+ Optional<DirectoryEntry> optionalDirectoryEntry = directoryEntries
+ .stream()
+ .filter(directoryEntry ->
+ directoryEntry.type.equals(DirectoryEntryType.FILE)
+ )
+ .filter(directoryEntry ->
+ directoryEntry.lastModifiedTime.after(latestTimestampListedDate)
+ )
+ .filter(directoryEntry ->
+ directoryEntry.name.matches(fileFilter)
+ )
+ .map(directoryEntry -> {
+ createAndTransferFlowFile(processSession, directoryEntry);
+ return directoryEntry;
+ })
+ .max(Comparator.comparing(de -> de.lastModifiedTime));
+
+ Date currentRunLatestModifiedTimestamp;
+ try {
+ currentRunLatestModifiedTimestamp = optionalDirectoryEntry.get().lastModifiedTime;
+ } catch (NoSuchElementException e) {
+ getLogger().debug("There is no data to list. Yielding.");
+ processContext.yield();
+ return;
+ }
+
+ if(currentRunLatestModifiedTimestamp.after(latestTimestampListedDate)) {
+ processSession.commit();
+ getLogger().debug("Session commited");
+ try {
+ saveState(processContext, currentRunLatestModifiedTimestamp);
+ } catch (IOException e) {
+ getLogger().warn("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", e);
+ }
+ }
+ }
+
+ private void saveState(ProcessContext processContext, final Date listedTimestamp) throws IOException {
+ final Map<String, String> updatedState = new HashMap<>(1);
+ updatedState.put(LISTING_TIMESTAMP_KEY, String.valueOf(listedTimestamp.getTime()));
+ processContext.getStateManager().setState(updatedState, Scope.CLUSTER);
+ getLogger().debug("New state saved with listed time: {}", new Object[] {listedTimestamp.toString()});
+ }
+
+ private void createAndTransferFlowFile(ProcessSession processSession, DirectoryEntry directoryEntry) {
+ final Map<String, String> attributes = createAttributes(directoryEntry);
+ FlowFile flowFile = processSession.create();
+ flowFile = processSession.putAllAttributes(flowFile, attributes);
+ processSession.transfer(flowFile, REL_SUCCESS);
+ }
+
+ private Map<String, String> createAttributes(final DirectoryEntry directoryEntry) {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(CoreAttributes.FILENAME.key(), directoryEntry.name);
+ Path fullNamePath = Paths.get(directoryEntry.fullName);
+ fullNamePath.toAbsolutePath().getParent().toString();
+ attributes.put(CoreAttributes.PATH.key(), fullNamePath.getParent().toString());
+ attributes.put(CoreAttributes.ABSOLUTE_PATH.key(), fullNamePath.toAbsolutePath().getParent().toString());
+ attributes.put("adls.owner", directoryEntry.user);
+ attributes.put("adls.group", directoryEntry.group);
+ attributes.put("adls.lastModified", String.valueOf(directoryEntry.lastModifiedTime));
+ attributes.put("adls.length", String.valueOf(directoryEntry.length));
+ attributes.put("adls.permissions", directoryEntry.permission);
+ if(directoryEntry.expiryTime != null)
--- End diff --
You need brackets. Please confirm the build is passing with ``-Pcontrib-check`` profile. Right now, you have the following checkstyle issues:
````
[WARNING] src/main/java/org/apache/nifi/processors/adls/ADLSAbstractProcessor.java[35] (imports) AvoidStarImport: Using the '.*' form of import should be avoided - org.apache.nifi.processors.adls.ADLSConstants.*.
[WARNING] src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java[22] (imports) AvoidStarImport: Using the '.*' form of import should be avoided - org.apache.nifi.annotation.behavior.*.
[WARNING] src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java[40] (imports) AvoidStarImport: Using the '.*' form of import should be avoided - java.util.*.
[WARNING] src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java[23] (imports) AvoidStarImport: Using the '.*' form of import should be avoided - org.apache.nifi.annotation.behavior.*.
[WARNING] src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java[48] (imports) AvoidStarImport: Using the '.*' form of import should be avoided - org.apache.nifi.processors.adls.ADLSConstants.*.
[WARNING] src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java[223:9] (blocks) RightCurly: '}' should be on the same line.
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestFetchADLSFile.java[27:8] (imports) UnusedImports: Unused import - org.apache.nifi.util.MockFlowFile.
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestFetchADLSFile.java[34:8] (imports) UnusedImports: Unused import - java.io.File.
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestFetchADLSFile.java[35:8] (imports) UnusedImports: Unused import - java.io.FileInputStream.
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestFetchADLSFile.java[49] (sizes) LineLength: Line is longer than 200 characters (found 342).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestFetchADLSFile.java[53] (sizes) LineLength: Line is longer than 200 characters (found 337).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestFetchADLSFile.java[54] (sizes) LineLength: Line is longer than 200 characters (found 333).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestListADLSFile.java[57] (sizes) LineLength: Line is longer than 200 characters (found 910).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestListADLSFile.java[58] (sizes) LineLength: Line is longer than 200 characters (found 1,934).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestListADLSFile.java[59] (sizes) LineLength: Line is longer than 200 characters (found 936).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestListADLSFile.java[60] (sizes) LineLength: Line is longer than 200 characters (found 1,569).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestListADLSFile.java[61] (sizes) LineLength: Line is longer than 200 characters (found 922).
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestPutADLSFile.java[34:8] (imports) UnusedImports: Unused import - java.io.ByteArrayInputStream.
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestPutADLSFile.java[36:8] (imports) UnusedImports: Unused import - java.io.InputStream.
[WARNING] src/test/java/org/apache/nifi/processors/adls/TestPutADLSFile.java[37:8] (imports) UnusedImports: Unused import - java.nio.file.Paths.
````
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on the issue:
https://github.com/apache/nifi/pull/2158
@jzonthemtn @jzonthemtn @pvillard31 Please review.
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by mohitgargk <gi...@git.apache.org>.
Github user mohitgargk commented on the issue:
https://github.com/apache/nifi/pull/2158
Hi @milanchandna, was wondering if you are still working on this. I need this feature for my use case and was going the same path as you. Seems you are almost there. In case you are working on it, Can you please try syncing your fork and rebase your commits.
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on the issue:
https://github.com/apache/nifi/pull/2158
Removed the test resource files(as there were license issues) and resolved their dependencies. And there were environment specific characters in test cases. Got rid of them. Please check now.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r143404854
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java ---
@@ -0,0 +1,267 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.acl.AclEntry;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.*;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"hadoop", "ADLS", "put", "copy", "filesystem", "restricted"})
--- End diff --
done.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140447693
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java ---
@@ -0,0 +1,289 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.DirectoryEntry;
+import com.microsoft.azure.datalake.store.DirectoryEntryType;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ACCOUNT_NAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@TriggerSerially
+@TriggerWhenEmpty
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"hadoop", "ADLS", "get", "list", "ingest", "ingress", "source", "filesystem"})
--- End diff --
Could you add "azure"?
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140449021
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java ---
@@ -0,0 +1,267 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.acl.AclEntry;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.*;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"hadoop", "ADLS", "put", "copy", "filesystem", "restricted"})
+@CapabilityDescription("Write FlowFile data to Azure Data Lake Store (ADLS)")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the file written to ADLS comes from the value of this attribute."),
+ @ReadsAttribute(attribute = "filepath", description = "The relative path to the file on ADLS comes from the value of this attribute.")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = "filename", description = "The name of the file written to ADLS is stored in this attribute."),
+ @WritesAttribute(attribute = "absolute.adls.path", description = "The absolute path to the file on ADLS is stored in this attribute.")
+})
+@Restricted("Provides operator the ability to write to any file that NiFi has access to in ADLS.")
+public class PutADLSFile extends ADLSAbstractProcessor {
+
+ private static final String PART_FILE_EXTENSION = ".nifipart";
+ private static final String REPLACE_RESOLUTION = "replace";
+ private static final String FAIL_RESOLUTION = "fail";
+ private static final String APPEND_RESOLUTION = "append";
+
+ protected static final AllowableValue REPLACE_RESOLUTION_AV = new AllowableValue(REPLACE_RESOLUTION,
+ REPLACE_RESOLUTION, "Replaces the existing file if any.");
+ protected static final AllowableValue FAIL_RESOLUTION_AV = new AllowableValue(FAIL_RESOLUTION, FAIL_RESOLUTION,
+ "Penalizes the flow file and routes it to failure.");
+ protected static final AllowableValue APPEND_RESOLUTION_AV = new AllowableValue(APPEND_RESOLUTION, APPEND_RESOLUTION,
+ "Appends to the existing file if any, creates a new file otherwise.");
+
+ // properties
+ protected static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file " +
+ "with the same name already exists in the output directory")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION_AV.getValue())
+ .allowableValues(REPLACE_RESOLUTION_AV, FAIL_RESOLUTION_AV, APPEND_RESOLUTION_AV)
+ .build();
+
+ protected static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The directory where files will be written")
+ .required(true)
+ .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected static final PropertyDescriptor UMASK = new PropertyDescriptor.Builder()
--- End diff --
default value? required? validator? is it OK if the value is null?
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140503030
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ADLSConstants.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+public class ADLSConstants {
+
+ public static final int CHUNK_SIZE_IN_BYTES = 4000000;
+
+ public static final PropertyDescriptor PATH_NAME = new PropertyDescriptor.Builder()
+ .name("Path")
+ .description("Path for file in Azure Data Lake, e.g. /adlshome/")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor RETRY_ON_FAIL = new PropertyDescriptor.Builder()
+ .name("Overwrite policy")
+ .description("How many times to retry if read fails per chunk read, defaults to 3")
+ .required(true)
+ //TODO add Integer validator
+ .defaultValue("3")
+ .build();
+
+ public static final PropertyDescriptor ACCOUNT_NAME = new PropertyDescriptor.Builder()
+ .name("Account FQDN")
+ .description("Azure account fully qualified domain name eg: accountname.azuredatalakestore.net")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLIENT_ID = new PropertyDescriptor.Builder()
+ .name("Client ID")
+ .description("Azure client ID")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor CLIENT_SECRET = new PropertyDescriptor.Builder()
+ .name("Client secret")
+ .description("Azure client secret")
+ .required(true)
+ .sensitive(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor AUTH_TOKEN_ENDPOINT = new PropertyDescriptor.Builder()
+ .name("Auth token endpoint")
+ .description("Azure client secret")
--- End diff --
Removed it, wasn't being used.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140503615
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java ---
@@ -0,0 +1,267 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.acl.AclEntry;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.*;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"hadoop", "ADLS", "put", "copy", "filesystem", "restricted"})
+@CapabilityDescription("Write FlowFile data to Azure Data Lake Store (ADLS)")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the file written to ADLS comes from the value of this attribute."),
+ @ReadsAttribute(attribute = "filepath", description = "The relative path to the file on ADLS comes from the value of this attribute.")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = "filename", description = "The name of the file written to ADLS is stored in this attribute."),
+ @WritesAttribute(attribute = "absolute.adls.path", description = "The absolute path to the file on ADLS is stored in this attribute.")
+})
+@Restricted("Provides operator the ability to write to any file that NiFi has access to in ADLS.")
+public class PutADLSFile extends ADLSAbstractProcessor {
+
+ private static final String PART_FILE_EXTENSION = ".nifipart";
+ private static final String REPLACE_RESOLUTION = "replace";
+ private static final String FAIL_RESOLUTION = "fail";
+ private static final String APPEND_RESOLUTION = "append";
+
+ protected static final AllowableValue REPLACE_RESOLUTION_AV = new AllowableValue(REPLACE_RESOLUTION,
+ REPLACE_RESOLUTION, "Replaces the existing file if any.");
+ protected static final AllowableValue FAIL_RESOLUTION_AV = new AllowableValue(FAIL_RESOLUTION, FAIL_RESOLUTION,
+ "Penalizes the flow file and routes it to failure.");
+ protected static final AllowableValue APPEND_RESOLUTION_AV = new AllowableValue(APPEND_RESOLUTION, APPEND_RESOLUTION,
+ "Appends to the existing file if any, creates a new file otherwise.");
+
+ // properties
+ protected static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file " +
+ "with the same name already exists in the output directory")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION_AV.getValue())
+ .allowableValues(REPLACE_RESOLUTION_AV, FAIL_RESOLUTION_AV, APPEND_RESOLUTION_AV)
+ .build();
+
+ protected static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The directory where files will be written")
+ .required(true)
+ .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected static final PropertyDescriptor UMASK = new PropertyDescriptor.Builder()
--- End diff --
default value is decided from server if null is passed, custom validator already being used, null is OK.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140449340
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/adls/TestPutADLSFile.java ---
@@ -0,0 +1,522 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.ADLStoreOptions;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.QueueDispatcher;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+
+public class TestPutADLSFile {
+
+ private MockWebServer server;
+ private ADLStoreClient client;
+ private Processor processor;
+ private TestRunner runner;
+ Map<String, String> fileAttributes;
+
+ @Before
+ public void init() throws IOException, InitializationException {
+ server = new MockWebServer();
+ QueueDispatcher dispatcher = new QueueDispatcher();
+ dispatcher.setFailFast(new MockResponse().setResponseCode(400));
+ server.setDispatcher(dispatcher);
+ server.start();
+ String accountFQDN = server.getHostName() + ":" + server.getPort();
+ String dummyToken = "testDummyAadToken";
+
+ client = ADLStoreClient.createClient(accountFQDN, dummyToken);
+ client.setOptions(new ADLStoreOptions().setInsecureTransport());
+
+ processor = new PutADLSWithMockClient(client);
+
+ runner = TestRunners.newTestRunner(processor);
+
+ runner.setProperty(ADLSConstants.ACCOUNT_NAME, accountFQDN);
+ runner.setProperty(ADLSConstants.CLIENT_ID, "foobar");
+ runner.setProperty(ADLSConstants.CLIENT_SECRET, "foobar");
+ runner.setProperty(ADLSConstants.AUTH_TOKEN_ENDPOINT, "foobar");
+ runner.setProperty(PutADLSFile.DIRECTORY, "/sample/");
+ runner.setProperty(PutADLSFile.CONFLICT_RESOLUTION, PutADLSFile.FAIL_RESOLUTION_AV);
+ runner.removeProperty(PutADLSFile.UMASK);
+ runner.removeProperty(PutADLSFile.ACL);
+
+ fileAttributes = new HashMap<>();
+ fileAttributes.put("filename", "sample.txt");
+ fileAttributes.put("path", "/root/");
+ }
+
+ @Test
+ public void testPutConflictFail() throws IOException, InterruptedException {
+
+ //for create call
+ server.enqueue(new MockResponse().setResponseCode(200));
+ //for append call(internal call while writing to stream)
+ server.enqueue(new MockResponse().setResponseCode(200));
+ //for rename call(since its fail conflict)
+ server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"boolean\":true}"));
+ //for delete call(removing temp file)
+ server.enqueue(new MockResponse().setResponseCode(200));
+
+// String bodySimpleFileContent = new String(Files.readAllBytes(Paths.get(bodySimpleFileName)));
+ String bodySimpleFileContent = "sample content to be send to external adls resource";
+ runner.enqueue(bodySimpleFileContent, fileAttributes);
+
+ runner.run();
+
+ RecordedRequest createRequest = server.takeRequest(1, TimeUnit.SECONDS);
+ String queryOpCreateRequest = createRequest.getRequestUrl().queryParameter("op");
+ String pathCreateRequest = createRequest.getPath();
+ String queryOverwiteCreateRequest = createRequest.getRequestUrl().queryParameter("overwrite");
+ Assert.assertEquals("CREATE", queryOpCreateRequest);
+ Assert.assertEquals("true", queryOverwiteCreateRequest);
+ String expectedPath = "sample.txt.nifipart";
+ Assert.assertThat(pathCreateRequest, CoreMatchers.containsString(expectedPath));
+
+ RecordedRequest appendRequest = server.takeRequest(1, TimeUnit.SECONDS);
+ String pathAppendRequest = appendRequest.getPath();
+ String bodyAppendRequest = appendRequest.getBody().toString();
+ String queryOpAppendRequest = appendRequest.getRequestUrl().queryParameter("op");
+ Assert.assertEquals("APPEND", queryOpAppendRequest);
+ Assert.assertEquals("[text="+bodySimpleFileContent+"]", bodyAppendRequest);
+ Assert.assertThat(pathAppendRequest, CoreMatchers.containsString(expectedPath));
+
+ RecordedRequest renameRequest = server.takeRequest(1, TimeUnit.SECONDS);
+ String pathRenameRequest = renameRequest.getPath();
+ String queryOpRenameRequest = renameRequest.getRequestUrl().queryParameter("op");
+ String queryOverwriteRenameRequest = renameRequest.getRequestUrl().queryParameter("overwrite");
+ Assert.assertThat(pathRenameRequest, CoreMatchers.containsString(expectedPath));
+ Assert.assertEquals("RENAME", queryOpRenameRequest);
+ Assert.assertEquals(null, queryOverwriteRenameRequest);
+
+ RecordedRequest deleteRequest = server.takeRequest(1, TimeUnit.SECONDS);
+ String pathDeleteRequest = deleteRequest.getPath();
+ String queryOpDeleteRequest = deleteRequest.getRequestUrl().queryParameter("op");
+ Assert.assertEquals("DELETE", queryOpDeleteRequest);
+ Assert.assertThat(pathDeleteRequest, CoreMatchers.containsString(expectedPath));
+
+ runner.assertAllFlowFilesTransferred(ADLSConstants.REL_SUCCESS, 1);
+ }
+
+ @Test
+ public void testPutConflictReplace() throws IOException, InterruptedException {
--- End diff --
````
testPutConflictReplace(org.apache.nifi.processors.adls.TestPutADLSFile) Time elapsed: 0.031 sec <<< FAILURE!
java.lang.AssertionError: expected:<true> but was:<null>
at org.junit.Assert.fail(Assert.java:88)
at org.junit.Assert.failNotEquals(Assert.java:834)
at org.junit.Assert.assertEquals(Assert.java:118)
at org.junit.Assert.assertEquals(Assert.java:144)
at org.apache.nifi.processors.adls.TestPutADLSFile.testPutConflictReplace(TestPutADLSFile.java:176)
Results :
Failed tests:
TestPutADLSFile.testPutConflictReplace:176 expected:<true> but was:<null>
````
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on the issue:
https://github.com/apache/nifi/pull/2158
@milanchandna - you can go in ``nifi-nar-bundles/nifi-azure-bundle`` and build from this directory ``mvn clean install -Pcontrib-check``
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by jzonthemtn <gi...@git.apache.org>.
Github user jzonthemtn commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140665864
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java ---
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.acl.AclEntry;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.behavior.Restricted;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ERR_ACL_ENTRY;
+import static org.apache.nifi.processors.adls.ADLSConstants.ERR_FLOWFILE_CORE_ATTR_FILENAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_FAILURE;
+import static org.apache.nifi.processors.adls.ADLSConstants.FILE_NAME_ATTRIBUTE;
+import static org.apache.nifi.processors.adls.ADLSConstants.CHUNK_SIZE_IN_BYTES;
+import static org.apache.nifi.processors.adls.ADLSConstants.ADLS_FILE_PATH_ATTRIBUTE;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"azure", "hadoop", "ADLS", "put", "egress", "copy", "filesystem", "restricted"})
+@CapabilityDescription("Write FlowFile data to Azure Data Lake Store (ADLS)")
+@SeeAlso({ListADLSFile.class, FetchADLSFile.class})
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the file written to ADLS comes from the value of this attribute."),
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = "filename", description = "The name of the file written to ADLS is stored in this attribute."),
+ @WritesAttribute(attribute = "absolute.adls.path", description = "The absolute path to the file on ADLS is stored in this attribute.")
+})
+@Restricted("Provides operator the ability to write to any file that NiFi has access to in ADLS.")
+public class PutADLSFile extends ADLSAbstractProcessor {
+
+ private static final String PART_FILE_EXTENSION = ".nifipart";
+ private static final String REPLACE_RESOLUTION = "replace";
+ private static final String FAIL_RESOLUTION = "fail";
+ private static final String APPEND_RESOLUTION = "append";
+
+ protected static final AllowableValue REPLACE_RESOLUTION_AV = new AllowableValue(REPLACE_RESOLUTION,
+ REPLACE_RESOLUTION, "Replaces the existing file if any.");
+ protected static final AllowableValue FAIL_RESOLUTION_AV = new AllowableValue(FAIL_RESOLUTION, FAIL_RESOLUTION,
+ "Penalizes the flow file and routes it to failure.");
+ protected static final AllowableValue APPEND_RESOLUTION_AV = new AllowableValue(APPEND_RESOLUTION, APPEND_RESOLUTION,
+ "Appends to the existing file if any, creates a new file otherwise.");
+
+ // properties
+ protected static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("adls-put-conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file " +
+ "with the same name already exists in the output directory")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION_AV.getValue())
+ .allowableValues(REPLACE_RESOLUTION_AV, FAIL_RESOLUTION_AV, APPEND_RESOLUTION_AV)
+ .build();
+
+ protected static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("adls-put-directory")
+ .displayName("Directory")
+ .description("The directory where files will be written")
+ .required(true)
+ .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected static final PropertyDescriptor UMASK = new PropertyDescriptor.Builder()
+ .name("adls-put-permissions-umask")
+ .displayName("Permissions umask")
+ .description(
+ "A umask represented as an octal number which determines the permissions of files written to ADLS.")
+ .addValidator(ADLSValidators.UMASK_VALIDATOR)
+ .build();
+
+ protected static final PropertyDescriptor ACL = new PropertyDescriptor.Builder()
+ .name("adls-put-access-control-list")
+ .displayName("Access Control List")
+ .description(
+ "Comma separated list of ACL entry. An ACL entry consists of a scope (access or default)" +
+ " the type of the ACL (user, group, other or mask), the name of the user or group" +
+ " associated with this ACL (can be blank to specify the default permissions for" +
+ " users and groups, and must be blank for mask entries), and the action permitted" +
+ " by this ACL entry. eg: \"default:user:bob:r-x,default:user:adam:rw\"")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ @Override
+ protected void init(ProcessorInitializationContext context) {
+ super.init(context);
+
+ super.descriptors.add(CONFLICT_RESOLUTION);
+ super.descriptors.add(DIRECTORY);
+ super.descriptors.add(UMASK);
+ super.descriptors.add(ACL);
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ final FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final StopWatch dataRateStopWatch = new StopWatch();
+ final StopWatch completeProcessStopWatch = new StopWatch(true);
+
+ ADLStoreClient adlsClient = getAdlStoreClient();
+ //safe check
--- End diff --
This check for `null` is duplicated in a few places. Is it possible to move the check to `getAdlStoreClient()`, and if it is going to return `null`, then create the client (`createADLSClient()`) in order to get rid of the checks?
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r142037960
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/PutADLSFile.java ---
@@ -0,0 +1,282 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.acl.AclEntry;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.behavior.Restricted;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StopWatch;
+import org.apache.nifi.util.StringUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ERR_ACL_ENTRY;
+import static org.apache.nifi.processors.adls.ADLSConstants.ERR_FLOWFILE_CORE_ATTR_FILENAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_FAILURE;
+import static org.apache.nifi.processors.adls.ADLSConstants.FILE_NAME_ATTRIBUTE;
+import static org.apache.nifi.processors.adls.ADLSConstants.CHUNK_SIZE_IN_BYTES;
+import static org.apache.nifi.processors.adls.ADLSConstants.ADLS_FILE_PATH_ATTRIBUTE;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@Tags({"azure", "hadoop", "ADLS", "put", "egress", "copy", "filesystem", "restricted"})
+@CapabilityDescription("Write FlowFile data to Azure Data Lake Store (ADLS)")
+@SeeAlso({ListADLSFile.class, FetchADLSFile.class})
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the file written to ADLS comes from the value of this attribute."),
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = "filename", description = "The name of the file written to ADLS is stored in this attribute."),
+ @WritesAttribute(attribute = "absolute.adls.path", description = "The absolute path to the file on ADLS is stored in this attribute.")
+})
+@Restricted("Provides operator the ability to write to any file that NiFi has access to in ADLS.")
+public class PutADLSFile extends ADLSAbstractProcessor {
+
+ private static final String PART_FILE_EXTENSION = ".nifipart";
+ private static final String REPLACE_RESOLUTION = "replace";
+ private static final String FAIL_RESOLUTION = "fail";
+ private static final String APPEND_RESOLUTION = "append";
+
+ protected static final AllowableValue REPLACE_RESOLUTION_AV = new AllowableValue(REPLACE_RESOLUTION,
+ REPLACE_RESOLUTION, "Replaces the existing file if any.");
+ protected static final AllowableValue FAIL_RESOLUTION_AV = new AllowableValue(FAIL_RESOLUTION, FAIL_RESOLUTION,
+ "Penalizes the flow file and routes it to failure.");
+ protected static final AllowableValue APPEND_RESOLUTION_AV = new AllowableValue(APPEND_RESOLUTION, APPEND_RESOLUTION,
+ "Appends to the existing file if any, creates a new file otherwise.");
+
+ // properties
+ protected static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("adls-put-conflict-resolution-strategy")
+ .displayName("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file " +
+ "with the same name already exists in the output directory")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION_AV.getValue())
+ .allowableValues(REPLACE_RESOLUTION_AV, FAIL_RESOLUTION_AV, APPEND_RESOLUTION_AV)
+ .build();
+
+ protected static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("adls-put-directory")
+ .displayName("Directory")
+ .description("The directory where files will be written")
+ .required(true)
+ .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected static final PropertyDescriptor UMASK = new PropertyDescriptor.Builder()
+ .name("adls-put-permissions-umask")
+ .displayName("Permissions umask")
+ .description(
+ "A umask represented as an octal number which determines the permissions of files written to ADLS.")
+ .addValidator(ADLSValidators.UMASK_VALIDATOR)
+ .build();
+
+ protected static final PropertyDescriptor ACL = new PropertyDescriptor.Builder()
+ .name("adls-put-access-control-list")
+ .displayName("Access Control List")
+ .description(
+ "Comma separated list of ACL entry. An ACL entry consists of a scope (access or default)" +
+ " the type of the ACL (user, group, other or mask), the name of the user or group" +
+ " associated with this ACL (can be blank to specify the default permissions for" +
+ " users and groups, and must be blank for mask entries), and the action permitted" +
+ " by this ACL entry. eg: \"default:user:bob:r-x,default:user:adam:rw\"")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ @Override
+ protected void init(ProcessorInitializationContext context) {
+ super.init(context);
+
+ super.descriptors.add(CONFLICT_RESOLUTION);
+ super.descriptors.add(DIRECTORY);
+ super.descriptors.add(UMASK);
+ super.descriptors.add(ACL);
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ final FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final StopWatch dataRateStopWatch = new StopWatch();
+ final StopWatch completeProcessStopWatch = new StopWatch(true);
+
+ ADLStoreClient adlsClient = getAdlStoreClient();
+ //safe check
--- End diff --
@jzonthemtn Yes, it makes sense to push the lazy initialization to Getter, did that. Thanks for review.
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by pvillard31 <gi...@git.apache.org>.
Github user pvillard31 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r140446472
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ADLSConstants.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+public class ADLSConstants {
+
+ public static final int CHUNK_SIZE_IN_BYTES = 4000000;
+
+ public static final PropertyDescriptor PATH_NAME = new PropertyDescriptor.Builder()
+ .name("Path")
+ .description("Path for file in Azure Data Lake, e.g. /adlshome/")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor RETRY_ON_FAIL = new PropertyDescriptor.Builder()
+ .name("Overwrite policy")
+ .description("How many times to retry if read fails per chunk read, defaults to 3")
+ .required(true)
+ //TODO add Integer validator
--- End diff --
TODO?
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r143404758
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ADLSConstants.java ---
@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+
+
+public class ADLSConstants {
+
+ public static final int CHUNK_SIZE_IN_BYTES = 4000000;
+
+ public static final PropertyDescriptor PATH_NAME = new PropertyDescriptor.Builder()
+ .name("Path")
+ .description("Path for file in Azure Data Lake, e.g. /adlshome/")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor RETRY_ON_FAIL = new PropertyDescriptor.Builder()
+ .name("Overwrite policy")
+ .description("How many times to retry if read fails per chunk read, defaults to 3")
+ .required(true)
+ //TODO add Integer validator
--- End diff --
property wasn't being used anymore, removed it.
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by joewitt <gi...@git.apache.org>.
Github user joewitt commented on the issue:
https://github.com/apache/nifi/pull/2158
Lots of test failures.
/code
Tests run: 19, Failures: 3, Errors: 0, Skipped: 0, Time elapsed: 0.847 sec <<< FAILURE! - in org.apache.nifi.processors.adls.TestPutADLSFile
testPutConflictReplace(org.apache.nifi.processors.adls.TestPutADLSFile) Time elapsed: 0.102 sec <<< FAILURE!
java.lang.AssertionError:
Expected: a string containing "%5Csample%5Csample.txt.nifipart"
but: was "/webhdfs/v1/sample/sample.txt.nifipart?op=CREATE&syncFlag=DATA&overwrite=true&write=true&filesessionid=3834110d-e3b1-41cc-9ff4-022da542ae4b&leaseid=3834110d-e3b1-41cc-9ff4-022da542ae4b&api-version=2016-11-01"
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
at org.junit.Assert.assertThat(Assert.java:956)
at org.junit.Assert.assertThat(Assert.java:923)
at org.apache.nifi.processors.adls.TestPutADLSFile.testPutConflictReplace(TestPutADLSFile.java:161)
testPutConflictAppend(org.apache.nifi.processors.adls.TestPutADLSFile) Time elapsed: 0.01 sec <<< FAILURE!
java.lang.AssertionError:
Expected: a string containing "%5Csample%5Csample.txt.nifipart"
but: was "/webhdfs/v1/sample/sample.txt.nifipart?op=CREATE&syncFlag=DATA&overwrite=true&write=true&filesessionid=0c697d9a-fa07-47b7-adfd-877b90e15717&leaseid=0c697d9a-fa07-47b7-adfd-877b90e15717&api-version=2016-11-01"
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
at org.junit.Assert.assertThat(Assert.java:956)
at org.junit.Assert.assertThat(Assert.java:923)
at org.apache.nifi.processors.adls.TestPutADLSFile.testPutConflictAppend(TestPutADLSFile.java:215)
testPutConflictFail(org.apache.nifi.processors.adls.TestPutADLSFile) Time elapsed: 0 sec <<< FAILURE!
java.lang.AssertionError:
Expected: a string containing "%5Csample%5Csample.txt.nifipart"
but: was "/webhdfs/v1/sample/sample.txt.nifipart?op=CREATE&syncFlag=DATA&overwrite=true&write=true&filesessionid=a6fe439c-b2f1-41c2-ba5d-e7d75836d53b&leaseid=a6fe439c-b2f1-41c2-ba5d-e7d75836d53b&api-version=2016-11-01"
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
at org.junit.Assert.assertThat(Assert.java:956)
at org.junit.Assert.assertThat(Assert.java:923)
at org.apache.nifi.processors.adls.TestPutADLSFile.testPutConflictFail(TestPutADLSFile.java:108)
Tests run: 7, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.249 sec - in org.apache.nifi.ranger.authorization.TestRangerBasePluginWithPolicies
Running org.apache.nifi.ranger.authorization.TestRangerNiFiAuthorizer
Results :
Failed tests:
TestListADLSFile.testFlowFileAttributes:233 expected:<[\]test> but was:<[/]test>
TestPutADLSFile.testPutConflictAppend:215
Expected: a string containing "%5Csample%5Csample.txt.nifipart"
but: was "/webhdfs/v1/sample/sample.txt.nifipart?op=CREATE&syncFlag=DATA&overwrite=true&write=true&filesessionid=0c697d9a-fa07-47b7-adfd-877b90e15717&leaseid=0c697d9a-fa07-47b7-adfd-877b90e15717&api-version=2016-11-01"
TestPutADLSFile.testPutConflictFail:108
Expected: a string containing "%5Csample%5Csample.txt.nifipart"
but: was "/webhdfs/v1/sample/sample.txt.nifipart?op=CREATE&syncFlag=DATA&overwrite=true&write=true&filesessionid=a6fe439c-b2f1-41c2-ba5d-e7d75836d53b&leaseid=a6fe439c-b2f1-41c2-ba5d-e7d75836d53b&api-version=2016-11-01"
TestPutADLSFile.testPutConflictReplace:161
Expected: a string containing "%5Csample%5Csample.txt.nifipart"
but: was "/webhdfs/v1/sample/sample.txt.nifipart?op=CREATE&syncFlag=DATA&overwrite=true&write=true&filesessionid=3834110d-e3b1-41cc-9ff4-022da542ae4b&leaseid=3834110d-e3b1-41cc-9ff4-022da542ae4b&api-version=2016-11-01"
---
[GitHub] nifi issue #2158: NIFI-4360 Adding support for ADLS Processors. Feature incl...
Posted by joewitt <gi...@git.apache.org>.
Github user joewitt commented on the issue:
https://github.com/apache/nifi/pull/2158
@milanchandna am checking into the dependency tree and L&N. Looks good so far.
However, please get rid of the davinci and davinci4MB files. The origin of them is unclear and it looks like it comes from common ADLS test files. We have to cite them as source material in such cases. We're better off just making up our own original test files OR removing them altogether.
Thanks
---
[GitHub] nifi pull request #2158: NIFI-4360 Adding support for ADLS Processors. Featu...
Posted by milanchandna <gi...@git.apache.org>.
Github user milanchandna commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2158#discussion_r143404820
--- Diff: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/adls/ListADLSFile.java ---
@@ -0,0 +1,289 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.processors.adls;
+
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.DirectoryEntry;
+import com.microsoft.azure.datalake.store.DirectoryEntryType;
+import org.apache.nifi.annotation.behavior.*;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.nifi.processors.adls.ADLSConstants.ACCOUNT_NAME;
+import static org.apache.nifi.processors.adls.ADLSConstants.REL_SUCCESS;
+
+@TriggerSerially
+@TriggerWhenEmpty
+@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
+@Tags({"hadoop", "ADLS", "get", "list", "ingest", "ingress", "source", "filesystem"})
+@CapabilityDescription("Retrieves a listing of files from ADLS. For each file that is "
+ + "listed in ADLS, this processor creates a FlowFile that represents the ADLS file to be fetched in conjunction with FetchADLSFile. This Processor is "
+ + "designed to run on Primary Node only in a cluster. If the primary node changes, the new Primary Node will pick up where the previous node left "
+ + "off without duplicating all of the data.")
+@WritesAttributes({
+ @WritesAttribute(attribute="filename", description="The name of the file that was read from ADLS"),
+ @WritesAttribute(attribute="path", description="The path is set to the path of the file on ADLS"),
+ @WritesAttribute(attribute="adls.owner", description="The user that owns the file"),
+ @WritesAttribute(attribute="adls.group", description="The group that owns the file"),
+ @WritesAttribute(attribute="adls.lastModified", description="The timestamp of when the file was last modified, as milliseconds since midnight Jan 1, 1970 UTC"),
+ @WritesAttribute(attribute="adls.length", description="The number of bytes in the file"),
+ @WritesAttribute(attribute="adls.expirytime", description="The number of bytes in the file"),
+ @WritesAttribute(attribute="adls.permissions", description="The permissions for the file in ADLS. This is formatted as 3 characters for the owner, "
+ + "3 for the group, and 3 for other users. For example rw-rw-r--")
+})
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a listing of ADLS files, the latest timestamp of all the files transferred is stored. "
+ + "This allows the Processor to list only files that have been added or modified after "
+ + "this date the next time that the Processor is run, without having to store all of the actual filenames/paths which could lead to performance "
+ + "problems. State is stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary "
+ + "Node is selected, the new node can pick up where the previous node left off, without duplicating the data.")
+public class ListADLSFile extends ADLSAbstractProcessor {
+
+ public static final PropertyDescriptor DISTRIBUTED_CACHE_SERVICE = new PropertyDescriptor.Builder()
+ .name("Distributed Cache Service")
+ .description("Specifies the Controller Service that should be used to maintain state about what has been pulled from ADLS so that if a new node "
+ + "begins pulling data, it won't duplicate all of the work that has been done.")
+ .required(false)
+ .identifiesControllerService(DistributedMapCacheClient.class)
+ .build();
+
+ public static final PropertyDescriptor RECURSE_SUBDIRS = new PropertyDescriptor.Builder()
+ .name("Recurse Subdirectories")
+ .description("Indicates whether to list files from subdirectories of the ADLS directory")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
+
+ public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
+ .name("File Filter")
+ .description("Only files whose names match the given regular expression will be picked up")
+ .required(true)
+ .defaultValue("[^\\.].*")
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The ADLS directory from which files should be listed")
+ .required(true)
+ .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+
+ // increase this factor from test case to test minimum duration between runs functionality
+ protected long testingMinimumDurationMultiplicationFactor = 1;
+
+ private volatile long latestTimestampListed = 0;
+ private volatile long lastRunTimestamp = 0;
+
+ protected static final String LISTING_TIMESTAMP_KEY = "adls.listing.timestamp";
+
+ protected static final long MINIMUM_DURATON_BW_RUNS_NANOS = TimeUnit.MILLISECONDS.toNanos(100L);
+
+ @Override
+ protected void init(ProcessorInitializationContext context) {
+ super.init(context);
+
+ super.descriptors.add(DISTRIBUTED_CACHE_SERVICE);
+ super.descriptors.add(RECURSE_SUBDIRS);
+ super.descriptors.add(FILE_FILTER);
+ super.descriptors.add(DIRECTORY);
+ }
+
+ @Override
+ public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) {
+ super.onPropertyModified(descriptor, oldValue, newValue);
+ if (isConfigurationRestored()
+ && (descriptor.equals(DIRECTORY)
+ || descriptor.equals(FILE_FILTER)
+ || descriptor.equals(ACCOUNT_NAME))) {
+ latestTimestampListed = 0;
+ }
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession) throws ProcessException {
+ final long now = System.nanoTime();
+ if (now - lastRunTimestamp < MINIMUM_DURATON_BW_RUNS_NANOS * testingMinimumDurationMultiplicationFactor) {
+ processContext.yield();
+ return;
+ }
+ lastRunTimestamp = now;
+
+ final String fileFilter = processContext.getProperty(FILE_FILTER).getValue();
+ final String directoryValue = processContext.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue();
+ final String directory = Paths.get(directoryValue).normalize().toString();
+ final Date latestTimestampListedDate;
+ try {
+ latestTimestampListed = getLatestTimestampListedFromState(processContext);
+ latestTimestampListedDate = new Date(latestTimestampListed);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve timestamp of last listing from Distributed Cache Service. Will not perform listing until this is accomplished.");
+ processContext.yield();
+ return;
+ }
+
+ ADLStoreClient adlStoreClient = getAdlStoreClient();
+ if(adlStoreClient == null) {
+ createADLSClient(processContext);
+ adlStoreClient = getAdlStoreClient();
+ }
+ final boolean recursive = processContext.getProperty(RECURSE_SUBDIRS).asBoolean();
+
+ List<DirectoryEntry> directoryEntries = null;
+ try {
+ if(directory == null || directory.isEmpty())
+ throw new IOException("The ADLS directory from which files should be listed is empty");
+ directoryEntries = getDirectoryListing(adlStoreClient, directory, recursive);
+ } catch (IOException e) {
+ getLogger().error("Failed to perform listing of ADLS due to {}", new Object[] {e});
+ return;
+ }
+
+ if(directoryEntries.size() == 0) {
+ getLogger().debug("There is no data to list. Yielding.");
+ processContext.yield();
+ return;
+ }
+
+ Optional<DirectoryEntry> optionalDirectoryEntry = directoryEntries
+ .stream()
+ .filter(directoryEntry ->
+ directoryEntry.type.equals(DirectoryEntryType.FILE)
+ )
+ .filter(directoryEntry ->
+ directoryEntry.lastModifiedTime.after(latestTimestampListedDate)
+ )
+ .filter(directoryEntry ->
+ directoryEntry.name.matches(fileFilter)
+ )
+ .map(directoryEntry -> {
+ createAndTransferFlowFile(processSession, directoryEntry);
+ return directoryEntry;
+ })
+ .max(Comparator.comparing(de -> de.lastModifiedTime));
+
+ Date currentRunLatestModifiedTimestamp;
+ try {
+ currentRunLatestModifiedTimestamp = optionalDirectoryEntry.get().lastModifiedTime;
+ } catch (NoSuchElementException e) {
+ getLogger().debug("There is no data to list. Yielding.");
+ processContext.yield();
+ return;
+ }
+
+ if(currentRunLatestModifiedTimestamp.after(latestTimestampListedDate)) {
+ processSession.commit();
+ getLogger().debug("Session commited");
+ try {
+ saveState(processContext, currentRunLatestModifiedTimestamp);
+ } catch (IOException e) {
+ getLogger().warn("Failed to save cluster-wide state. If NiFi is restarted, data duplication may occur", e);
+ }
+ }
+ }
+
+ private void saveState(ProcessContext processContext, final Date listedTimestamp) throws IOException {
+ final Map<String, String> updatedState = new HashMap<>(1);
+ updatedState.put(LISTING_TIMESTAMP_KEY, String.valueOf(listedTimestamp.getTime()));
+ processContext.getStateManager().setState(updatedState, Scope.CLUSTER);
+ getLogger().debug("New state saved with listed time: {}", new Object[] {listedTimestamp.toString()});
+ }
+
+ private void createAndTransferFlowFile(ProcessSession processSession, DirectoryEntry directoryEntry) {
+ final Map<String, String> attributes = createAttributes(directoryEntry);
+ FlowFile flowFile = processSession.create();
+ flowFile = processSession.putAllAttributes(flowFile, attributes);
+ processSession.transfer(flowFile, REL_SUCCESS);
+ }
+
+ private Map<String, String> createAttributes(final DirectoryEntry directoryEntry) {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(CoreAttributes.FILENAME.key(), directoryEntry.name);
+ Path fullNamePath = Paths.get(directoryEntry.fullName);
+ fullNamePath.toAbsolutePath().getParent().toString();
+ attributes.put(CoreAttributes.PATH.key(), fullNamePath.getParent().toString());
+ attributes.put(CoreAttributes.ABSOLUTE_PATH.key(), fullNamePath.toAbsolutePath().getParent().toString());
+ attributes.put("adls.owner", directoryEntry.user);
+ attributes.put("adls.group", directoryEntry.group);
+ attributes.put("adls.lastModified", String.valueOf(directoryEntry.lastModifiedTime));
+ attributes.put("adls.length", String.valueOf(directoryEntry.length));
+ attributes.put("adls.permissions", directoryEntry.permission);
+ if(directoryEntry.expiryTime != null)
--- End diff --
done.
---