You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-issues@hadoop.apache.org by GitBox <gi...@apache.org> on 2019/06/03 12:50:46 UTC
[GitHub] [hadoop] bgaborg commented on a change in pull request #843:
HADOOP-15183 S3Guard store becomes inconsistent after partial failure of
rename
bgaborg commented on a change in pull request #843: HADOOP-15183 S3Guard store becomes inconsistent after partial failure of rename
URL: https://github.com/apache/hadoop/pull/843#discussion_r289827719
##########
File path: hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
##########
@@ -1225,130 +1301,292 @@ private boolean innerRename(Path source, Path dest)
}
}
- // If we have a MetadataStore, track deletions/creations.
- Collection<Path> srcPaths = null;
- List<PathMetadata> dstMetas = null;
- if (hasMetadataStore()) {
- srcPaths = new HashSet<>(); // srcPaths need fast look up before put
- dstMetas = new ArrayList<>();
- }
- // TODO S3Guard HADOOP-13761: retries when source paths are not visible yet
+ // Validation completed: time to begin the operation.
+ // The store-specific rename operation is used to keep the store
+ // to date with the in-progress operation.
+ // for the null store, these are all no-ops.
+ final RenameTracker renameTracker =
+ metadataStore.initiateRenameOperation(
+ createStoreContext(),
+ src, srcStatus, dest);
+ final AtomicLong bytesCopied = new AtomicLong();
+ int renameParallelLimit = RENAME_PARALLEL_LIMIT;
+ final List<CompletableFuture<Path>> activeCopies =
+ new ArrayList<>(renameParallelLimit);
+ // aggregate operation to wait for the copies to complete then reset
+ // the list.
+ final FunctionsRaisingIOE.FunctionRaisingIOE<String, Void>
+ completeActiveCopies = (String reason) -> {
+ LOG.debug("Waiting for {} active copies to complete: {}",
+ activeCopies.size(), reason);
+ waitForCompletion(activeCopies);
+ activeCopies.clear();
+ return null;
+ };
+
// TODO S3Guard: performance: mark destination dirs as authoritative
// Ok! Time to start
- if (srcStatus.isFile()) {
- LOG.debug("rename: renaming file {} to {}", src, dst);
- long length = srcStatus.getLen();
- S3ObjectAttributes objectAttributes =
- createObjectAttributes(srcStatus.getPath(),
- srcStatus.getETag(), srcStatus.getVersionId());
- S3AReadOpContext readContext = createReadContext(srcStatus, inputPolicy,
- changeDetectionPolicy, readAhead);
- if (dstStatus != null && dstStatus.isDirectory()) {
- String newDstKey = maybeAddTrailingSlash(dstKey);
- String filename =
- srcKey.substring(pathToKey(src.getParent()).length()+1);
- newDstKey = newDstKey + filename;
- CopyResult copyResult = copyFile(srcKey, newDstKey, length,
- objectAttributes, readContext);
- S3Guard.addMoveFile(metadataStore, srcPaths, dstMetas, src,
- keyToQualifiedPath(newDstKey), length, getDefaultBlockSize(dst),
- username, copyResult.getETag(), copyResult.getVersionId());
+ try {
+ if (srcStatus.isFile()) {
+ // the source is a file.
+ Path copyDestinationPath = dst;
+ String copyDestinationKey = dstKey;
+ S3ObjectAttributes sourceAttributes =
+ createObjectAttributes(srcStatus);
+ S3AReadOpContext readContext = createReadContext(srcStatus, inputPolicy,
+ changeDetectionPolicy, readAhead);
+ if (dstStatus != null && dstStatus.isDirectory()) {
+ // destination is a directory: build the final destination underneath
+ String newDstKey = maybeAddTrailingSlash(dstKey);
+ String filename =
+ srcKey.substring(pathToKey(src.getParent()).length() + 1);
+ newDstKey = newDstKey + filename;
+ copyDestinationKey = newDstKey;
+ copyDestinationPath = keyToQualifiedPath(newDstKey);
+ }
+ // destination either does not exist or is a file to overwrite.
+ LOG.debug("rename: renaming file {} to {}", src, copyDestinationPath);
+ copySourceAndUpdateTracker(renameTracker,
+ src,
+ srcKey,
+ sourceAttributes,
+ readContext,
+ copyDestinationPath,
+ copyDestinationKey,
+ false);
+ bytesCopied.addAndGet(srcStatus.getLen());
+ // delete the source
+ deleteObjectAtPath(src, srcKey, true);
+ // and update the tracker
+ renameTracker.sourceObjectsDeleted(Lists.newArrayList(src));
} else {
- CopyResult copyResult = copyFile(srcKey, dstKey, srcStatus.getLen(),
- objectAttributes, readContext);
- S3Guard.addMoveFile(metadataStore, srcPaths, dstMetas, src, dst,
- length, getDefaultBlockSize(dst), username,
- copyResult.getETag(), copyResult.getVersionId());
- }
- innerDelete(srcStatus, false);
- } else {
- LOG.debug("rename: renaming directory {} to {}", src, dst);
-
- // This is a directory to directory copy
- dstKey = maybeAddTrailingSlash(dstKey);
- srcKey = maybeAddTrailingSlash(srcKey);
+ LOG.debug("rename: renaming directory {} to {}", src, dst);
- //Verify dest is not a child of the source directory
- if (dstKey.startsWith(srcKey)) {
- throw new RenameFailedException(srcKey, dstKey,
- "cannot rename a directory to a subdirectory of itself ");
- }
+ // This is a directory-to-directory copy
+ dstKey = maybeAddTrailingSlash(dstKey);
+ srcKey = maybeAddTrailingSlash(srcKey);
- List<DeleteObjectsRequest.KeyVersion> keysToDelete = new ArrayList<>();
- if (dstStatus != null && dstStatus.isEmptyDirectory() == Tristate.TRUE) {
- // delete unnecessary fake directory.
- keysToDelete.add(new DeleteObjectsRequest.KeyVersion(dstKey));
- }
+ //Verify dest is not a child of the source directory
Review comment:
nit: space
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org