You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by su...@apache.org on 2016/11/11 22:43:11 UTC
[24/50] [abbrv] hadoop git commit: HADOOP-13037. Refactor Azure Data
Lake Store as an independent FileSystem. Contributed by Vishwajeet Dusane
HADOOP-13037. Refactor Azure Data Lake Store as an independent FileSystem. Contributed by Vishwajeet Dusane
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/5c61ad24
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/5c61ad24
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/5c61ad24
Branch: refs/heads/YARN-2915
Commit: 5c61ad24887f76dfc5a5935b2c5dceb6bfd99417
Parents: a9ad5d6
Author: Chris Douglas <cd...@apache.org>
Authored: Thu Nov 10 21:55:55 2016 -0800
Committer: Chris Douglas <cd...@apache.org>
Committed: Fri Nov 11 11:15:07 2016 -0800
----------------------------------------------------------------------
.../src/main/resources/core-default.xml | 48 +-
.../dev-support/findbugs-exclude.xml | 24 -
hadoop-tools/hadoop-azure-datalake/pom.xml | 47 +-
.../main/java/org/apache/hadoop/fs/adl/Adl.java | 4 +
.../org/apache/hadoop/fs/adl/AdlConfKeys.java | 92 ++
.../org/apache/hadoop/fs/adl/AdlFileSystem.java | 892 +++++++++++++-
.../apache/hadoop/fs/adl/AdlFsInputStream.java | 149 +++
.../apache/hadoop/fs/adl/AdlFsOutputStream.java | 82 ++
.../org/apache/hadoop/fs/adl/AdlPermission.java | 69 ++
.../hadoop/fs/adl/SdkTokenProviderAdapter.java | 41 +
.../apache/hadoop/fs/adl/TokenProviderType.java | 25 +
.../fs/adl/oauth2/AzureADTokenProvider.java | 70 ++
...hedRefreshTokenBasedAccessTokenProvider.java | 135 ---
.../hadoop/fs/adl/oauth2/package-info.java | 2 +-
.../org/apache/hadoop/fs/adl/package-info.java | 2 +-
.../org/apache/hadoop/hdfs/web/ADLConfKeys.java | 61 -
.../apache/hadoop/hdfs/web/BufferManager.java | 180 ---
.../web/PrivateAzureDataLakeFileSystem.java | 1116 ------------------
...ClientCredentialBasedAccesTokenProvider.java | 156 ---
...hedRefreshTokenBasedAccessTokenProvider.java | 37 -
.../hadoop/hdfs/web/oauth2/package-info.java | 24 -
.../apache/hadoop/hdfs/web/package-info.java | 25 -
.../hadoop/hdfs/web/resources/ADLFlush.java | 49 -
.../hdfs/web/resources/ADLGetOpParam.java | 96 --
.../hdfs/web/resources/ADLPostOpParam.java | 97 --
.../hdfs/web/resources/ADLPutOpParam.java | 94 --
.../hdfs/web/resources/ADLVersionInfo.java | 51 -
.../web/resources/AppendADLNoRedirectParam.java | 45 -
.../web/resources/CreateADLNoRedirectParam.java | 44 -
.../hadoop/hdfs/web/resources/LeaseParam.java | 53 -
.../web/resources/ReadADLNoRedirectParam.java | 44 -
.../hadoop/hdfs/web/resources/package-info.java | 27 -
.../META-INF/org.apache.hadoop.fs.FileSystem | 16 +
.../src/site/markdown/index.md | 126 +-
.../apache/hadoop/fs/adl/AdlMockWebServer.java | 99 ++
.../apache/hadoop/fs/adl/TestACLFeatures.java | 262 ++++
.../hadoop/fs/adl/TestADLResponseData.java | 67 +-
.../org/apache/hadoop/fs/adl/TestAdlRead.java | 196 +++
.../hadoop/fs/adl/TestAzureADTokenProvider.java | 133 +++
.../adl/TestConcurrentDataReadOperations.java | 299 +++++
.../hadoop/fs/adl/TestCustomTokenProvider.java | 136 +++
.../apache/hadoop/fs/adl/TestGetFileStatus.java | 33 +-
.../apache/hadoop/fs/adl/TestListStatus.java | 34 +-
.../fs/adl/TestRelativePathFormation.java | 61 +
.../fs/adl/TestValidateConfiguration.java | 103 ++
.../hadoop/fs/adl/TestableAdlFileSystem.java | 3 +-
.../fs/adl/common/CustomMockTokenProvider.java | 61 +
.../hadoop/fs/adl/common/ExpectedResponse.java | 71 ++
.../hadoop/fs/adl/common/Parallelized.java | 60 +
.../hadoop/fs/adl/common/TestDataForRead.java | 122 ++
.../fs/adl/live/AdlStorageConfiguration.java | 50 +-
.../hadoop/fs/adl/live/AdlStorageContract.java | 19 +-
.../live/TestAdlDifferentSizeWritesLive.java | 2 +-
.../adl/live/TestAdlFileSystemContractLive.java | 2 +-
.../hadoop/fs/adl/live/TestAdlReadLive.java | 342 ------
...estAdlWebHdfsFileContextCreateMkdirLive.java | 79 --
...AdlWebHdfsFileContextMainOperationsLive.java | 104 --
...hedRefreshTokenBasedAccessTokenProvider.java | 149 ---
.../hadoop/fs/common/AdlMockWebServer.java | 116 --
.../hadoop/fs/common/ExpectedResponse.java | 72 --
.../hadoop/fs/common/TestDataForRead.java | 120 --
.../org/apache/hadoop/hdfs/web/TestAdlRead.java | 205 ----
.../web/TestConcurrentDataReadOperations.java | 306 -----
.../hdfs/web/TestConfigurationSetting.java | 138 ---
.../hdfs/web/TestSplitSizeCalculation.java | 123 --
.../src/test/resources/adls.xml | 11 +-
.../test/resources/contract-test-options.xml | 62 +-
.../src/test/resources/log4j.properties | 30 +
68 files changed, 3277 insertions(+), 4416 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index 327acfa..05004c1 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -2382,53 +2382,6 @@
<!-- Azure Data Lake File System Configurations -->
-
- <property>
- <name>adl.feature.override.readahead</name>
- <value>true</value>
- <description>
- Enables read aheads in the ADL client, the feature is used to
- improve read throughput.
- This works in conjunction with the value set in
- adl.feature.override.readahead.max.buffersize.
- When set to false the read ahead feature is turned off.
- Default : True if not configured.
- </description>
- </property>
-
- <property>
- <name>adl.feature.override.readahead.max.buffersize</name>
- <value>8388608</value>
- <description>
- Define maximum buffer size to cache read ahead data, this is
- allocated per process to
- cache read ahead data. Applicable only when
- adl.feature.override.readahead is set to true.
- Default : 8388608 Byte i.e. 8MB if not configured.
- </description>
- </property>
-
- <property>
- <name>adl.feature.override.readahead.max.concurrent.connection</name>
- <value>2</value>
- <description>
- Define maximum concurrent connection can be established to
- read ahead. If the data size is less than 4MB then only 1 read n/w
- connection
- is set. If the data size is less than 4MB but less than 8MB then 2 read
- n/w connection
- is set. Data greater than 8MB then value set under the property would
- take
- effect. Applicable only when adl.feature.override.readahead is set
- to true and buffer size is greater than 8MB.
- It is recommended to reset this property if the
- adl.feature.override.readahead.max.buffersize
- is less than 8MB to gain performance. Application has to consider
- throttling limit for the account as well before configuring large
- buffer size.
- </description>
- </property>
-
<property>
<name>fs.adl.impl</name>
<value>org.apache.hadoop.fs.adl.AdlFileSystem</value>
@@ -2438,6 +2391,7 @@
<name>fs.AbstractFileSystem.adl.impl</name>
<value>org.apache.hadoop.fs.adl.Adl</value>
</property>
+ <!-- Azure Data Lake File System Configurations Ends Here-->
<property>
<name>hadoop.caller.context.enabled</name>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/dev-support/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/dev-support/findbugs-exclude.xml b/hadoop-tools/hadoop-azure-datalake/dev-support/findbugs-exclude.xml
deleted file mode 100644
index 4fd36ef..0000000
--- a/hadoop-tools/hadoop-azure-datalake/dev-support/findbugs-exclude.xml
+++ /dev/null
@@ -1,24 +0,0 @@
-<!--
- 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.
--->
-<FindBugsFilter>
- <!-- Buffer object is accessed withing trusted code and intentionally assigned instead of array copy -->
- <Match>
- <Class name="org.apache.hadoop.hdfs.web.PrivateAzureDataLakeFileSystem$BatchAppendOutputStream$CommitTask"/>
- <Bug pattern="EI_EXPOSE_REP2"/>
- <Priority value="2"/>
- </Match>
-</FindBugsFilter>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/pom.xml b/hadoop-tools/hadoop-azure-datalake/pom.xml
index 9a15b04..eba0979 100644
--- a/hadoop-tools/hadoop-azure-datalake/pom.xml
+++ b/hadoop-tools/hadoop-azure-datalake/pom.xml
@@ -35,22 +35,17 @@
<file.encoding>UTF-8</file.encoding>
<downloadSources>true</downloadSources>
</properties>
-
+ <repositories>
+ <repository>
+ <id>snapshots-repo</id>
+ <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+ <releases><enabled>false</enabled></releases>
+ <snapshots><enabled>true</enabled></snapshots>
+ </repository>
+ </repositories>
<build>
<plugins>
<plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>findbugs-maven-plugin</artifactId>
- <configuration>
- <findbugsXmlOutput>true</findbugsXmlOutput>
- <xmlOutput>true</xmlOutput>
- <excludeFilterFile>
- ${basedir}/dev-support/findbugs-exclude.xml
- </excludeFilterFile>
- <effort>Max</effort>
- </configuration>
- </plugin>
- <plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-project-info-reports-plugin</artifactId>
@@ -130,20 +125,13 @@
</build>
<dependencies>
+ <!-- SDK Dependency -->
<dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-client</artifactId>
- <exclusions>
- <exclusion>
- <artifactId>javax.servlet-api</artifactId>
- <groupId>javax.servlet</groupId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-hdfs-client</artifactId>
+ <groupId>com.microsoft.azure</groupId>
+ <artifactId>azure-data-lake-store-sdk</artifactId>
+ <version>2.0.4-SNAPSHOT</version>
</dependency>
+ <!-- ENDS HERE-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
@@ -159,11 +147,6 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-all</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
<groupId>com.eclipsesource.minimal-json</groupId>
<artifactId>minimal-json</artifactId>
<version>0.9.1</version>
@@ -181,9 +164,5 @@
<version>2.4.0</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/Adl.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/Adl.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/Adl.java
index 4642d6b..7ec04cf 100644
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/Adl.java
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/Adl.java
@@ -19,6 +19,8 @@
package org.apache.hadoop.fs.adl;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.DelegateToFileSystem;
@@ -29,6 +31,8 @@ import java.net.URISyntaxException;
/**
* Expose adl:// scheme to access ADL file system.
*/
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
public class Adl extends DelegateToFileSystem {
Adl(URI theUri, Configuration conf) throws IOException, URISyntaxException {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlConfKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlConfKeys.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlConfKeys.java
new file mode 100644
index 0000000..21120df
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlConfKeys.java
@@ -0,0 +1,92 @@
+/*
+ * 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.hadoop.fs.adl;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Constants.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public final class AdlConfKeys {
+ // OAuth2 Common Configuration
+ public static final String AZURE_AD_REFRESH_URL_KEY = "dfs.adls.oauth2"
+ + ".refresh.url";
+
+ // optional when provider type is refresh or client id.
+ public static final String AZURE_AD_TOKEN_PROVIDER_CLASS_KEY =
+ "dfs.adls.oauth2.access.token.provider";
+ public static final String AZURE_AD_CLIENT_ID_KEY =
+ "dfs.adls.oauth2.client.id";
+ public static final String AZURE_AD_TOKEN_PROVIDER_TYPE_KEY =
+ "dfs.adls.oauth2.access.token.provider.type";
+
+ // OAuth Refresh Token Configuration
+ public static final String AZURE_AD_REFRESH_TOKEN_KEY =
+ "dfs.adls.oauth2.refresh.token";
+
+ public static final String TOKEN_PROVIDER_TYPE_REFRESH_TOKEN = "RefreshToken";
+ // OAuth Client Cred Token Configuration
+ public static final String AZURE_AD_CLIENT_SECRET_KEY =
+ "dfs.adls.oauth2.credential";
+ public static final String TOKEN_PROVIDER_TYPE_CLIENT_CRED =
+ "ClientCredential";
+
+ public static final String READ_AHEAD_BUFFER_SIZE_KEY =
+ "adl.feature.client.cache.readahead";
+
+ public static final String WRITE_BUFFER_SIZE_KEY =
+ "adl.feature.client.cache.drop.behind.writes";
+ static final String SECURE_TRANSPORT_SCHEME = "https";
+ static final String INSECURE_TRANSPORT_SCHEME = "http";
+ static final String ADL_DEBUG_OVERRIDE_LOCAL_USER_AS_OWNER =
+ "adl.debug.override.localuserasfileowner";
+
+ static final boolean ADL_DEBUG_SET_LOCAL_USER_AS_OWNER_DEFAULT = false;
+ static final long ADL_BLOCK_SIZE = 256 * 1024 * 1024;
+ static final int ADL_REPLICATION_FACTOR = 1;
+ static final String ADL_HADOOP_CLIENT_NAME = "hadoop-azure-datalake-";
+ static final String ADL_HADOOP_CLIENT_VERSION =
+ "2.0.0-SNAPSHOT";
+ static final String ADL_EVENTS_TRACKING_SOURCE = "adl.events.tracking.source";
+ static final String ADL_EVENTS_TRACKING_CLUSTERNAME =
+ "adl.events.tracking.clustername";
+
+ static final String ADL_EVENTS_TRACKING_CLUSTERTYPE =
+ "adl.events.tracking.clustertype";
+ static final int DEFAULT_READ_AHEAD_BUFFER_SIZE = 4 * 1024 * 1024;
+ static final int DEFAULT_WRITE_AHEAD_BUFFER_SIZE = 4 * 1024 * 1024;
+
+ static final String LATENCY_TRACKER_KEY =
+ "adl.dfs.enable.client.latency.tracker";
+ static final boolean LATENCY_TRACKER_DEFAULT = true;
+
+ static final String ADL_EXPERIMENT_POSITIONAL_READ_KEY =
+ "adl.feature.experiment.positional.read.enable";
+ static final boolean ADL_EXPERIMENT_POSITIONAL_READ_DEFAULT = true;
+
+ static final String ADL_SUPPORT_ACL_BIT_IN_FSPERMISSION =
+ "adl.feature.support.acl.bit";
+ static final boolean ADL_SUPPORT_ACL_BIT_IN_FSPERMISSION_DEFAULT = true;
+
+ private AdlConfKeys() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFileSystem.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFileSystem.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFileSystem.java
index 11e1e0b..9083afc 100644
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFileSystem.java
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFileSystem.java
@@ -19,23 +19,905 @@
package org.apache.hadoop.fs.adl;
-import org.apache.hadoop.hdfs.web.PrivateAzureDataLakeFileSystem;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.microsoft.azure.datalake.store.ADLStoreClient;
+import com.microsoft.azure.datalake.store.ADLStoreOptions;
+import com.microsoft.azure.datalake.store.DirectoryEntry;
+import com.microsoft.azure.datalake.store.DirectoryEntryType;
+import com.microsoft.azure.datalake.store.IfExists;
+import com.microsoft.azure.datalake.store.LatencyTracker;
+import com.microsoft.azure.datalake.store.oauth2.AccessTokenProvider;
+import com.microsoft.azure.datalake.store.oauth2.ClientCredsTokenProvider;
+import com.microsoft.azure.datalake.store.oauth2.RefreshTokenBasedTokenProvider;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.ContentSummary.Builder;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.InvalidPathException;
+import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.Options.Rename;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.adl.oauth2.AzureADTokenProvider;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclStatus;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.Progressable;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.VersionInfo;
+import static org.apache.hadoop.fs.adl.AdlConfKeys.*;
/**
- * Expose adl:// scheme to access ADL file system.
+ * A FileSystem to access Azure Data Lake Store.
*/
-public class AdlFileSystem extends PrivateAzureDataLakeFileSystem {
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class AdlFileSystem extends FileSystem {
+ static final String SCHEME = "adl";
+ static final int DEFAULT_PORT = 443;
+ private URI uri;
+ private String userName;
+ private boolean overrideOwner;
+ private ADLStoreClient adlClient;
+ private Path workingDirectory;
+ private boolean aclBitStatus;
- public static final String SCHEME = "adl";
- public static final int DEFAULT_PORT = 443;
+ // retained for tests
+ private AccessTokenProvider tokenProvider;
+ private AzureADTokenProvider azureTokenProvider;
@Override
public String getScheme() {
return SCHEME;
}
+ public URI getUri() {
+ return uri;
+ }
+
@Override
public int getDefaultPort() {
return DEFAULT_PORT;
}
+
+ @Override
+ public boolean supportsSymlinks() {
+ return false;
+ }
+
+ /**
+ * Called after a new FileSystem instance is constructed.
+ *
+ * @param storeUri a uri whose authority section names the host, port, etc.
+ * for this FileSystem
+ * @param conf the configuration
+ */
+ @Override
+ public void initialize(URI storeUri, Configuration conf) throws IOException {
+ super.initialize(storeUri, conf);
+ this.setConf(conf);
+ this.uri = URI
+ .create(storeUri.getScheme() + "://" + storeUri.getAuthority());
+
+ try {
+ userName = UserGroupInformation.getCurrentUser().getShortUserName();
+ } catch (IOException e) {
+ userName = "hadoop";
+ }
+
+ this.setWorkingDirectory(getHomeDirectory());
+
+ overrideOwner = getConf().getBoolean(ADL_DEBUG_OVERRIDE_LOCAL_USER_AS_OWNER,
+ ADL_DEBUG_SET_LOCAL_USER_AS_OWNER_DEFAULT);
+
+ aclBitStatus = conf.getBoolean(ADL_SUPPORT_ACL_BIT_IN_FSPERMISSION,
+ ADL_SUPPORT_ACL_BIT_IN_FSPERMISSION_DEFAULT);
+
+ String accountFQDN = null;
+ String mountPoint = null;
+ String hostname = storeUri.getHost();
+ if (!hostname.contains(".") && !hostname.equalsIgnoreCase(
+ "localhost")) { // this is a symbolic name. Resolve it.
+ String hostNameProperty = "dfs.adls." + hostname + ".hostname";
+ String mountPointProperty = "dfs.adls." + hostname + ".mountpoint";
+ accountFQDN = getNonEmptyVal(conf, hostNameProperty);
+ mountPoint = getNonEmptyVal(conf, mountPointProperty);
+ } else {
+ accountFQDN = hostname;
+ }
+
+ if (storeUri.getPort() > 0) {
+ accountFQDN = accountFQDN + ":" + storeUri.getPort();
+ }
+
+ adlClient = ADLStoreClient
+ .createClient(accountFQDN, getAccessTokenProvider(conf));
+
+ ADLStoreOptions options = new ADLStoreOptions();
+ options.enableThrowingRemoteExceptions();
+
+ if (getTransportScheme().equalsIgnoreCase(INSECURE_TRANSPORT_SCHEME)) {
+ options.setInsecureTransport();
+ }
+
+ if (mountPoint != null) {
+ options.setFilePathPrefix(mountPoint);
+ }
+
+ String clusterName = conf.get(ADL_EVENTS_TRACKING_CLUSTERNAME, "UNKNOWN");
+ String clusterType = conf.get(ADL_EVENTS_TRACKING_CLUSTERTYPE, "UNKNOWN");
+
+ String clientVersion = ADL_HADOOP_CLIENT_NAME + (StringUtils
+ .isEmpty(VersionInfo.getVersion().trim()) ?
+ ADL_HADOOP_CLIENT_VERSION.trim() :
+ VersionInfo.getVersion().trim());
+ options.setUserAgentSuffix(clientVersion + "/" +
+ VersionInfo.getVersion().trim() + "/" + clusterName + "/"
+ + clusterType);
+
+ adlClient.setOptions(options);
+
+ boolean trackLatency = conf
+ .getBoolean(LATENCY_TRACKER_KEY, LATENCY_TRACKER_DEFAULT);
+ if (!trackLatency) {
+ LatencyTracker.disable();
+ }
+ }
+
+ /**
+ * This method is provided for convenience for derived classes to define
+ * custom {@link AzureADTokenProvider} instance.
+ *
+ * In order to ensure secure hadoop infrastructure and user context for which
+ * respective {@link AdlFileSystem} instance is initialized,
+ * Loading {@link AzureADTokenProvider} is not sufficient.
+ *
+ * The order of loading {@link AzureADTokenProvider} is to first invoke
+ * {@link #getCustomAccessTokenProvider(Configuration)}, If method return null
+ * which means no implementation provided by derived classes, then
+ * configuration object is loaded to retrieve token configuration as specified
+ * is documentation.
+ *
+ * Custom token management takes the higher precedence during initialization.
+ *
+ * @param conf Configuration object
+ * @return null if the no custom {@link AzureADTokenProvider} token management
+ * is specified.
+ * @throws IOException if failed to initialize token provider.
+ */
+ protected synchronized AzureADTokenProvider getCustomAccessTokenProvider(
+ Configuration conf) throws IOException {
+ String className = getNonEmptyVal(conf, AZURE_AD_TOKEN_PROVIDER_CLASS_KEY);
+
+ Class<? extends AzureADTokenProvider> azureADTokenProviderClass =
+ conf.getClass(AZURE_AD_TOKEN_PROVIDER_CLASS_KEY, null,
+ AzureADTokenProvider.class);
+ if (azureADTokenProviderClass == null) {
+ throw new IllegalArgumentException(
+ "Configuration " + className + " " + "not defined/accessible.");
+ }
+
+ azureTokenProvider = ReflectionUtils
+ .newInstance(azureADTokenProviderClass, conf);
+ if (azureTokenProvider == null) {
+ throw new IllegalArgumentException("Failed to initialize " + className);
+ }
+
+ azureTokenProvider.initialize(conf);
+ return azureTokenProvider;
+ }
+
+ private AccessTokenProvider getAccessTokenProvider(Configuration conf)
+ throws IOException {
+ TokenProviderType type = conf.getEnum(
+ AdlConfKeys.AZURE_AD_TOKEN_PROVIDER_TYPE_KEY, TokenProviderType.Custom);
+
+ switch (type) {
+ case RefreshToken:
+ tokenProvider = getConfRefreshTokenBasedTokenProvider(conf);
+ break;
+ case ClientCredential:
+ tokenProvider = getConfCredentialBasedTokenProvider(conf);
+ break;
+ case Custom:
+ default:
+ AzureADTokenProvider azureADTokenProvider = getCustomAccessTokenProvider(
+ conf);
+ tokenProvider = new SdkTokenProviderAdapter(azureADTokenProvider);
+ break;
+ }
+
+ return tokenProvider;
+ }
+
+ private AccessTokenProvider getConfCredentialBasedTokenProvider(
+ Configuration conf) {
+ String clientId = getNonEmptyVal(conf, AZURE_AD_CLIENT_ID_KEY);
+ String refreshUrl = getNonEmptyVal(conf, AZURE_AD_REFRESH_URL_KEY);
+ String clientSecret = getNonEmptyVal(conf, AZURE_AD_CLIENT_SECRET_KEY);
+ return new ClientCredsTokenProvider(refreshUrl, clientId, clientSecret);
+ }
+
+ private AccessTokenProvider getConfRefreshTokenBasedTokenProvider(
+ Configuration conf) {
+ String clientId = getNonEmptyVal(conf, AZURE_AD_CLIENT_ID_KEY);
+ String refreshToken = getNonEmptyVal(conf, AZURE_AD_REFRESH_TOKEN_KEY);
+ return new RefreshTokenBasedTokenProvider(clientId, refreshToken);
+ }
+
+ @VisibleForTesting
+ AccessTokenProvider getTokenProvider() {
+ return tokenProvider;
+ }
+
+ @VisibleForTesting
+ AzureADTokenProvider getAzureTokenProvider() {
+ return azureTokenProvider;
+ }
+
+ /**
+ * Constructing home directory locally is fine as long as Hadoop
+ * local user name and ADL user name relationship story is not fully baked
+ * yet.
+ *
+ * @return Hadoop local user home directory.
+ */
+ @Override
+ public Path getHomeDirectory() {
+ return makeQualified(new Path("/user/" + userName));
+ }
+
+ /**
+ * Create call semantic is handled differently in case of ADL. Create
+ * semantics is translated to Create/Append
+ * semantics.
+ * 1. No dedicated connection to server.
+ * 2. Buffering is locally done, Once buffer is full or flush is invoked on
+ * the by the caller. All the pending
+ * data is pushed to ADL as APPEND operation code.
+ * 3. On close - Additional call is send to server to close the stream, and
+ * release lock from the stream.
+ *
+ * Necessity of Create/Append semantics is
+ * 1. ADL backend server does not allow idle connection for longer duration
+ * . In case of slow writer scenario,
+ * observed connection timeout/Connection reset causing occasional job
+ * failures.
+ * 2. Performance boost to jobs which are slow writer, avoided network latency
+ * 3. ADL equally better performing with multiple of 4MB chunk as append
+ * calls.
+ *
+ * @param f File path
+ * @param permission Access permission for the newly created file
+ * @param overwrite Remove existing file and recreate new one if true
+ * otherwise throw error if file exist
+ * @param bufferSize Buffer size, ADL backend does not honour
+ * @param replication Replication count, ADL backend does not honour
+ * @param blockSize Block size, ADL backend does not honour
+ * @param progress Progress indicator
+ * @return FSDataOutputStream OutputStream on which application can push
+ * stream of bytes
+ * @throws IOException when system error, internal server error or user error
+ */
+ @Override
+ public FSDataOutputStream create(Path f, FsPermission permission,
+ boolean overwrite, int bufferSize, short replication, long blockSize,
+ Progressable progress) throws IOException {
+ statistics.incrementWriteOps(1);
+ IfExists overwriteRule = overwrite ? IfExists.OVERWRITE : IfExists.FAIL;
+ return new FSDataOutputStream(new AdlFsOutputStream(adlClient
+ .createFile(toRelativeFilePath(f), overwriteRule,
+ Integer.toOctalString(applyUMask(permission).toShort()), true),
+ getConf()), this.statistics);
+ }
+
+ /**
+ * Opens an FSDataOutputStream at the indicated Path with write-progress
+ * reporting. Same as create(), except fails if parent directory doesn't
+ * already exist.
+ *
+ * @param f the file name to open
+ * @param permission Access permission for the newly created file
+ * @param flags {@link CreateFlag}s to use for this stream.
+ * @param bufferSize the size of the buffer to be used. ADL backend does
+ * not honour
+ * @param replication required block replication for the file. ADL backend
+ * does not honour
+ * @param blockSize Block size, ADL backend does not honour
+ * @param progress Progress indicator
+ * @throws IOException when system error, internal server error or user error
+ * @see #setPermission(Path, FsPermission)
+ * @deprecated API only for 0.20-append
+ */
+ @Deprecated
+ @Override
+ public FSDataOutputStream createNonRecursive(Path f, FsPermission permission,
+ EnumSet<CreateFlag> flags, int bufferSize, short replication,
+ long blockSize, Progressable progress) throws IOException {
+ statistics.incrementWriteOps(1);
+ IfExists overwriteRule = IfExists.FAIL;
+ for (CreateFlag flag : flags) {
+ if (flag == CreateFlag.OVERWRITE) {
+ overwriteRule = IfExists.OVERWRITE;
+ break;
+ }
+ }
+
+ return new FSDataOutputStream(new AdlFsOutputStream(adlClient
+ .createFile(toRelativeFilePath(f), overwriteRule,
+ Integer.toOctalString(applyUMask(permission).toShort()), false),
+ getConf()), this.statistics);
+ }
+
+ /**
+ * Append to an existing file (optional operation).
+ *
+ * @param f the existing file to be appended.
+ * @param bufferSize the size of the buffer to be used. ADL backend does
+ * not honour
+ * @param progress Progress indicator
+ * @throws IOException when system error, internal server error or user error
+ */
+ @Override
+ public FSDataOutputStream append(Path f, int bufferSize,
+ Progressable progress) throws IOException {
+ statistics.incrementWriteOps(1);
+ return new FSDataOutputStream(
+ new AdlFsOutputStream(adlClient.getAppendStream(toRelativeFilePath(f)),
+ getConf()), this.statistics);
+ }
+
+ /**
+ * Azure data lake does not support user configuration for data replication
+ * hence not leaving system to query on
+ * azure data lake.
+ *
+ * Stub implementation
+ *
+ * @param p Not honoured
+ * @param replication Not honoured
+ * @return True hard coded since ADL file system does not support
+ * replication configuration
+ * @throws IOException No exception would not thrown in this case however
+ * aligning with parent api definition.
+ */
+ @Override
+ public boolean setReplication(final Path p, final short replication)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ return true;
+ }
+
+ /**
+ * Open call semantic is handled differently in case of ADL. Instead of
+ * network stream is returned to the user,
+ * Overridden FsInputStream is returned.
+ *
+ * @param f File path
+ * @param buffersize Buffer size, Not honoured
+ * @return FSDataInputStream InputStream on which application can read
+ * stream of bytes
+ * @throws IOException when system error, internal server error or user error
+ */
+ @Override
+ public FSDataInputStream open(final Path f, final int buffersize)
+ throws IOException {
+ statistics.incrementReadOps(1);
+ return new FSDataInputStream(
+ new AdlFsInputStream(adlClient.getReadStream(toRelativeFilePath(f)),
+ statistics, getConf()));
+ }
+
+ /**
+ * Return a file status object that represents the path.
+ *
+ * @param f The path we want information from
+ * @return a FileStatus object
+ * @throws IOException when the path does not exist or any other error;
+ * IOException see specific implementation
+ */
+ @Override
+ public FileStatus getFileStatus(final Path f) throws IOException {
+ statistics.incrementReadOps(1);
+ DirectoryEntry entry = adlClient.getDirectoryEntry(toRelativeFilePath(f));
+ return toFileStatus(entry, f);
+ }
+
+ /**
+ * List the statuses of the files/directories in the given path if the path is
+ * a directory.
+ *
+ * @param f given path
+ * @return the statuses of the files/directories in the given patch
+ * @throws IOException when the path does not exist or any other error;
+ * IOException see specific implementation
+ */
+ @Override
+ public FileStatus[] listStatus(final Path f) throws IOException {
+ statistics.incrementReadOps(1);
+ List<DirectoryEntry> entries =
+ adlClient.enumerateDirectory(toRelativeFilePath(f));
+ return toFileStatuses(entries, f);
+ }
+
+ /**
+ * Renames Path src to Path dst. Can take place on local fs
+ * or remote DFS.
+ *
+ * ADLS support POSIX standard for rename operation.
+ *
+ * @param src path to be renamed
+ * @param dst new path after rename
+ * @return true if rename is successful
+ * @throws IOException on failure
+ */
+ @Override
+ public boolean rename(final Path src, final Path dst) throws IOException {
+ statistics.incrementWriteOps(1);
+ return adlClient.rename(toRelativeFilePath(src), toRelativeFilePath(dst));
+ }
+
+ @Override
+ @Deprecated
+ public void rename(final Path src, final Path dst,
+ final Options.Rename... options) throws IOException {
+ statistics.incrementWriteOps(1);
+ boolean overwrite = false;
+ for (Rename renameOption : options) {
+ if (renameOption == Rename.OVERWRITE) {
+ overwrite = true;
+ break;
+ }
+ }
+ adlClient
+ .rename(toRelativeFilePath(src), toRelativeFilePath(dst), overwrite);
+ }
+
+ /**
+ * Concat existing files together.
+ *
+ * @param trg the path to the target destination.
+ * @param srcs the paths to the sources to use for the concatenation.
+ * @throws IOException when system error, internal server error or user error
+ */
+ @Override
+ public void concat(final Path trg, final Path[] srcs) throws IOException {
+ statistics.incrementWriteOps(1);
+ List<String> sourcesList = new ArrayList<String>();
+ for (Path entry : srcs) {
+ sourcesList.add(toRelativeFilePath(entry));
+ }
+ adlClient.concatenateFiles(toRelativeFilePath(trg), sourcesList);
+ }
+
+ /**
+ * Delete a file.
+ *
+ * @param path the path to delete.
+ * @param recursive if path is a directory and set to
+ * true, the directory is deleted else throws an exception.
+ * In case of a file the recursive can be set to either
+ * true or false.
+ * @return true if delete is successful else false.
+ * @throws IOException when system error, internal server error or user error
+ */
+ @Override
+ public boolean delete(final Path path, final boolean recursive)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ return recursive ?
+ adlClient.deleteRecursive(toRelativeFilePath(path)) :
+ adlClient.delete(toRelativeFilePath(path));
+ }
+
+ /**
+ * Make the given file and all non-existent parents into
+ * directories. Has the semantics of Unix 'mkdir -p'.
+ * Existence of the directory hierarchy is not an error.
+ *
+ * @param path path to create
+ * @param permission to apply to path
+ */
+ @Override
+ public boolean mkdirs(final Path path, final FsPermission permission)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ return adlClient.createDirectory(toRelativeFilePath(path),
+ Integer.toOctalString(applyUMask(permission).toShort()));
+ }
+
+ private FileStatus[] toFileStatuses(final List<DirectoryEntry> entries,
+ final Path parent) {
+ FileStatus[] fileStatuses = new FileStatus[entries.size()];
+ int index = 0;
+ for (DirectoryEntry entry : entries) {
+ FileStatus status = toFileStatus(entry, parent);
+ if (!(entry.name == null || entry.name == "")) {
+ status.setPath(
+ new Path(parent.makeQualified(uri, workingDirectory), entry.name));
+ }
+
+ fileStatuses[index++] = status;
+ }
+
+ return fileStatuses;
+ }
+
+ private FsPermission applyUMask(FsPermission permission) {
+ if (permission == null) {
+ permission = FsPermission.getDefault();
+ }
+ return permission.applyUMask(FsPermission.getUMask(getConf()));
+ }
+
+ private FileStatus toFileStatus(final DirectoryEntry entry, final Path f) {
+ boolean isDirectory = entry.type == DirectoryEntryType.DIRECTORY;
+ long lastModificationData = entry.lastModifiedTime.getTime();
+ long lastAccessTime = entry.lastAccessTime.getTime();
+ FsPermission permission = new AdlPermission(aclBitStatus,
+ Short.valueOf(entry.permission, 8));
+ String user = entry.user;
+ String group = entry.group;
+
+ FileStatus status;
+ if (overrideOwner) {
+ status = new FileStatus(entry.length, isDirectory, ADL_REPLICATION_FACTOR,
+ ADL_BLOCK_SIZE, lastModificationData, lastAccessTime, permission,
+ userName, "hdfs", this.makeQualified(f));
+ } else {
+ status = new FileStatus(entry.length, isDirectory, ADL_REPLICATION_FACTOR,
+ ADL_BLOCK_SIZE, lastModificationData, lastAccessTime, permission,
+ user, group, this.makeQualified(f));
+ }
+
+ return status;
+ }
+
+ /**
+ * Set owner of a path (i.e. a file or a directory).
+ * The parameters owner and group cannot both be null.
+ *
+ * @param path The path
+ * @param owner If it is null, the original username remains unchanged.
+ * @param group If it is null, the original groupname remains unchanged.
+ */
+ @Override
+ public void setOwner(final Path path, final String owner, final String group)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ adlClient.setOwner(toRelativeFilePath(path), owner, group);
+ }
+
+ /**
+ * Set permission of a path.
+ *
+ * @param path The path
+ * @param permission Access permission
+ */
+ @Override
+ public void setPermission(final Path path, final FsPermission permission)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ adlClient.setPermission(toRelativeFilePath(path),
+ Integer.toOctalString(permission.toShort()));
+ }
+
+ /**
+ * Modifies ACL entries of files and directories. This method can add new ACL
+ * entries or modify the permissions on existing ACL entries. All existing
+ * ACL entries that are not specified in this call are retained without
+ * changes. (Modifications are merged into the current ACL.)
+ *
+ * @param path Path to modify
+ * @param aclSpec List of AclEntry describing modifications
+ * @throws IOException if an ACL could not be modified
+ */
+ @Override
+ public void modifyAclEntries(final Path path, final List<AclEntry> aclSpec)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ List<com.microsoft.azure.datalake.store.acl.AclEntry> msAclEntries = new
+ ArrayList<com.microsoft.azure.datalake.store.acl.AclEntry>();
+ for (AclEntry aclEntry : aclSpec) {
+ msAclEntries.add(com.microsoft.azure.datalake.store.acl.AclEntry
+ .parseAclEntry(aclEntry.toString()));
+ }
+ adlClient.modifyAclEntries(toRelativeFilePath(path), msAclEntries);
+ }
+
+ /**
+ * Removes ACL entries from files and directories. Other ACL entries are
+ * retained.
+ *
+ * @param path Path to modify
+ * @param aclSpec List of AclEntry describing entries to remove
+ * @throws IOException if an ACL could not be modified
+ */
+ @Override
+ public void removeAclEntries(final Path path, final List<AclEntry> aclSpec)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ List<com.microsoft.azure.datalake.store.acl.AclEntry> msAclEntries = new
+ ArrayList<com.microsoft.azure.datalake.store.acl.AclEntry>();
+ for (AclEntry aclEntry : aclSpec) {
+ msAclEntries.add(com.microsoft.azure.datalake.store.acl.AclEntry
+ .parseAclEntry(aclEntry.toString(), true));
+ }
+ adlClient.removeAclEntries(toRelativeFilePath(path), msAclEntries);
+ }
+
+ /**
+ * Removes all default ACL entries from files and directories.
+ *
+ * @param path Path to modify
+ * @throws IOException if an ACL could not be modified
+ */
+ @Override
+ public void removeDefaultAcl(final Path path) throws IOException {
+ statistics.incrementWriteOps(1);
+ adlClient.removeDefaultAcls(toRelativeFilePath(path));
+ }
+
+ /**
+ * Removes all but the base ACL entries of files and directories. The entries
+ * for user, group, and others are retained for compatibility with permission
+ * bits.
+ *
+ * @param path Path to modify
+ * @throws IOException if an ACL could not be removed
+ */
+ @Override
+ public void removeAcl(final Path path) throws IOException {
+ statistics.incrementWriteOps(1);
+ adlClient.removeAllAcls(toRelativeFilePath(path));
+ }
+
+ /**
+ * Fully replaces ACL of files and directories, discarding all existing
+ * entries.
+ *
+ * @param path Path to modify
+ * @param aclSpec List of AclEntry describing modifications, must include
+ * entries for user, group, and others for compatibility with
+ * permission bits.
+ * @throws IOException if an ACL could not be modified
+ */
+ @Override
+ public void setAcl(final Path path, final List<AclEntry> aclSpec)
+ throws IOException {
+ statistics.incrementWriteOps(1);
+ List<com.microsoft.azure.datalake.store.acl.AclEntry> msAclEntries = new
+ ArrayList<com.microsoft.azure.datalake.store.acl.AclEntry>();
+ for (AclEntry aclEntry : aclSpec) {
+ msAclEntries.add(com.microsoft.azure.datalake.store.acl.AclEntry
+ .parseAclEntry(aclEntry.toString()));
+ }
+
+ adlClient.setAcl(toRelativeFilePath(path), msAclEntries);
+ }
+
+ /**
+ * Gets the ACL of a file or directory.
+ *
+ * @param path Path to get
+ * @return AclStatus describing the ACL of the file or directory
+ * @throws IOException if an ACL could not be read
+ */
+ @Override
+ public AclStatus getAclStatus(final Path path) throws IOException {
+ statistics.incrementReadOps(1);
+ com.microsoft.azure.datalake.store.acl.AclStatus adlStatus = adlClient
+ .getAclStatus(toRelativeFilePath(path));
+ AclStatus.Builder aclStatusBuilder = new AclStatus.Builder();
+ aclStatusBuilder.owner(adlStatus.owner);
+ aclStatusBuilder.group(adlStatus.group);
+ aclStatusBuilder.setPermission(
+ new FsPermission(Short.valueOf(adlStatus.octalPermissions, 8)));
+ aclStatusBuilder.stickyBit(adlStatus.stickyBit);
+ String aclListString = com.microsoft.azure.datalake.store.acl.AclEntry
+ .aclListToString(adlStatus.aclSpec);
+ List<AclEntry> aclEntries = AclEntry.parseAclSpec(aclListString, true);
+ aclStatusBuilder.addEntries(aclEntries);
+ return aclStatusBuilder.build();
+ }
+
+ /**
+ * Checks if the user can access a path. The mode specifies which access
+ * checks to perform. If the requested permissions are granted, then the
+ * method returns normally. If access is denied, then the method throws an
+ * {@link AccessControlException}.
+ *
+ * @param path Path to check
+ * @param mode type of access to check
+ * @throws AccessControlException if access is denied
+ * @throws java.io.FileNotFoundException if the path does not exist
+ * @throws IOException see specific implementation
+ */
+ @Override
+ public void access(final Path path, FsAction mode) throws IOException {
+ statistics.incrementReadOps(1);
+ if (!adlClient.checkAccess(toRelativeFilePath(path), mode.SYMBOL)) {
+ throw new AccessControlException("Access Denied : " + path.toString());
+ }
+ }
+
+ /**
+ * Return the {@link ContentSummary} of a given {@link Path}.
+ *
+ * @param f path to use
+ */
+ @Override
+ public ContentSummary getContentSummary(Path f) throws IOException {
+ statistics.incrementReadOps(1);
+ com.microsoft.azure.datalake.store.ContentSummary msSummary = adlClient
+ .getContentSummary(toRelativeFilePath(f));
+ return new Builder().length(msSummary.length)
+ .directoryCount(msSummary.directoryCount).fileCount(msSummary.fileCount)
+ .spaceConsumed(msSummary.spaceConsumed).build();
+ }
+
+ @VisibleForTesting
+ protected String getTransportScheme() {
+ return SECURE_TRANSPORT_SCHEME;
+ }
+
+ @VisibleForTesting
+ String toRelativeFilePath(Path path) {
+ return path.makeQualified(uri, workingDirectory).toUri().getPath();
+ }
+
+ /**
+ * Get the current working directory for the given file system.
+ *
+ * @return the directory pathname
+ */
+ @Override
+ public Path getWorkingDirectory() {
+ return workingDirectory;
+ }
+
+ /**
+ * Set the current working directory for the given file system. All relative
+ * paths will be resolved relative to it.
+ *
+ * @param dir Working directory path.
+ */
+ @Override
+ public void setWorkingDirectory(final Path dir) {
+ if (dir == null) {
+ throw new InvalidPathException("Working directory cannot be set to NULL");
+ }
+
+ /**
+ * Do not validate the scheme and URI of the passsed parameter. When Adls
+ * runs as additional file system, working directory set has the default
+ * file system scheme and uri.
+ *
+ * Found a problem during PIG execution in
+ * https://github.com/apache/pig/blob/branch-0
+ * .15/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer
+ * /PigInputFormat.java#L235
+ * However similar problem would be present in other application so
+ * defaulting to build working directory using relative path only.
+ */
+ this.workingDirectory = this.makeAbsolute(dir);
+ }
+
+ /**
+ * Return the number of bytes that large input files should be optimally
+ * be split into to minimize i/o time.
+ *
+ * @deprecated use {@link #getDefaultBlockSize(Path)} instead
+ */
+ @Deprecated
+ public long getDefaultBlockSize() {
+ return ADL_BLOCK_SIZE;
+ }
+
+ /**
+ * Return the number of bytes that large input files should be optimally
+ * be split into to minimize i/o time. The given path will be used to
+ * locate the actual filesystem. The full path does not have to exist.
+ *
+ * @param f path of file
+ * @return the default block size for the path's filesystem
+ */
+ public long getDefaultBlockSize(Path f) {
+ return getDefaultBlockSize();
+ }
+
+ /**
+ * Get the block size.
+ * @param f the filename
+ * @return the number of bytes in a block
+ */
+ /**
+ * @deprecated Use getFileStatus() instead
+ */
+ @Deprecated
+ public long getBlockSize(Path f) throws IOException {
+ return ADL_BLOCK_SIZE;
+ }
+
+ @Override
+ public BlockLocation[] getFileBlockLocations(final FileStatus status,
+ final long offset, final long length) throws IOException {
+ if (status == null) {
+ return null;
+ }
+
+ if ((offset < 0) || (length < 0)) {
+ throw new IllegalArgumentException("Invalid start or len parameter");
+ }
+
+ if (status.getLen() < offset) {
+ return new BlockLocation[0];
+ }
+
+ final String[] name = {"localhost"};
+ final String[] host = {"localhost"};
+ long blockSize = ADL_BLOCK_SIZE;
+ int numberOfLocations =
+ (int) (length / blockSize) + ((length % blockSize == 0) ? 0 : 1);
+ BlockLocation[] locations = new BlockLocation[numberOfLocations];
+ for (int i = 0; i < locations.length; i++) {
+ long currentOffset = offset + (i * blockSize);
+ long currentLength = Math.min(blockSize, offset + length - currentOffset);
+ locations[i] = new BlockLocation(name, host, currentOffset,
+ currentLength);
+ }
+
+ return locations;
+ }
+
+ @Override
+ public BlockLocation[] getFileBlockLocations(final Path p, final long offset,
+ final long length) throws IOException {
+ // read ops incremented in getFileStatus
+ FileStatus fileStatus = getFileStatus(p);
+ return getFileBlockLocations(fileStatus, offset, length);
+ }
+
+ /**
+ * Get replication.
+ *
+ * @param src file name
+ * @return file replication
+ * @deprecated Use getFileStatus() instead
+ */
+ @Deprecated
+ public short getReplication(Path src) {
+ return ADL_REPLICATION_FACTOR;
+ }
+
+ private Path makeAbsolute(Path path) {
+ return path.isAbsolute() ? path : new Path(this.workingDirectory, path);
+ }
+
+ private static String getNonEmptyVal(Configuration conf, String key) {
+ String value = conf.get(key);
+ if (StringUtils.isEmpty(value)) {
+ throw new IllegalArgumentException(
+ "No value for " + key + " found in conf file.");
+ }
+ return value;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsInputStream.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsInputStream.java
new file mode 100644
index 0000000..5248cbf
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsInputStream.java
@@ -0,0 +1,149 @@
+/*
+ * 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.hadoop.fs.adl;
+
+import com.microsoft.azure.datalake.store.ADLFileInputStream;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+
+import java.io.IOException;
+
+import static org.apache.hadoop.fs.adl.AdlConfKeys
+ .ADL_EXPERIMENT_POSITIONAL_READ_DEFAULT;
+import static org.apache.hadoop.fs.adl.AdlConfKeys
+ .ADL_EXPERIMENT_POSITIONAL_READ_KEY;
+import static org.apache.hadoop.fs.adl.AdlConfKeys
+ .DEFAULT_READ_AHEAD_BUFFER_SIZE;
+import static org.apache.hadoop.fs.adl.AdlConfKeys.READ_AHEAD_BUFFER_SIZE_KEY;
+
+/**
+ * Wraps {@link ADLFileInputStream} implementation.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public final class AdlFsInputStream extends FSInputStream {
+
+ private final ADLFileInputStream in;
+ private final Statistics stat;
+ private final boolean enablePositionalReadExperiment;
+
+ public AdlFsInputStream(ADLFileInputStream inputStream, Statistics statistics,
+ Configuration conf) throws IOException {
+ this.in = inputStream;
+ this.in.setBufferSize(conf.getInt(READ_AHEAD_BUFFER_SIZE_KEY,
+ DEFAULT_READ_AHEAD_BUFFER_SIZE));
+ enablePositionalReadExperiment = conf
+ .getBoolean(ADL_EXPERIMENT_POSITIONAL_READ_KEY,
+ ADL_EXPERIMENT_POSITIONAL_READ_DEFAULT);
+ stat = statistics;
+ }
+
+ @Override
+ public synchronized void seek(long pos) throws IOException {
+ in.seek(pos);
+ }
+
+ /**
+ * Return the current offset from the start of the file.
+ */
+ @Override
+ public synchronized long getPos() throws IOException {
+ return in.getPos();
+ }
+
+ @Override
+ public boolean seekToNewSource(long l) throws IOException {
+ return false;
+ }
+
+ @Override
+ public synchronized int read() throws IOException {
+ int ch = in.read();
+ if (stat != null && ch != -1) {
+ stat.incrementBytesRead(1);
+ }
+ return ch;
+ }
+
+ @Override
+ public int read(long position, byte[] buffer, int offset, int length)
+ throws IOException {
+ int numberOfByteRead = 0;
+ if (enablePositionalReadExperiment) {
+ numberOfByteRead = in.read(position, buffer, offset, length);
+ } else {
+ numberOfByteRead = super.read(position, buffer, offset, length);
+ }
+
+ if (stat != null && numberOfByteRead > 0) {
+ stat.incrementBytesRead(numberOfByteRead);
+ }
+ return numberOfByteRead;
+ }
+
+ @Override
+ public synchronized int read(byte[] buffer, int offset, int length)
+ throws IOException {
+ int numberOfByteRead = in.read(buffer, offset, length);
+ if (stat != null && numberOfByteRead > 0) {
+ stat.incrementBytesRead(numberOfByteRead);
+ }
+ return numberOfByteRead;
+ }
+
+ /**
+ * This method returns the remaining bytes in the stream, rather than the
+ * expected Java
+ * interpretation of {@link java.io.InputStream#available()}, which expects
+ * the
+ * number of remaining
+ * bytes in the local buffer. Moreover, it caps the value returned to a
+ * maximum of Integer.MAX_VALUE.
+ * These changed behaviors are to ensure compatibility with the
+ * expectations of HBase WAL reader,
+ * which depends on available() returning the number of bytes in stream.
+ *
+ * Given all other FileSystems in the hadoop ecosystem (especially HDFS) do
+ * this, it is possible other
+ * apps other than HBase would also pick up expectation of this behavior
+ * based on HDFS implementation.
+ * Therefore keeping this quirky behavior here, to ensure compatibility.
+ *
+ * @return remaining bytes in the stream, with maximum of Integer.MAX_VALUE.
+ * @throws IOException If fails to get the position or file length from SDK.
+ */
+ @Override
+ public synchronized int available() throws IOException {
+ return (int) Math.min(in.length() - in.getPos(), Integer.MAX_VALUE);
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ in.close();
+ }
+
+ @Override
+ public synchronized long skip(long pos) throws IOException {
+ return in.skip(pos);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsOutputStream.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsOutputStream.java
new file mode 100644
index 0000000..2b89fb0
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlFsOutputStream.java
@@ -0,0 +1,82 @@
+/*
+ * 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.hadoop.fs.adl;
+
+import com.microsoft.azure.datalake.store.ADLFileOutputStream;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Syncable;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import static org.apache.hadoop.fs.adl.AdlConfKeys
+ .DEFAULT_WRITE_AHEAD_BUFFER_SIZE;
+import static org.apache.hadoop.fs.adl.AdlConfKeys.WRITE_BUFFER_SIZE_KEY;
+
+/**
+ * Wraps {@link com.microsoft.azure.datalake.store.ADLFileOutputStream}
+ * implementation.
+ *
+ * Flush semantics.
+ * no-op, since some parts of hadoop ecosystem call flush(), expecting it to
+ * have no perf impact. In hadoop filesystems, flush() itself guarantees no
+ * durability: that is achieved by calling hflush() or hsync()
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public final class AdlFsOutputStream extends OutputStream implements Syncable {
+ private final ADLFileOutputStream out;
+
+ public AdlFsOutputStream(ADLFileOutputStream out, Configuration configuration)
+ throws IOException {
+ this.out = out;
+ out.setBufferSize(configuration
+ .getInt(WRITE_BUFFER_SIZE_KEY, DEFAULT_WRITE_AHEAD_BUFFER_SIZE));
+ }
+
+ @Override
+ public synchronized void write(int b) throws IOException {
+ out.write(b);
+ }
+
+ @Override
+ public synchronized void write(byte[] b, int off, int len)
+ throws IOException {
+ out.write(b, off, len);
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ out.close();
+ }
+
+ public synchronized void sync() throws IOException {
+ out.flush();
+ }
+
+ public synchronized void hflush() throws IOException {
+ out.flush();
+ }
+
+ public synchronized void hsync() throws IOException {
+ out.flush();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlPermission.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlPermission.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlPermission.java
new file mode 100644
index 0000000..af3342a
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/AdlPermission.java
@@ -0,0 +1,69 @@
+/*
+ * 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.hadoop.fs.adl;
+
+import org.apache.hadoop.fs.permission.FsPermission;
+
+/**
+ * Hadoop shell command -getfacl does not invoke getAclStatus if FsPermission
+ * from getFileStatus has not set ACL bit to true. By default getAclBit returns
+ * false.
+ *
+ * Provision to make additional call to invoke getAclStatus would be redundant
+ * when adls is running as additional FS. To avoid this redundancy, provided
+ * configuration to return true/false on getAclBit.
+ */
+class AdlPermission extends FsPermission {
+ private final boolean aclBit;
+
+ AdlPermission(boolean aclBitStatus, Short aShort) {
+ super(aShort);
+ this.aclBit = aclBitStatus;
+ }
+
+ /**
+ * Returns true if "adl.feature.support.acl.bit" configuration is set to
+ * true.
+ *
+ * If configuration is not set then default value is true.
+ *
+ * @return If configuration is not set then default value is true.
+ */
+ public boolean getAclBit() {
+ return aclBit;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof FsPermission) {
+ FsPermission that = (FsPermission) obj;
+ return this.getUserAction() == that.getUserAction()
+ && this.getGroupAction() == that.getGroupAction()
+ && this.getOtherAction() == that.getOtherAction()
+ && this.getStickyBit() == that.getStickyBit();
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return toShort();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/SdkTokenProviderAdapter.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/SdkTokenProviderAdapter.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/SdkTokenProviderAdapter.java
new file mode 100644
index 0000000..7b107ae
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/SdkTokenProviderAdapter.java
@@ -0,0 +1,41 @@
+/*
+ * 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.hadoop.fs.adl;
+
+import com.microsoft.azure.datalake.store.oauth2.AccessTokenProvider;
+import com.microsoft.azure.datalake.store.oauth2.AzureADToken;
+import org.apache.hadoop.fs.adl.oauth2.AzureADTokenProvider;
+
+import java.io.IOException;
+
+final class SdkTokenProviderAdapter extends AccessTokenProvider {
+
+ private AzureADTokenProvider tokenProvider;
+
+ SdkTokenProviderAdapter(AzureADTokenProvider tp) {
+ this.tokenProvider = tp;
+ }
+
+ protected AzureADToken refreshToken() throws IOException {
+ AzureADToken azureADToken = new AzureADToken();
+ azureADToken.accessToken = tokenProvider.getAccessToken();
+ azureADToken.expiry = tokenProvider.getExpiryTime();
+ return azureADToken;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/TokenProviderType.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/TokenProviderType.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/TokenProviderType.java
new file mode 100644
index 0000000..9fd4f4f
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/TokenProviderType.java
@@ -0,0 +1,25 @@
+/*
+ * 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.hadoop.fs.adl;
+
+enum TokenProviderType {
+ RefreshToken,
+ ClientCredential,
+ Custom
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/AzureADTokenProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/AzureADTokenProvider.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/AzureADTokenProvider.java
new file mode 100644
index 0000000..a0b3922
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/AzureADTokenProvider.java
@@ -0,0 +1,70 @@
+/*
+ * 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.hadoop.fs.adl.oauth2;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.Date;
+
+/**
+ * Provide an Azure Active Directory supported
+ * OAuth2 access token to be used to authenticate REST calls against Azure data
+ * lake file system {@link org.apache.hadoop.fs.adl.AdlFileSystem}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public abstract class AzureADTokenProvider {
+
+ /**
+ * Initialize with supported configuration. This method is invoked when the
+ * {@link org.apache.hadoop.fs.adl.AdlFileSystem#initialize
+ * (URI, Configuration)} method is invoked.
+ *
+ * @param configuration Configuration object
+ * @throws IOException if instance can not be configured.
+ */
+ public abstract void initialize(Configuration configuration)
+ throws IOException;
+
+ /**
+ * Obtain the access token that should be added to https connection's header.
+ * Will be called depending upon {@link #getExpiryTime()} expiry time is set,
+ * so implementations should be performant. Implementations are responsible
+ * for any refreshing of the token.
+ *
+ * @return String containing the access token
+ * @throws IOException if there is an error fetching the token
+ */
+ public abstract String getAccessToken() throws IOException;
+
+ /**
+ * Obtain expiry time of the token. If implementation is performant enough to
+ * maintain expiry and expect {@link #getAccessToken()} call for every
+ * connection then safe to return current or past time.
+ *
+ * However recommended to use the token expiry time received from Azure Active
+ * Directory.
+ *
+ * @return Date to expire access token retrieved from AAD.
+ */
+ public abstract Date getExpiryTime();
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/CachedRefreshTokenBasedAccessTokenProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/CachedRefreshTokenBasedAccessTokenProvider.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/CachedRefreshTokenBasedAccessTokenProvider.java
deleted file mode 100644
index b7f3b00..0000000
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/CachedRefreshTokenBasedAccessTokenProvider.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * 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.hadoop.fs.adl.oauth2;
-
-import java.io.IOException;
-import java.util.Map;
-import java.util.LinkedHashMap;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.web.oauth2.AccessTokenProvider;
-import org.apache.hadoop.hdfs.web.oauth2.ConfRefreshTokenBasedAccessTokenProvider;
-import org.apache.hadoop.hdfs.web.oauth2.PrivateCachedRefreshTokenBasedAccessTokenProvider;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.OAUTH_CLIENT_ID_KEY;
-import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.OAUTH_REFRESH_URL_KEY;
-import static org.apache.hadoop.hdfs.web.oauth2.ConfRefreshTokenBasedAccessTokenProvider.OAUTH_REFRESH_TOKEN_KEY;
-
-/**
- * Share refresh tokens across all ADLS instances with a common client ID. The
- * {@link AccessTokenProvider} can be shared across multiple instances,
- * amortizing the cost of refreshing tokens.
- */
-public class CachedRefreshTokenBasedAccessTokenProvider
- extends PrivateCachedRefreshTokenBasedAccessTokenProvider {
-
- public static final String FORCE_REFRESH = "adl.force.token.refresh";
-
- private static final Logger LOG =
- LoggerFactory.getLogger(CachedRefreshTokenBasedAccessTokenProvider.class);
-
- /** Limit size of provider cache. */
- static final int MAX_PROVIDERS = 10;
- @SuppressWarnings("serial")
- private static final Map<String, AccessTokenProvider> CACHE =
- new LinkedHashMap<String, AccessTokenProvider>() {
- @Override
- public boolean removeEldestEntry(
- Map.Entry<String, AccessTokenProvider> e) {
- return size() > MAX_PROVIDERS;
- }
- };
-
- private AccessTokenProvider instance = null;
-
- /**
- * Create handle for cached instance.
- */
- public CachedRefreshTokenBasedAccessTokenProvider() {
- }
-
- /**
- * Gets the access token from internally cached
- * ConfRefreshTokenBasedAccessTokenProvider instance.
- *
- * @return Valid OAuth2 access token for the user.
- * @throws IOException when system error, internal server error or user error
- */
- @Override
- public synchronized String getAccessToken() throws IOException {
- return instance.getAccessToken();
- }
-
- /**
- * @return A cached Configuration consistent with the parameters of this
- * instance.
- */
- @Override
- public synchronized Configuration getConf() {
- return instance.getConf();
- }
-
- /**
- * Configure cached instance. Note that the Configuration instance returned
- * from subsequent calls to {@link #getConf() getConf} may be from a
- * previous, cached entry.
- * @param conf Configuration instance
- */
- @Override
- public synchronized void setConf(Configuration conf) {
- String id = conf.get(OAUTH_CLIENT_ID_KEY);
- if (null == id) {
- throw new IllegalArgumentException("Missing client ID");
- }
- synchronized (CACHE) {
- instance = CACHE.get(id);
- if (null == instance
- || conf.getBoolean(FORCE_REFRESH, false)
- || replace(instance, conf)) {
- instance = newInstance();
- // clone configuration
- instance.setConf(new Configuration(conf));
- CACHE.put(id, instance);
- LOG.debug("Created new client {}", id);
- }
- }
- }
-
- AccessTokenProvider newInstance() {
- return new ConfRefreshTokenBasedAccessTokenProvider();
- }
-
- private static boolean replace(AccessTokenProvider cached, Configuration c2) {
- // ConfRefreshTokenBasedAccessTokenProvider::setConf asserts !null
- final Configuration c1 = cached.getConf();
- for (String key : new String[] {
- OAUTH_REFRESH_TOKEN_KEY, OAUTH_REFRESH_URL_KEY }) {
- if (!c1.get(key).equals(c2.get(key))) {
- // replace cached instance for this clientID
- return true;
- }
- }
- return false;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/package-info.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/package-info.java
index b444984..1613941 100644
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/package-info.java
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/oauth2/package-info.java
@@ -20,4 +20,4 @@
/**
* public interface to expose OAuth2 authentication related features.
*/
-package org.apache.hadoop.fs.adl.oauth2;
+package org.apache.hadoop.fs.adl.oauth2;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/package-info.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/package-info.java
index 98e6a77..456eebc 100644
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/package-info.java
+++ b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/fs/adl/package-info.java
@@ -20,4 +20,4 @@
/**
* Supporting classes for metrics instrumentation.
*/
-package org.apache.hadoop.fs.adl;
+package org.apache.hadoop.fs.adl;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/ADLConfKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/ADLConfKeys.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/ADLConfKeys.java
deleted file mode 100644
index a7f932f..0000000
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/ADLConfKeys.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.hadoop.hdfs.web;
-
-/**
- * Constants.
- */
-public final class ADLConfKeys {
- public static final String
- ADL_FEATURE_CONCURRENT_READ_AHEAD_MAX_CONCURRENT_CONN =
- "adl.feature.override.readahead.max.concurrent.connection";
- public static final int
- ADL_FEATURE_CONCURRENT_READ_AHEAD_MAX_CONCURRENT_CONN_DEFAULT = 2;
- public static final String ADL_WEBSDK_VERSION_KEY = "ADLFeatureSet";
- static final String ADL_DEBUG_OVERRIDE_LOCAL_USER_AS_OWNER =
- "adl.debug.override.localuserasfileowner";
- static final boolean ADL_DEBUG_SET_LOCAL_USER_AS_OWNER_DEFAULT = false;
- static final String ADL_FEATURE_REDIRECT_OFF =
- "adl.feature.override.redirection.off";
- static final boolean ADL_FEATURE_REDIRECT_OFF_DEFAULT = true;
- static final String ADL_FEATURE_GET_BLOCK_LOCATION_LOCALLY_BUNDLED =
- "adl.feature.override.getblocklocation.locally.bundled";
- static final boolean ADL_FEATURE_GET_BLOCK_LOCATION_LOCALLY_BUNDLED_DEFAULT
- = true;
- static final String ADL_FEATURE_CONCURRENT_READ_WITH_READ_AHEAD =
- "adl.feature.override.readahead";
- static final boolean ADL_FEATURE_CONCURRENT_READ_WITH_READ_AHEAD_DEFAULT =
- true;
- static final String ADL_FEATURE_CONCURRENT_READ_WITH_READ_AHEAD_BUFFER_SIZE =
- "adl.feature.override.readahead.max.buffersize";
-
- static final int KB = 1024;
- static final int MB = KB * KB;
- static final int DEFAULT_BLOCK_SIZE = 4 * MB;
- static final int DEFAULT_EXTENT_SIZE = 256 * MB;
- static final int DEFAULT_TIMEOUT_IN_SECONDS = 120;
- static final int
- ADL_FEATURE_CONCURRENT_READ_WITH_READ_AHEAD_BUFFER_SIZE_DEFAULT =
- 8 * MB;
-
- private ADLConfKeys() {
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/5c61ad24/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/BufferManager.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/BufferManager.java b/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/BufferManager.java
deleted file mode 100644
index 350c6e7..0000000
--- a/hadoop-tools/hadoop-azure-datalake/src/main/java/org/apache/hadoop/hdfs/web/BufferManager.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/*
- * 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.hadoop.hdfs.web;
-
-/**
- * Responsible for holding buffered data in the process. Hold only 1 and only
- * 1 buffer block in the memory. Buffer block
- * information is for the given file and the offset from the which the block
- * is fetched. Across the webhdfs instances if
- * same buffer block has been used then backend trip is avoided. Buffer block
- * is certainly important since ADL fetches
- * large amount of data (Default is 4MB however can be configured through
- * core-site.xml) from the backend.
- * Observation is in case of ORC/Avro kind of compressed file, buffer block
- * does not avoid few backend calls across
- * webhdfs
- * instances.
- */
-final class BufferManager {
- private static final BufferManager BUFFER_MANAGER_INSTANCE = new
- BufferManager();
- private static Object lock = new Object();
- private Buffer buffer = null;
- private String fileName;
-
- /**
- * Constructor.
- */
- private BufferManager() {
- }
-
- public static Object getLock() {
- return lock;
- }
-
- public static BufferManager getInstance() {
- return BUFFER_MANAGER_INSTANCE;
- }
-
- /**
- * Validate if the current buffer block is of given stream.
- *
- * @param path ADL stream path
- * @param offset Stream offset that caller is interested in
- * @return True if the buffer block is available otherwise false
- */
- boolean hasValidDataForOffset(String path, long offset) {
- if (this.fileName == null) {
- return false;
- }
-
- if (!this.fileName.equals(path)) {
- return false;
- }
-
- if (buffer == null) {
- return false;
- }
-
- if ((offset < buffer.offset) || (offset >= (buffer.offset
- + buffer.data.length))) {
- return false;
- }
-
- return true;
- }
-
- /**
- * Clean buffer block.
- */
- void clear() {
- buffer = null;
- }
-
- /**
- * Validate if the current buffer block is of given stream. For now partial
- * data available is not supported.
- * Data must be available exactly or within the range of offset and size
- * passed as parameter.
- *
- * @param path Stream path
- * @param offset Offset of the stream
- * @param size Size of the data from the offset of the stream caller
- * interested in
- * @return True if the data is available from the given offset and of the
- * size caller is interested in.
- */
- boolean hasData(String path, long offset, int size) {
-
- if (!hasValidDataForOffset(path, offset)) {
- return false;
- }
-
- if ((size + offset) > (buffer.data.length + buffer.offset)) {
- return false;
- }
- return true;
- }
-
- /**
- * Return the buffer block from the requested offset. It is caller
- * responsibility to check if the buffer block is
- * of there interest and offset is valid.
- *
- * @param data Byte array to be filed from the buffer block
- * @param offset Data to be fetched from the offset.
- */
- void get(byte[] data, long offset) {
- System.arraycopy(buffer.data, (int) (offset - buffer.offset), data, 0,
- data.length);
- }
-
- /**
- * Create new empty buffer block of the given size.
- *
- * @param len Size of the buffer block.
- * @return Empty byte array.
- */
- byte[] getEmpty(int len) {
- return new byte[len];
- }
-
- /**
- * This function allows caller to specify new buffer block for the stream
- * which is pulled from the backend.
- *
- * @param data Buffer
- * @param path Stream path to which buffer belongs to
- * @param offset Stream offset where buffer start with
- */
- void add(byte[] data, String path, long offset) {
- if (data == null) {
- return;
- }
-
- buffer = new Buffer();
- buffer.data = data;
- buffer.offset = offset;
- this.fileName = path;
- }
-
- /**
- * @return Size of the buffer.
- */
- int getBufferSize() {
- return buffer.data.length;
- }
-
- /**
- * @return Stream offset where buffer start with
- */
- long getBufferOffset() {
- return buffer.offset;
- }
-
- /**
- * Buffer container.
- */
- static class Buffer {
- private byte[] data;
- private long offset;
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org