Skip to content

Code to set prefix mode based on account type #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
saxenapranav wants to merge 1 commit into
base: cherryPick-3-apr-3
Choose a base branch
from
Open

Code to set prefix mode based on account type #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.io.IOException;
import java.lang.reflect.Field;

import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;

Expand Down Expand Up @@ -83,6 +84,7 @@ public class AbfsConfiguration{
private final Configuration rawConfig;
private final String accountName;
private final boolean isSecure;
private PrefixMode prefixMode;
private static final Logger LOG = LoggerFactory.getLogger(AbfsConfiguration.class);

@StringConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_ACCOUNT_IS_HNS_ENABLED,
Expand Down Expand Up @@ -336,6 +338,14 @@ public Trilean getIsNamespaceEnabledAccount() {
return Trilean.getTrilean(isNamespaceEnabledAccount);
}

public PrefixMode getPrefixMode() {
return prefixMode;
}

public void setPrefixMode(final PrefixMode prefixMode) {
this.prefixMode = prefixMode;
}

/**
* Gets the Azure Storage account name corresponding to this instance of configuration.
* @return the Azure Storage account name
Expand Down
23 changes: 21 additions & 2 deletions ...p-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.hadoop.fs.azurebfs.services.PrefixMode;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
Expand Down Expand Up @@ -142,6 +143,8 @@ public class AzureBlobFileSystem extends FileSystem
private DataBlocks.BlockFactory blockFactory;
/** Maximum Active blocks per OutputStream. */
private int blockOutputActiveBlocks;
private PrefixMode prefixMode = PrefixMode.DFS;
private boolean isNamespaceEnabled;

@Override
public void initialize(URI uri, Configuration configuration)
Expand Down Expand Up @@ -190,9 +193,25 @@ public void initialize(URI uri, Configuration configuration)
tracingHeaderFormat = abfsConfiguration.getTracingHeaderFormat();
this.setWorkingDirectory(this.getHomeDirectory());

TracingContext tracingContext = new TracingContext(clientCorrelationId,
fileSystemId, FSOperationType.CREATE_FILESYSTEM, tracingHeaderFormat, listener);
try {
isNamespaceEnabled = getIsNamespaceEnabled(tracingContext);
} catch (AbfsRestOperationException ex) {
/* since the filesystem has not been created. The API for HNS account would
* return 404 status.
*/
if(ex.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
isNamespaceEnabled = true;
} else {
throw ex;
}
}
if (!isNamespaceEnabled && uri.toString().contains(FileSystemUriSchemes.WASB_DNS_PREFIX)) {
this.prefixMode = PrefixMode.BLOB;
}
abfsConfiguration.setPrefixMode(this.prefixMode);
if (abfsConfiguration.getCreateRemoteFileSystemDuringInitialization()) {
TracingContext tracingContext = new TracingContext(clientCorrelationId,
fileSystemId, FSOperationType.CREATE_FILESYSTEM, tracingHeaderFormat, listener);
if (this.tryGetFileStatus(new Path(AbfsHttpConstants.ROOT_PATH), tracingContext) == null) {
try {
this.createFileSystem(tracingContext);
Expand Down
81 changes: 67 additions & 14 deletions ...p-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,9 @@
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.*;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_DELETE_CONSIDERED_IDEMPOTENT;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.SERVER_SIDE_ENCRYPTION_ALGORITHM;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_DNS_PREFIX;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.HTTPS_SCHEME;
import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.WASB_DNS_PREFIX;
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.*;
import static org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams.*;

Expand Down Expand Up @@ -175,6 +177,15 @@ private byte[] getSHA256Hash(String key) throws IOException {
}
}

private URL changePrefixFromBlobtoDfs(URL url) throws InvalidUriException {
try {
url = new URL(url.toString().replace(WASB_DNS_PREFIX, ABFS_DNS_PREFIX));
} catch (MalformedURLException ex) {
throw new InvalidUriException(url.toString());
}
return url;
}

private String getBase64EncodedString(String key) {
return getBase64EncodedString(key.getBytes(StandardCharsets.UTF_8));
}
Expand Down Expand Up @@ -243,7 +254,10 @@ public AbfsRestOperation createFilesystem(TracingContext tracingContext) throws
final AbfsUriQueryBuilder abfsUriQueryBuilder = new AbfsUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);

final URL url = createRequestUrl(abfsUriQueryBuilder.toString());
URL url = createRequestUrl(abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.CreateFileSystem,
this,
Expand All @@ -267,7 +281,10 @@ public AbfsRestOperation setFilesystemProperties(final String properties, Tracin
final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);

final URL url = createRequestUrl(abfsUriQueryBuilder.toString());
URL url = createRequestUrl(abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.SetFileSystemProperties,
this,
Expand All @@ -292,7 +309,10 @@ public AbfsRestOperation listPath(final String relativePath, final boolean recur
abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_UPN, String.valueOf(abfsConfiguration.isUpnUsed()));
appendSASTokenToQuery(relativePath, SASTokenProvider.LIST_OPERATION, abfsUriQueryBuilder);

final URL url = createRequestUrl(abfsUriQueryBuilder.toString());
URL url = createRequestUrl(abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.ListPaths,
this,
Expand All @@ -309,7 +329,10 @@ public AbfsRestOperation getFilesystemProperties(TracingContext tracingContext)
final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);

final URL url = createRequestUrl(abfsUriQueryBuilder.toString());
URL url = createRequestUrl(abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.GetFileSystemProperties,
this,
Expand All @@ -326,7 +349,10 @@ public AbfsRestOperation deleteFilesystem(TracingContext tracingContext) throws
final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, FILESYSTEM);

final URL url = createRequestUrl(abfsUriQueryBuilder.toString());
URL url = createRequestUrl(abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.DeleteFileSystem,
this,
Expand Down Expand Up @@ -407,7 +433,10 @@ public AbfsRestOperation acquireLease(final String path, int duration, TracingCo

final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.LeasePath,
this,
Expand All @@ -427,7 +456,10 @@ public AbfsRestOperation renewLease(final String path, final String leaseId,

final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.LeasePath,
this,
Expand All @@ -447,7 +479,10 @@ public AbfsRestOperation releaseLease(final String path,

final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.LeasePath,
this,
Expand All @@ -467,7 +502,10 @@ public AbfsRestOperation breakLease(final String path,

final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.LeasePath,
this,
Expand Down Expand Up @@ -651,7 +689,10 @@ public AbfsRestOperation setPathProperties(final String path, final String prope
abfsUriQueryBuilder.addQuery(QUERY_PARAM_ACTION, SET_PROPERTIES_ACTION);
appendSASTokenToQuery(path, SASTokenProvider.SET_PROPERTIES_OPERATION, abfsUriQueryBuilder);

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.SetPathProperties,
this,
Expand Down Expand Up @@ -680,7 +721,10 @@ public AbfsRestOperation getPathStatus(final String path, final boolean includeP
abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_UPN, String.valueOf(abfsConfiguration.isUpnUsed()));
appendSASTokenToQuery(path, operation, abfsUriQueryBuilder);

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.GetPathStatus,
this,
Expand Down Expand Up @@ -731,7 +775,10 @@ public AbfsRestOperation deletePath(final String path, final boolean recursive,
String operation = recursive ? SASTokenProvider.DELETE_RECURSIVE_OPERATION : SASTokenProvider.DELETE_OPERATION;
appendSASTokenToQuery(path, operation, abfsUriQueryBuilder);

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.DeletePath,
this,
Expand Down Expand Up @@ -815,7 +862,10 @@ public AbfsRestOperation setOwner(final String path, final String owner, final S
abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_ACTION, AbfsHttpConstants.SET_ACCESS_CONTROL);
appendSASTokenToQuery(path, SASTokenProvider.SET_OWNER_OPERATION, abfsUriQueryBuilder);

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.SetOwner,
this,
Expand All @@ -841,7 +891,10 @@ public AbfsRestOperation setPermission(final String path, final String permissio
abfsUriQueryBuilder.addQuery(HttpQueryParams.QUERY_PARAM_ACTION, AbfsHttpConstants.SET_ACCESS_CONTROL);
appendSASTokenToQuery(path, SASTokenProvider.SET_PERMISSION_OPERATION, abfsUriQueryBuilder);

final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
if (url.toString().contains(WASB_DNS_PREFIX)) {
url = changePrefixFromBlobtoDfs(url);
}
final AbfsRestOperation op = new AbfsRestOperation(
AbfsRestOperationType.SetPermissions,
this,
Expand Down