Skip to content

Commit 507086a

Browse files
bhattmanish98bhattmanish98
authored
HADOOP-19450: [ABFS] Rename/Create path idempotency client-level resolution (#7364)
Contributed by Manish Bhatt
1 parent e614453 commit 507086a

18 files changed

+731
-33
lines changed

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -440,6 +440,10 @@ public class AbfsConfiguration{
440440
DefaultValue = DEFAULT_HTTP_CLIENT_CONN_MAX_IDLE_TIME)
441441
private long maxApacheHttpClientConnectionIdleTime;
442442

443+
@BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID,
444+
DefaultValue = DEFAULT_FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID)
445+
private boolean enableClientTransactionId;
446+
443447
private String clientProvidedEncryptionKey;
444448
private String clientProvidedEncryptionKeySHA;
445449

@@ -1077,6 +1081,10 @@ public long getMaxApacheHttpClientConnectionIdleTime() {
10771081
return maxApacheHttpClientConnectionIdleTime;
10781082
}
10791083

1084+
public boolean getIsClientTransactionIdEnabled() {
1085+
return enableClientTransactionId;
1086+
}
1087+
10801088
/**
10811089
* Enum config to allow user to pick format of x-ms-client-request-id header
10821090
* @return tracingContextFormat config if valid, else default ALL_ID_FORMAT

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import org.slf4j.Logger;
5555
import org.slf4j.LoggerFactory;
5656

57+
import org.apache.commons.lang3.StringUtils;
5758
import org.apache.hadoop.classification.InterfaceAudience;
5859
import org.apache.hadoop.classification.InterfaceStability;
5960
import org.apache.hadoop.classification.VisibleForTesting;
@@ -147,7 +148,6 @@
147148
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_STAR;
148149
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_UNDERSCORE;
149150
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DIRECTORY;
150-
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
151151
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FILE;
152152
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
153153
import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.SINGLE_WHITE_SPACE;
@@ -1876,7 +1876,7 @@ private long extractContentLength(AbfsHttpOperation op) {
18761876
long contentLength;
18771877
String contentLengthHeader = op.getResponseHeader(
18781878
HttpHeaderConfigurations.CONTENT_LENGTH);
1879-
if (!contentLengthHeader.equals(EMPTY_STRING)) {
1879+
if (!StringUtils.isEmpty(contentLengthHeader)) {
18801880
contentLength = Long.parseLong(contentLengthHeader);
18811881
} else {
18821882
contentLength = 0;
@@ -2161,6 +2161,11 @@ void setClient(AbfsClient client) {
21612161
this.client = client;
21622162
}
21632163

2164+
@VisibleForTesting
2165+
void setClientHandler(AbfsClientHandler clientHandler) {
2166+
this.clientHandler = clientHandler;
2167+
}
2168+
21642169
@VisibleForTesting
21652170
DataBlocks.BlockFactory getBlockFactory() {
21662171
return blockFactory;

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -184,7 +184,8 @@ public enum ApiVersion {
184184

185185
DEC_12_2019("2019-12-12"),
186186
APR_10_2021("2021-04-10"),
187-
AUG_03_2023("2023-08-03");
187+
AUG_03_2023("2023-08-03"),
188+
NOV_04_2024("2024-11-04");
188189

189190
private final String xMsApiVersion;
190191

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -383,6 +383,8 @@ public static String accountProperty(String property, String account) {
383383
public static final String FS_AZURE_BLOB_DIR_RENAME_MAX_THREAD = "fs.azure.blob.dir.rename.max.thread";
384384
/**Maximum number of thread per blob-delete orchestration: {@value}*/
385385
public static final String FS_AZURE_BLOB_DIR_DELETE_MAX_THREAD = "fs.azure.blob.dir.delete.max.thread";
386+
/**Flag to enable/disable sending client transactional ID during create/rename operations: {@value}*/
387+
public static final String FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID = "fs.azure.enable.client.transaction.id";
386388

