[
https://issues.apache.org/jira/browse/HADOOP-19233?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17913702#comment-17913702
]
ASF GitHub Bot commented on HADOOP-19233:
-----------------------------------------
bhattmanish98 commented on code in PR #7265:
URL: https://github.com/apache/hadoop/pull/7265#discussion_r1918225101
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlobClient.java:
##########
@@ -359,6 +375,34 @@ public AbfsRestOperation listPath(final String
relativePath, final boolean recur
return op;
}
+ private void fixAtomicEntriesInListResults(final AbfsRestOperation op,
+ final TracingContext
tracingContext) throws AzureBlobFileSystemException {
+ /*
+ * Crashed HBase log rename recovery is done by Filesystem.getFileStatus
and
+ * Filesystem.listStatus.
+ */
+ if (tracingContext == null
+ || tracingContext.getOpType() != FSOperationType.LISTSTATUS
+ || op == null || op.getResult() == null
+ || op.getResult().getStatusCode() != HTTP_OK) {
+ return;
+ }
+ BlobListResultSchema listResultSchema
+ = (BlobListResultSchema) op.getResult().getListResultSchema();
+ if (listResultSchema == null) {
+ return;
+ }
+ List<BlobListResultEntrySchema> filteredEntries = new ArrayList<>();
+ for (BlobListResultEntrySchema entry : listResultSchema.paths()) {
+ if (!takeListPathAtomicRenameKeyAction(entry.path(),
+ (int) (long) entry.contentLength(), tracingContext)) {
Review Comment:
Taken!
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlobClient.java:
##########
@@ -499,21 +556,50 @@ public AbfsRestOperation breakLease(final String path,
* @param isMetadataIncompleteState was there a rename failure due to
* incomplete metadata state?
* @param isNamespaceEnabled whether namespace enabled account or not
- * @return result of rename operation
- * @throws IOException if rename operation fails.
+ *
+ * @return AbfsClientRenameResult result of rename operation indicating the
+ * AbfsRest operation, rename recovery and incomplete metadata state failure.
+ *
+ * @throws IOException failure, excluding any recovery from overload
failures.
*/
@Override
public AbfsClientRenameResult renamePath(final String source,
- final String destination,
- final String continuation,
- final TracingContext tracingContext,
- final String sourceEtag,
- final boolean isMetadataIncompleteState,
- final boolean isNamespaceEnabled) throws IOException {
- /**
- * TODO: [FnsOverBlob] To be implemented as part of rename-delete over
blob endpoint work. <a
href="https://issues.apache.org/jira/browse/HADOOP-19233";>HADOOP-19233</a>.
- */
- throw new NotImplementedException("Rename operation on Blob endpoint yet
to be implemented.");
+ final String destination,
+ final String continuation,
+ final TracingContext tracingContext,
+ String sourceEtag,
+ boolean isMetadataIncompleteState,
+ boolean isNamespaceEnabled)
+ throws IOException {
+ BlobRenameHandler blobRenameHandler = getBlobRenameHandler(source,
+ destination, sourceEtag, isAtomicRenameKey(source), tracingContext
+ );
+ incrementAbfsRenamePath();
+ if (blobRenameHandler.execute()) {
+ final AbfsUriQueryBuilder abfsUriQueryBuilder =
createDefaultUriQueryBuilder();
+ final URL url = createRequestUrl(destination,
abfsUriQueryBuilder.toString());
+ final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
+ final AbfsRestOperation successOp = getAbfsRestOperation(
+ AbfsRestOperationType.RenamePath, HTTP_METHOD_PUT,
+ url, requestHeaders);
+ successOp.hardSetResult(HttpURLConnection.HTTP_OK);
+ return new AbfsClientRenameResult(successOp, true, false);
+ } else {
+ throw new AbfsRestOperationException(HTTP_INTERNAL_ERROR,
+ AzureServiceErrorCode.UNKNOWN.getErrorCode(),
+ "FNS-Blob Rename was not successfull",
Review Comment:
Taken!
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/BlobDeleteHandler.java:
##########
@@ -0,0 +1,201 @@
+/**
+ * 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.azurebfs.services;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathIOException;
+import
org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
+import
org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+
+import static java.net.HttpURLConnection.HTTP_CONFLICT;
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.DIRECTORY_NOT_EMPTY_DELETE;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.PATH_NOT_FOUND;
+
+/**
+ * Orchestrator for delete over Blob endpoint. Blob endpoint for flat-namespace
+ * account does not support director delete. This class is responsible for
Review Comment:
Resolved!
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/BlobRenameHandler.java:
##########
@@ -0,0 +1,652 @@
+/**
+ * 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.azurebfs.services;
+
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathIOException;
+import
org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
+import
org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
+import org.apache.hadoop.fs.azurebfs.enums.BlobCopyProgress;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+
+import static
org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.extractEtagHeader;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.COPY_STATUS_ABORTED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.COPY_STATUS_FAILED;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.COPY_STATUS_SUCCESS;
+import static
org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_COPY_ID;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_COPY_SOURCE;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_COPY_STATUS;
+import static
org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_COPY_STATUS_DESCRIPTION;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.COPY_BLOB_ABORTED;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.COPY_BLOB_FAILED;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.RENAME_DESTINATION_PARENT_PATH_NOT_FOUND;
+
+/**
+ * Orchestrator for rename over Blob endpoint. Handles both directory and file
+ * renames. Blob Endpoint does not expose rename API, this class is responsible
+ * for copying the blobs and deleting the source blobs.
+ * <p>
+ * For directory rename, it recursively lists the blobs in the source
directory and
+ * copies them to the destination directory.
+ */
+public class BlobRenameHandler extends ListActionTaker {
+
+ public static final Logger LOG = LoggerFactory.getLogger(AbfsClient.class);
+
+ private final String srcEtag;
+
+ private final Path src, dst;
+
+ private final boolean isAtomicRename, isAtomicRenameRecovery;
+
+ private final TracingContext tracingContext;
+
+ private AbfsLease srcAbfsLease;
+
+ private String srcLeaseId;
+
+ private final List<AbfsLease> leases = new ArrayList<>();
+
+ private final AtomicInteger operatedBlobCount = new AtomicInteger(0);
+
+ /** Constructor.
+ *
+ * @param src source path
+ * @param dst destination path
+ * @param abfsClient AbfsBlobClient to use for the rename operation
+ * @param srcEtag eTag of the source path
+ * @param isAtomicRename true if the rename operation is atomic
+ * @param isAtomicRenameRecovery true if the rename operation is a
recovery of a previous failed atomic rename operation
+ * @param tracingContext object of tracingContext used for the tracing of
the server calls.
+ */
+ public BlobRenameHandler(final String src,
+ final String dst,
+ final AbfsBlobClient abfsClient,
+ final String srcEtag,
+ final boolean isAtomicRename,
+ final boolean isAtomicRenameRecovery,
+ final TracingContext tracingContext) {
+ super(new Path(src), abfsClient, tracingContext);
+ this.srcEtag = srcEtag;
+ this.tracingContext = tracingContext;
+ this.src = new Path(src);
+ this.dst = new Path(dst);
+ this.isAtomicRename = isAtomicRename;
+ this.isAtomicRenameRecovery = isAtomicRenameRecovery;
+ }
+
+ public BlobRenameHandler(final String src,
+ final String dst,
+ final AbfsBlobClient abfsClient,
+ final String srcEtag,
+ final boolean isAtomicRename,
+ final boolean isAtomicRenameRecovery,
+ final AbfsLease srcAbfsLease,
+ final TracingContext tracingContext) {
+ this(src, dst, abfsClient, srcEtag, isAtomicRename,
isAtomicRenameRecovery,
+ tracingContext);
+ this.srcAbfsLease = srcAbfsLease;
+ }
+
+ @Override
+ int getMaxConsumptionParallelism() {
+ return getAbfsClient().getAbfsConfiguration()
+ .getBlobRenameDirConsumptionParallelism();
+ }
+
+ /**
+ * Orchestrates the rename operation.
+ *
+ * @return AbfsClientRenameResult containing the result of the rename
operation
+ * @throws AzureBlobFileSystemException if server call fails
+ */
+ public boolean execute() throws AzureBlobFileSystemException {
+ PathInformation pathInformation = new PathInformation();
+ boolean result = false;
+ if (preCheck(src, dst, pathInformation)) {
+ RenameAtomicity renameAtomicity = null;
+ if (pathInformation.getIsDirectory()
+ && pathInformation.getIsImplicit()) {
+ AbfsRestOperation createMarkerOp =
getAbfsClient().createPath(src.toUri().getPath(),
+ false, false, null,
+ false, null, null, tracingContext);
+
pathInformation.setETag(extractEtagHeader(createMarkerOp.getResult()));
+ }
+ try {
+ if (isAtomicRename) {
+ /*
+ * Conditionally get a lease on the source blob to prevent
other writers
+ * from changing it. This is used for correctness in HBase
when log files
+ * are renamed. When the HBase master renames a log file
folder, the lease
+ * locks out other writers. This prevents a region server
that the master
+ * thinks is dead, but is still alive, from committing
additional updates.
+ * This is different than when HBase runs on HDFS, where
the region server
+ * recovers the lease on a log file, to gain exclusive
access to it, before
+ * it splits it.
+ */
+ if (srcAbfsLease == null) {
+ srcAbfsLease = takeLease(src, srcEtag);
+ }
+ srcLeaseId = srcAbfsLease.getLeaseID();
+ if (!isAtomicRenameRecovery &&
pathInformation.getIsDirectory()) {
+ /*
+ * if it is not a resume of a previous failed atomic
rename operation,
+ * perform the pre-rename operation.
+ */
+ renameAtomicity = getRenameAtomicity(pathInformation);
+ renameAtomicity.preRename();
+ }
+ }
+ if (pathInformation.getIsDirectory()) {
+ result = listRecursiveAndTakeAction() && finalSrcRename();
+ } else {
+ result = renameInternal(src, dst);
+ }
+ } finally {
+ if (srcAbfsLease != null) {
+ // If the operation is successful, cancel the timer and no
need to release
+ // the lease as delete on the blob-path has taken place.
+ if (result) {
+ srcAbfsLease.cancelTimer();
+ } else {
+ srcAbfsLease.free();
+ }
+ }
+ }
+ if (result && renameAtomicity != null) {
+ renameAtomicity.postRename();
+ }
+ }
+ return result;
+ }
+
+ /** Final rename operation after all the blobs have been copied.
+ *
+ * @return true if rename is successful
+ * @throws AzureBlobFileSystemException if server call fails
+ */
+ private boolean finalSrcRename() throws AzureBlobFileSystemException {
+ tracingContext.setOperatedBlobCount(operatedBlobCount.get() + 1);
+ try {
+ return renameInternal(src, dst);
+ } finally {
+ tracingContext.setOperatedBlobCount(null);
+ }
+ }
+
+ /** Gets the rename atomicity object.
+ *
+ * @param pathInformation object containing the path information of the
source path
+ *
+ * @return RenameAtomicity object
+ */
+ @VisibleForTesting
+ public RenameAtomicity getRenameAtomicity(final PathInformation
pathInformation) {
+ return new RenameAtomicity(src,
+ dst,
+ new Path(src.getParent(), src.getName() +
RenameAtomicity.SUFFIX),
+ tracingContext,
+ pathInformation.getETag(),
+ getAbfsClient());
+ }
+
+ /** Takes a lease on the path.
+ *
+ * @param path path on which the lease is to be taken
+ * @param eTag eTag of the path
+ *
+ * @return object containing the lease information
+ * @throws AzureBlobFileSystemException if server call fails
+ */
+ private AbfsLease takeLease(final Path path, final String eTag)
+ throws AzureBlobFileSystemException {
+ AbfsLease lease = new AbfsLease(getAbfsClient(),
path.toUri().getPath(), false,
+ getAbfsClient().getAbfsConfiguration()
+ .getAtomicRenameLeaseRefreshDuration(),
+ eTag, tracingContext);
+ leases.add(lease);
+ return lease;
+ }
+
+ /** Checks if the path contains a colon.
+ *
+ * @param p path to check
+ *
+ * @return true if the path contains a colon
+ */
+ private boolean containsColon(Path p) {
+ return p.toUri().getPath().contains(":");
Review Comment:
Taken!
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsBlobClient.java:
##########
@@ -1318,6 +1421,128 @@ public static String getDirectoryQueryParameter(final
String path) {
return directory;
}
+ public boolean isAtomicRenameKey(String key) {
+ return isKeyForDirectorySet(key, azureAtomicRenameDirSet);
+ }
+
+ /**
+ * Action to be taken when atomic-key is present on a getPathStatus path.
+ *
+ * @param path path of the pendingJson for the atomic path.
+ * @param pathLease lease on the path.
+ * @param tracingContext tracing context.
+ *
+ * @throws AzureBlobFileSystemException server error or the path is
renamePending json file and action is taken.
+ */
+ public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
+ final AbfsLease pathLease,
+ final TracingContext
tracingContext)
+ throws AzureBlobFileSystemException {
+ if (path == null || path.isRoot() ||
!isAtomicRenameKey(path.toUri().getPath())) {
+ return;
+ }
+ AbfsRestOperation pendingJsonFileStatus;
+ Path pendingJsonPath = new Path(path.getParent(),
+ path.toUri().getPath() + RenameAtomicity.SUFFIX);
+ try {
+ pendingJsonFileStatus = getPathStatus(
+ pendingJsonPath.toUri().getPath(), tracingContext, null, false);
+ if (checkIsDir(pendingJsonFileStatus.getResult())) {
+ return;
+ }
+ } catch (AbfsRestOperationException ex) {
+ if (ex.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
+ return;
+ }
+ throw ex;
+ }
+
+ boolean renameSrcHasChanged;
+ try {
+ RenameAtomicity renameAtomicity = getRedoRenameAtomicity(
+ pendingJsonPath,
Integer.parseInt(pendingJsonFileStatus.getResult()
+
.getResponseHeader(HttpHeaderConfigurations.CONTENT_LENGTH)),
+ tracingContext, pathLease);
+ renameAtomicity.redo();
+ renameSrcHasChanged = false;
+ } catch (AbfsRestOperationException ex) {
+ /*
+ * At this point, the source marked by the renamePending json file,
might have
+ * already got renamed by some parallel thread, or at this point, the
path
+ * would have got modified which would result in eTag change, which
would lead
+ * to a HTTP_CONFLICT. In this case, no more operation needs to be
taken, and
+ * the calling getPathStatus can return this source path as result.
+ */
+ if (ex.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND
+ || ex.getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
+ renameSrcHasChanged = true;
+ } else {
+ throw ex;
+ }
+ }
+ if (!renameSrcHasChanged) {
+ throw new AbfsRestOperationException(
+ AzureServiceErrorCode.PATH_NOT_FOUND.getStatusCode(),
+ AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(),
+ ATOMIC_DIR_RENAME_RECOVERY_ON_GET_PATH_EXCEPTION,
+ null);
+ }
+ }
+
+ /**
+ * Action to be taken when atomic-key is present on a listPath path.
+ *
+ * @param path path of the pendingJson for the atomic path.
+ * @param renamePendingJsonLen length of the pendingJson file.
+ * @param tracingContext tracing context.
+ *
+ * @return true if action is taken.
+ * @throws AzureBlobFileSystemException server error
+ */
+ private boolean takeListPathAtomicRenameKeyAction(final Path path,
+ final int
renamePendingJsonLen,
+ final TracingContext
tracingContext)
+ throws AzureBlobFileSystemException {
+ if (path == null || path.isRoot() || !isAtomicRenameKey(
+ path.toUri().getPath()) || !path.toUri()
+ .getPath()
+ .endsWith(RenameAtomicity.SUFFIX)) {
+ return false;
+ }
+ try {
+ RenameAtomicity renameAtomicity
+ = getRedoRenameAtomicity(path, renamePendingJsonLen,
+ tracingContext, null);
+ renameAtomicity.redo();
+ } catch (AbfsRestOperationException ex) {
+ /*
+ * At this point, the source marked by the renamePending json file,
might have
+ * already got renamed by some parallel thread, or at this point, the
path
+ * would have got modified which would result in eTag change, which
would lead
+ * to a HTTP_CONFLICT. In this case, no more operation needs to be
taken, but
+ * since this is a renamePendingJson file and would be deleted by the
redo operation,
+ * the calling listPath should not return this json path as result.
+ */
+ if (ex.getStatusCode() != HttpURLConnection.HTTP_NOT_FOUND
+ && ex.getStatusCode() != HttpURLConnection.HTTP_CONFLICT) {
+ throw ex;
+ }
+ }
+ return true;
Review 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.
##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/BlobDeleteHandler.java:
##########
@@ -0,0 +1,201 @@
+/**
+ * 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.azurebfs.services;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathIOException;
+import
org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
+import
org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+
+import static java.net.HttpURLConnection.HTTP_CONFLICT;
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.DIRECTORY_NOT_EMPTY_DELETE;
+import static
org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.PATH_NOT_FOUND;
+
+/**
+ * Orchestrator for delete over Blob endpoint. Blob endpoint for flat-namespace
+ * account does not support director delete. This class is responsible for
+ * deleting the blobs and creating the parent directory marker file if needed.
+ */
+public class BlobDeleteHandler extends ListActionTaker {
+
+ private final Path path;
+
+ private final boolean recursive;
+
+ private boolean nonRecursiveDeleteDirectoryFailed = false;
+
+ private final TracingContext tracingContext;
+
+ private final AtomicInteger deleteCount = new AtomicInteger(0);
+
+ /** Constructor
+ *
+ * @param path path to delete.
+ * @param recursive if true, delete the path recursively.
+ * @param abfsBlobClient client to use for blob operations.
+ * @param tracingContext tracing context.
+ */
+ public BlobDeleteHandler(final Path path,
+ final boolean recursive,
+ final AbfsBlobClient abfsBlobClient,
+ final TracingContext tracingContext) {
+ super(path, abfsBlobClient, tracingContext);
+ this.path = path;
+ this.recursive = recursive;
+ this.tracingContext = tracingContext;
+ }
+
+ /**{@inheritDoc}
+ *
+ * @return the maximum number of parallelism for delete operation.
+ */
+ @Override
+ int getMaxConsumptionParallelism() {
+ return getAbfsClient().getAbfsConfiguration()
+ .getBlobDeleteDirConsumptionParallelism();
+ }
+
+ /** Delete the path.
+ *
+ * @param path path to delete.
+ * @return true if the path is deleted.
+ * @throws AzureBlobFileSystemException server error.
+ */
+ private boolean deleteInternal(final Path path)
+ throws AzureBlobFileSystemException {
+ getAbfsClient().deleteBlobPath(path, null, tracingContext);
+ deleteCount.incrementAndGet();
+ return true;
+ }
+
+ /**
+ * Orchestrate the delete operation.
+ *
+ * @return true if the delete operation is successful.
+ * @throws AzureBlobFileSystemException if deletion fails due to server
error or path doesn't exist.
+ */
+ public boolean execute() throws AzureBlobFileSystemException {
+ /*
+ * ABFS is not aware if it's a file or directory. So, we need to list
the
+ * path and delete the listed objects. The listing returns the
children of
+ * the path and not the path itself.
+ */
+ listRecursiveAndTakeAction();
+ if (nonRecursiveDeleteDirectoryFailed) {
+ throw new AbfsRestOperationException(HTTP_CONFLICT,
+ DIRECTORY_NOT_EMPTY_DELETE.getErrorCode(),
+ DIRECTORY_NOT_EMPTY_DELETE.getErrorMessage(),
+ new PathIOException(path.toString(),
+ "Non-recursive delete of non-empty directory"));
+ }
+ tracingContext.setOperatedBlobCount(deleteCount.get() + 1);
+ /*
+ * If path is actually deleted.
+ */
+ boolean deleted;
+ try {
+ /*
+ * Delete the required path.
+ * Directory needs to be safely delete the path, as the path can
be implicit.
Review Comment:
Updated!
> ABFS: [FnsOverBlob] Implementing Rename and Delete APIs over Blob Endpoint
> --------------------------------------------------------------------------
>
> Key: HADOOP-19233
> URL: https://issues.apache.org/jira/browse/HADOOP-19233
> Project: Hadoop Common
> Issue Type: Sub-task
> Components: fs/azure
> Affects Versions: 3.4.0
> Reporter: Anuj Modi
> Assignee: Manish Bhatt
> Priority: Major
> Labels: pull-request-available
>
> Currently, we only support rename and delete operations on the DFS endpoint.
> The reason for 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.
> * {*}Blob Endpoint Usage{*}: The task addresses the need for abstraction in
> the code to maintain HDFS contracts while performing rename and delete
> operations on the blob endpoint, which does not support hierarchy.
> * {*}Rename Operations{*}: The {{AzureBlobFileSystem#rename()}} method will
> use a {{RenameHandler}} instance to handle rename operations, with separate
> handlers for the DFS and blob endpoints. This method includes prechecks,
> destination adjustments, and orchestration of directory renaming for blobs.
> * {*}Atomic Rename{*}: Atomic renaming is essential for blob endpoints, as
> it requires orchestration to copy or delete each blob within the directory. A
> configuration will allow developers to specify directories for atomic
> renaming, with a JSON file to track the status of renames.
> * {*}Delete Operations{*}: Delete operations are simpler than renames,
> requiring fewer HDFS contract checks. For blob endpoints, the client must
> handle orchestration, including managing orphaned directories created by
> Az-copy.
> * {*}Orchestration for Rename/Delete{*}: Orchestration for rename and delete
> operations over blob endpoints involves listing blobs and performing actions
> on each blob. The process must be optimized to handle large numbers of blobs
> efficiently.
> * {*}Need for Optimization{*}: Optimization is crucial because the
> {{ListBlob}} API can return a maximum of 5000 blobs at once, necessitating
> multiple calls for large directories. The task proposes a producer-consumer
> model to handle blobs in parallel, thereby reducing processing time and
> memory usage.
> * {*}Producer-Consumer Design{*}: The proposed design includes a producer to
> list blobs, a queue to store the blobs, and a consumer to process them in
> parallel. This approach aims to improve efficiency and mitigate memory issues.
> More details will follow
> Perquisites for this Patch:
> 1. HADOOP-19187 ABFS: [FnsOverBlob]Making AbfsClient Abstract for supporting
> both DFS and Blob Endpoint - ASF JIRA (apache.org)
> 2. HADOOP-19226 ABFS: [FnsOverBlob]Implementing Azure Rest APIs on Blob
> Endpoint for AbfsBlobClient - ASF JIRA (apache.org)
> 3. HADOOP-19207 ABFS: [FnsOverBlob]Response Handling of Blob Endpoint APIs
> and Metadata APIs - ASF JIRA (apache.org)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
