HADOOP-19233: ABFS: [FnsOverBlob] Implementing Rename and Delete APIs over Blob Endpoint #7265
Conversation
bhattmanish98
changed the title
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
bhattmanish98
changed the title
This comment was marked as outdated.
This comment was marked as outdated.
| AUTHORIZATION_PERMISSION_MISS_MATCH("AuthorizationPermissionMismatch", HttpURLConnection.HTTP_FORBIDDEN, null), | ||
| ACCOUNT_REQUIRES_HTTPS("AccountRequiresHttps", HttpURLConnection.HTTP_BAD_REQUEST, null), | ||
| MD5_MISMATCH("Md5Mismatch", HttpURLConnection.HTTP_BAD_REQUEST, | ||
| "The MD5 value specified in the request did not match with the MD5 value calculated by the server."), | ||
| COPY_BLOB_FAILED("COPY_BLOB_FAILED", HttpURLConnection.HTTP_INTERNAL_ERROR, null), |
There was a problem hiding this comment.
Error Codes should be in camelcase as others
|
|
||
| package org.apache.hadoop.fs.azurebfs.enums; | ||
|
|
||
| public enum BlobCopyProgress { |
There was a problem hiding this comment.
Javadoc for class and enums
There was a problem hiding this comment.
Java Doc added.
| @@ -49,6 +49,9 @@ public interface SASTokenProvider { | |||
| String SET_PERMISSION_OPERATION = "set-permission"; | |||
| String SET_PROPERTIES_OPERATION = "set-properties"; | |||
| String WRITE_OPERATION = "write"; | |||
| String COPY_BLOB_DESTINATION = "copy-blob-dst"; | |||
There was a problem hiding this comment.
Need to discuss this change once, we do not support UDS for FNS Blob
There was a problem hiding this comment.
It was not in use anywhere, so removed it for now.
| destination, sourceEtag, isAtomicRenameKey(source), tracingContext | ||
| ); | ||
| incrementAbfsRenamePath(); | ||
| return blobRenameHandler.execute(); |
There was a problem hiding this comment.
This might need rechecking. We do not want to return op as null.
There was a problem hiding this comment.
As per the offline discussion, made the changes.
| final TracingContext tracingContext, | ||
| final boolean isNamespaceEnabled) throws AzureBlobFileSystemException { | ||
| getBlobDeleteHandler(path, recursive, tracingContext).execute(); | ||
| return null; |
There was a problem hiding this comment.
Why are we returning null here?
There was a problem hiding this comment.
Done the changes to return dummy response same as rename Path.
| @@ -201,4 +238,25 @@ public int getAcquireRetryCount() { | |||
| public TracingContext getTracingContext() { | |||
| return tracingContext; | |||
| } | |||
|
|
|||
There was a problem hiding this comment.
Java docs for all public methods and classes
There was a problem hiding this comment.
Java doc added.
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.hadoop.fs.azurebfs.services; |
There was a problem hiding this comment.
Java doc added.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
|
🎊 +1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
| * Blob copy API is an async API, this configuration defines polling duration | ||
| * for checking copy status {@value} | ||
| */ | ||
| public static final String FS_AZURE_BLOB_COPY_PROGRESS_WAIT_MILLIS = "fs.azure.blob.copy.progress.wait.millis"; |
There was a problem hiding this comment.
keep the comments formatting constant
| @@ -104,5 +104,9 @@ public final class HttpHeaderConfigurations { | |||
| */ | |||
| public static final String X_MS_BLOB_CONTENT_MD5 = "x-ms-blob-content-md5"; | |||
|
|
|||
| public static final String X_MS_COPY_ID = "x-ms-copy-id"; | |||
There was a problem hiding this comment.
javadocs for new constants
| @@ -359,6 +375,34 @@ public AbfsRestOperation listPath(final String relativePath, final boolean recur | |||
| return op; | |||
| } | |||
|
|
|||
| private void fixAtomicEntriesInListResults(final AbfsRestOperation op, | |||
There was a problem hiding this comment.
add javadocs for all methods wherever missing
| @@ -338,6 +353,7 @@ public AbfsRestOperation listPath(final String relativePath, final boolean recur | |||
| requestHeaders); | |||
|
|
|||
| op.execute(tracingContext); | |||
| fixAtomicEntriesInListResults(op, tracingContext); | |||
There was a problem hiding this comment.
explain the need for the same
| List<BlobListResultEntrySchema> filteredEntries = new ArrayList<>(); | ||
| for (BlobListResultEntrySchema entry : listResultSchema.paths()) { | ||
| if (!takeListPathAtomicRenameKeyAction(entry.path(), | ||
| (int) (long) entry.contentLength(), tracingContext)) { |
There was a problem hiding this comment.
double casting is not needed, we can use something like :-
Long longValue = 12345L;
int intValue = longValue.intValue();
| @@ -45,6 +45,8 @@ public enum AzureServiceErrorCode { | |||
| INVALID_SOURCE_OR_DESTINATION_RESOURCE_TYPE("InvalidSourceOrDestinationResourceType", HttpURLConnection.HTTP_CONFLICT, null), | |||
| RENAME_DESTINATION_PARENT_PATH_NOT_FOUND("RenameDestinationParentPathNotFound", HttpURLConnection.HTTP_NOT_FOUND, null), | |||
| INVALID_RENAME_SOURCE_PATH("InvalidRenameSourcePath", HttpURLConnection.HTTP_CONFLICT, null), | |||
| DIRECTORY_NOT_EMPTY_DELETE("DirectoryNotEmpty", HttpURLConnection.HTTP_CONFLICT, | |||
There was a problem hiding this comment.
naming could be changed
| } else { | ||
| throw new AbfsRestOperationException(HTTP_INTERNAL_ERROR, | ||
| AzureServiceErrorCode.UNKNOWN.getErrorCode(), | ||
| "FNS-Blob Rename was not successfull", |
There was a problem hiding this comment.
Should add path or file name as well where rename failed
| throw ex; | ||
| } | ||
| } | ||
| return true; |
There was a problem hiding this comment.
why are we directly returning true here and not checking for renameSrchasChanged as in the previous method ?
There was a problem hiding this comment.
In the above if condition we are filtering out paths for which we don't have to do the rename redo and returning false there itself. In case where path require rename redo, we are returning true if no exception is raised.
|
|
||
| /** | ||
| * Orchestrator for delete over Blob endpoint. Blob endpoint for flat-namespace | ||
| * account does not support director delete. This class is responsible for |
| try { | ||
| /* | ||
| * Delete the required path. | ||
| * Directory needs to be safely delete the path, as the path can be implicit. |
There was a problem hiding this comment.
typo: grammar issue
| this.isAtomicRenameRecovery = isAtomicRenameRecovery; | ||
| } | ||
|
|
||
| public BlobRenameHandler(final String src, |
| * @return true if the path contains a colon | ||
| */ | ||
| private boolean containsColon(Path p) { | ||
| return p.toUri().getPath().contains(":"); |
There was a problem hiding this comment.
COLON constant can be used
| @Override | ||
| boolean takeAction(final Path path) throws AzureBlobFileSystemException { | ||
| return renameInternal(path, | ||
| createDestinationPathForBlobPartOfRenameSrcDir(dst, path, src)); |
There was a problem hiding this comment.
Method name can be shortened
| try { | ||
| AbfsRestOperation copyPathOp = getAbfsClient().copyBlob(src, dst, leaseId, | ||
| tracingContext); | ||
| final String progress = copyPathOp.getResult() |
There was a problem hiding this comment.
check for copyPathOp != null && copyPathOp.getResult() != null
| tracingContext, null, false); | ||
| final String srcCopyPath = ROOT_PATH + getAbfsClient().getFileSystem() | ||
| + src.toUri().getPath(); | ||
| if (dstPathStatus.getResult() != null && (srcCopyPath.equals( |
There was a problem hiding this comment.
should have check for dstPathStatus != null as well
| } | ||
| final long pollWait = getAbfsClient().getAbfsConfiguration() | ||
| .getBlobCopyProgressPollWaitMillis(); | ||
| while (handleCopyInProgress(dst, tracingContext, copyId) |
There was a problem hiding this comment.
check copyId != null
There was a problem hiding this comment.
If handleCopyInProgress keeps returning PENDING, the code might enter an infinite loop of waiting. We should introduce maximum wait time and if exceeded fail.
|
|
||
| if (op.getResult() != null && copyId.equals( | ||
| op.getResult().getResponseHeader(X_MS_COPY_ID))) { | ||
| final String copyStatus = op.getResult() |
There was a problem hiding this comment.
check for response header is not null is needed
| getAbfsClient().checkIsDir(op.getResult()), | ||
| extractEtagHeader(op.getResult()), | ||
| op.getResult() instanceof AbfsHttpOperation.AbfsHttpOperationWithFixedResultForGetFileStatus); | ||
| } catch (AzureBlobFileSystemException e) { |
There was a problem hiding this comment.
better to catch AbfsRestOperationException itself ?
| import java.nio.charset.Charset; | ||
| import java.nio.charset.StandardCharsets; | ||
|
|
||
| //import java.util.ArrayList; |
There was a problem hiding this comment.
remove unused imports
There was a problem hiding this comment.
This will go once Ingress handler changes are taken.
| renamePendingJsonFormatObj = objectMapper.readValue(contents, | ||
| RenamePendingJsonFormat.class); | ||
| } catch (JsonProcessingException e) { | ||
| return; |
There was a problem hiding this comment.
Is it right to just return without throwing an exception ?
| abfsClient.append(path.toUri().getPath(), bytes, | ||
| appendRequestParameters, null, null, tracingContext); | ||
|
|
||
| // List<String> blockIdList = new ArrayList<>(Collections.singleton(blockId)); |
There was a problem hiding this comment.
Remove commented code
There was a problem hiding this comment.
This will go once Ingress handler changes are taken.
| // List<String> blockIdList = new ArrayList<>(Collections.singleton(blockId)); | ||
| // String blockList = generateBlockListXml(blockIdList); | ||
| // PutBlockList on the path. | ||
| String blockList = ""; |
There was a problem hiding this comment.
if flush is called on empty string, how does it take the blockId into usage ?
There was a problem hiding this comment.
Added this line just to pass the build. The above commented line is calling generateBlockListXml which requires ingress handler changes. Will pick the changes of ingress handler once it is merged and this line is no longer needed after that.
| * endpoint, the orchestration would be done by the client. The idempotency | ||
| * issue would not happen for blob endpoint. | ||
| */ | ||
| assertTrue(fs.getAbfsClient() instanceof AbfsDfsClient); |
There was a problem hiding this comment.
can use getAbfsServiceType()
| @@ -314,4 +347,278 @@ public void deleteBlobDirParallelThreadToDeleteOnDifferentTracingContext() | |||
| fs.delete(new Path("/testDir"), true); | |||
| fs.close(); | |||
| } | |||
|
|
|||
| private void assumeBlobClient() throws IOException { | |||
There was a problem hiding this comment.
[optional] This method can be avoided and all calls to it can be replaced by getAbfsServiceType() == AbfsServiceType.BLOB
Description of PR:
This PR is in correlation to the series of work done under Parent Jira: [HADOOP-19179]
Jira for this Patch: [HADOOP-19233]
Currently, we only support rename and delete operations on the DFS endpoint. The reason for not supporting rename and delete operations on the Blob endpoint is that the Blob endpoint does not account for hierarchy. We need to ensure that the HDFS contracts are maintained when performing rename and delete operations. Renaming or deleting a directory over the Blob endpoint requires the client to handle the orchestration and rename or delete all the blobs within the specified directory.
The task outlines the considerations for implementing rename and delete operations for the FNS-blob endpoint to ensure compatibility with HDFS contracts.