387389
private ConfigurationKeys() {}
388390
}

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,5 +198,7 @@ public final class FileSystemConfigurations {
198198

199199
public static final int DEFAULT_FS_AZURE_BLOB_DELETE_THREAD = DEFAULT_FS_AZURE_LISTING_ACTION_THREADS;
200200

201+
public static final boolean DEFAULT_FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID = false;
202+
201203
private FileSystemConfigurations() {}
202204
}

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/HttpHeaderConfigurations.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,5 +132,11 @@ public final class HttpHeaderConfigurations {
132132
*/
133133
public static final String X_MS_COPY_STATUS = "x-ms-copy-status";
134134

135+
/**
136+
* Http Request Header for create rename idempotence.
137+
* {@value}
138+
*/
139+
public static final String X_MS_CLIENT_TRANSACTION_ID = "x-ms-client-transaction-id";
140+
135141
private HttpHeaderConfigurations() {}
136142
}

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/exceptions/AbfsDriverException.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,4 +51,12 @@ public AbfsDriverException(final Exception innerException, final String activity
5151
: ERROR_MESSAGE + ", rId: " + activityId,
5252
null);
5353
}
54+
55+
public AbfsDriverException(final String errorMessage, final Exception innerException) {
56+
super(
57+
AzureServiceErrorCode.UNKNOWN.getStatusCode(),
58+
AzureServiceErrorCode.UNKNOWN.getErrorCode(),
59+
errorMessage,
60+
innerException);
61+
}
5462
}

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlobClient.java‎

Lines changed: 6 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -814,10 +814,9 @@ destination, sourceEtag, isAtomicRenameKey(source), tracingContext
814814
final URL url = createRequestUrl(destination,
815815
abfsUriQueryBuilder.toString());
816816
final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
817-
final AbfsRestOperation successOp = getAbfsRestOperation(
817+
final AbfsRestOperation successOp = getSuccessOp(
818818
AbfsRestOperationType.RenamePath, HTTP_METHOD_PUT,
819819
url, requestHeaders);
820-
successOp.hardSetResult(HTTP_OK);
821820
return new AbfsClientRenameResult(successOp, true, false);
822821
} else {
823822
throw new AbfsRestOperationException(HTTP_INTERNAL_ERROR,
@@ -1208,9 +1207,9 @@ public AbfsRestOperation getPathStatus(final String path,
12081207
if (op.getResult().getStatusCode() == HTTP_NOT_FOUND
12091208
&& isImplicitCheckRequired && isNonEmptyDirectory(path, tracingContext)) {
12101209
// Implicit path found.
1211-
AbfsRestOperation successOp = getAbfsRestOperation(
1212-
AbfsRestOperationType.GetPathStatus,
1213-
HTTP_METHOD_HEAD, url, requestHeaders);
1210+
AbfsRestOperation successOp = getSuccessOp(
1211+
AbfsRestOperationType.GetPathStatus, HTTP_METHOD_HEAD,
1212+
url, requestHeaders);
12141213
successOp.hardSetGetFileStatusResult(HTTP_OK);
12151214
return successOp;
12161215
}
@@ -1308,11 +1307,8 @@ public AbfsRestOperation deletePath(final String path,
13081307
= createDefaultUriQueryBuilder();
13091308
final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
13101309
final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
1311-
final AbfsRestOperation successOp = getAbfsRestOperation(
1312-
AbfsRestOperationType.DeletePath, HTTP_METHOD_DELETE,
1313-
url, requestHeaders);
1314-
successOp.hardSetResult(HTTP_OK);
1315-
return successOp;
1310+
return getSuccessOp(AbfsRestOperationType.DeletePath,
1311+
HTTP_METHOD_DELETE, url, requestHeaders);
13161312
} else {
13171313
throw new AbfsRestOperationException(HTTP_INTERNAL_ERROR,
13181314
AzureServiceErrorCode.UNKNOWN.getErrorCode(),

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java‎

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -216,6 +216,11 @@ private AbfsClient(final URL baseUrl,
216216
encryptionType = EncryptionType.GLOBAL_KEY;
217217
}
218218

219+
// Version update needed to support x-ms-client-transaction-id header
220+
if (abfsConfiguration.getIsClientTransactionIdEnabled()) {
221+
xMsVersion = ApiVersion.NOV_04_2024;
222+
}
223+
219224
String sslProviderName = null;
220225

221226
if (this.baseUrl.toString().startsWith(HTTPS_SCHEME)) {
@@ -938,14 +943,12 @@ public AbfsRestOperation deleteIdempotencyCheckOp(final AbfsRestOperation op) {
938943
&& DEFAULT_DELETE_CONSIDERED_IDEMPOTENT) {
939944
// Server has returned HTTP 404, which means path no longer
940945
// exists. Assuming delete result to be idempotent, return success.
941-
final AbfsRestOperation successOp = getAbfsRestOperation(
946+
LOG.debug("Returning success response from delete idempotency logic");
947+
return getSuccessOp(
942948
AbfsRestOperationType.DeletePath,
943949
HTTP_METHOD_DELETE,
944950
op.getUrl(),
945951
op.getRequestHeaders());
946-
successOp.hardSetResult(HttpURLConnection.HTTP_OK);
947-
LOG.debug("Returning success response from delete idempotency logic");
948-
return successOp;
949952
}
950953

951954
return op;
@@ -1733,4 +1736,21 @@ public abstract Hashtable<String, String> getXMSProperties(AbfsHttpOperation res
17331736
* @throws UnsupportedEncodingException if decoding fails
17341737
*/
17351738
public abstract String decodeAttribute(byte[] value) throws UnsupportedEncodingException;
1739+
1740+
/**
1741+
* Get the dummy success operation.
1742+
* @param operationType type of the operation
1743+
* @param httpMethod http method
1744+
* @param url url to be used
1745+
* @param requestHeaders list of headers to be sent with the request
1746+
* @return success operation
1747+
*/
1748+
protected AbfsRestOperation getSuccessOp(final AbfsRestOperationType operationType,
1749+
final String httpMethod, final URL url,
1750+
final List<AbfsHttpHeader> requestHeaders) {
1751+
final AbfsRestOperation successOp = getAbfsRestOperation(
1752+
operationType, httpMethod, url, requestHeaders);
1753+
successOp.hardSetResult(HttpURLConnection.HTTP_OK);
1754+
return successOp;
1755+
}
17361756
}

‎hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsDfsClient.java‎

Lines changed: 81 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import com.fasterxml.jackson.core.JsonToken;
4141
import com.fasterxml.jackson.databind.ObjectMapper;
4242

43+
import org.apache.hadoop.classification.VisibleForTesting;
4344
import org.apache.hadoop.fs.FileSystem;
4445
import org.apache.hadoop.fs.Path;
4546
import org.apache.hadoop.fs.FileAlreadyExistsException;
@@ -49,6 +50,7 @@
4950
import org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ApiVersion;
5051
import org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations;
5152
import org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams;
53+
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
5254
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsInvalidChecksumException;
5355
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
5456
import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
@@ -109,6 +111,7 @@
109111
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.RANGE;
110112
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.USER_AGENT;
111113
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_HTTP_METHOD_OVERRIDE;
114+
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_CLIENT_TRANSACTION_ID;
112115
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_EXISTING_RESOURCE_TYPE;
113116
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_LEASE_ACTION;
114117
import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_LEASE_BREAK_PERIOD;
@@ -134,7 +137,9 @@
134137
import static org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.RENAME_DESTINATION_PARENT_PATH_NOT_FOUND;
135138
import static org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.SOURCE_PATH_NOT_FOUND;
136139
import static org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.UNAUTHORIZED_BLOB_OVERWRITE;
140+
import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_CREATE_RECOVERY;
137141
import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_FILE_ALREADY_EXISTS;
142+
import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_RENAME_RECOVERY;
138143

139144
/**
140145
* AbfsClient interacting with the DFS Endpoint.
@@ -383,6 +388,9 @@ public AbfsRestOperation createPath(final String path,
383388
requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
384389
}
385390

391+
// Add the client transaction ID to the request headers.
392+
String clientTransactionId = addClientTransactionIdToHeader(requestHeaders);
393+
386394
final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
387395
abfsUriQueryBuilder.addQuery(QUERY_PARAM_RESOURCE, isFile ? FILE : DIRECTORY);
388396
if (isAppendBlob) {
@@ -405,11 +413,34 @@ public AbfsRestOperation createPath(final String path,
405413
if (!op.hasResult()) {
406414
throw ex;
407415
}
408-
if (!isFile && op.getResult().getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
409-
String existingResource =
410-
op.getResult().getResponseHeader(X_MS_EXISTING_RESOURCE_TYPE);
411-
if (existingResource != null && existingResource.equals(DIRECTORY)) {
412-
return op; //don't throw ex on mkdirs for existing directory
416+
if (!isFile) {
417+
if (op.getResult().getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
418+
String existingResource =
419+
op.getResult().getResponseHeader(X_MS_EXISTING_RESOURCE_TYPE);
420+
if (existingResource != null && existingResource.equals(DIRECTORY)) {
421+
//don't throw ex on mkdirs for existing directory
422+
return getSuccessOp(AbfsRestOperationType.CreatePath,
423+
HTTP_METHOD_PUT, url, requestHeaders);
424+
}
425+
}
426+
} else {
427+
// recovery using client transaction id only if it is a retried request.
428+
if (op.isARetriedRequest() && clientTransactionId != null
429+
&& (op.getResult().getStatusCode() == HttpURLConnection.HTTP_CONFLICT
430+
|| op.getResult().getStatusCode() == HttpURLConnection.HTTP_PRECON_FAILED)) {
431+
try {
432+
final AbfsHttpOperation getPathStatusOp =
433+
getPathStatus(path, false,
434+
tracingContext, null).getResult();
435+
if (clientTransactionId.equals(
436+
getPathStatusOp.getResponseHeader(
437+
X_MS_CLIENT_TRANSACTION_ID))) {
438+
return getSuccessOp(AbfsRestOperationType.CreatePath,
439+
HTTP_METHOD_PUT, url, requestHeaders);
440+
}
441+
} catch (AzureBlobFileSystemException exception) {
442+
throw new AbfsDriverException(ERR_CREATE_RECOVERY, exception);
443+
}
413444
}
414445
}
415446
throw ex;
@@ -681,6 +712,9 @@ public AbfsClientRenameResult renamePath(
681712
requestHeaders.add(new AbfsHttpHeader(X_MS_RENAME_SOURCE, encodedRenameSource));
682713
requestHeaders.add(new AbfsHttpHeader(IF_NONE_MATCH, STAR));
683714

715+
// Add the client transaction ID to the request headers.
716+
String clientTransactionId = addClientTransactionIdToHeader(requestHeaders);
717+
684718
final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
685719
abfsUriQueryBuilder.addQuery(QUERY_PARAM_CONTINUATION, continuation);
686720
appendSASTokenToQuery(destination,
@@ -705,11 +739,33 @@ public AbfsClientRenameResult renamePath(
705739
throw e;
706740
}
707741

742+
// recovery using client transaction id only if it is a retried request.
743+
if (op.isARetriedRequest() && clientTransactionId != null
744+
&& SOURCE_PATH_NOT_FOUND.getErrorCode().equalsIgnoreCase(
745+
op.getResult().getStorageErrorCode())) {
746+
try {
747+
final AbfsHttpOperation abfsHttpOperation =
748+
getPathStatus(destination, false,
749+
tracingContext, null).getResult();
750+
if (clientTransactionId.equals(
751+
abfsHttpOperation.getResponseHeader(
752+
X_MS_CLIENT_TRANSACTION_ID))) {
753+
return new AbfsClientRenameResult(
754+
getSuccessOp(AbfsRestOperationType.RenamePath,
755+
HTTP_METHOD_PUT, url, requestHeaders), true,
756+
isMetadataIncompleteState);
757+
}
758+
} catch (AzureBlobFileSystemException exception) {
759+
throw new AbfsDriverException(ERR_RENAME_RECOVERY, exception);
760+
}
761+
throw e;
762+
}
763+
708764
// ref: HADOOP-19393. Write permission checks can occur before validating
709765
// rename operation's validity. If there is an existing destination path, it may be rejected
710766
// with an authorization error. Catching and throwing FileAlreadyExistsException instead.
711767
if (op.getResult().getStorageErrorCode()
712-
.equals(UNAUTHORIZED_BLOB_OVERWRITE.getErrorCode())){
768+
.equals(UNAUTHORIZED_BLOB_OVERWRITE.getErrorCode())) {
713769
throw new FileAlreadyExistsException(ERR_FILE_ALREADY_EXISTS);
714770
}
715771

@@ -1589,4 +1645,23 @@ private Hashtable<String, String> parseCommaSeparatedXmsProperties(String xMsPro
15891645

15901646
return properties;
15911647
}
1648+
1649+
/**
1650+
* Add the client transaction id to the request header
1651+
* if {@link AbfsConfiguration#getIsClientTransactionIdEnabled()} is enabled.
1652+
* @param requestHeaders list of headers to be sent with the request
1653+
*
1654+
* @return client transaction id
1655+
*/
1656+
@VisibleForTesting
1657+
public String addClientTransactionIdToHeader(List<AbfsHttpHeader> requestHeaders) {
1658+
String clientTransactionId = null;
1659+
// Set client transaction ID if the namespace and client transaction ID config are enabled.
1660+
if (getIsNamespaceEnabled() && getAbfsConfiguration().getIsClientTransactionIdEnabled()) {
1661+
clientTransactionId = UUID.randomUUID().toString();
1662+
requestHeaders.add(
1663+
new AbfsHttpHeader(X_MS_CLIENT_TRANSACTION_ID, clientTransactionId));
1664+
}
1665+
return clientTransactionId;
1666+
}
15921667
}

0 commit comments

Comments
 (0)