From d28aba348f6df8bbfd734fb2360c885d3c826e66 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Thu, 12 Oct 2023 20:19:10 +0530 Subject: [PATCH 01/13] HDFS-17381. Distcp of EC files should not be limited to DFS. (cherry picked from commit 3b663d79ba08fa96a58b29fd495dee61ee7830b4) --- .../hadoop/tools/mapred/CopyMapper.java | 10 ++-- .../RetriableDirectoryCreateCommand.java | 35 ++++++++++--- .../mapred/RetriableFileCopyCommand.java | 49 ++++++++++++------- .../hadoop/tools/util/RetriableCommand.java | 35 +++++++++++++ 4 files changed, 100 insertions(+), 29 deletions(-) diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java index ad17e574ca9b82..a07e3084d7156f 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java @@ -205,7 +205,8 @@ public void map(Text relPath, CopyListingFileStatus sourceFileStatus, } if (sourceCurrStatus.isDirectory()) { - createTargetDirsWithRetry(description, target, context, sourceStatus); + createTargetDirsWithRetry(description, target, context, sourceStatus, + sourceFS); return; } @@ -295,10 +296,11 @@ private void copyFileWithRetry(String description, } private void createTargetDirsWithRetry(String description, Path target, - Context context, FileStatus sourceStatus) throws IOException { + Context context, FileStatus sourceStatus, FileSystem sourceFS) + throws IOException { try { - new RetriableDirectoryCreateCommand(description).execute(target, - context, sourceStatus); + new RetriableDirectoryCreateCommand(description).execute(target, context, + sourceStatus, sourceFS); } catch (Exception e) { throw new IOException("mkdir failed for " + target, e); } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java index f5d9246e6a3bf2..748949253fc6d8 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java @@ -22,6 +22,7 @@ import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; +import org.apache.hadoop.hdfs.protocol.SystemErasureCodingPolicies; import org.apache.hadoop.tools.DistCpOptions; import org.apache.hadoop.tools.util.RetriableCommand; import org.apache.hadoop.fs.Path; @@ -53,10 +54,11 @@ public RetriableDirectoryCreateCommand(String description) { */ @Override protected Object doExecute(Object... arguments) throws Exception { - assert arguments.length == 3 : "Unexpected argument list."; + assert arguments.length == 4 : "Unexpected argument list."; Path target = (Path)arguments[0]; Mapper.Context context = (Mapper.Context)arguments[1]; FileStatus sourceStatus = (FileStatus)arguments[2]; + FileSystem sourceFs = (FileSystem)arguments[3]; FileSystem targetFS = target.getFileSystem(context.getConfiguration()); if(!targetFS.mkdirs(target)) { @@ -65,12 +67,31 @@ protected Object doExecute(Object... arguments) throws Exception { boolean preserveEC = getFileAttributeSettings(context) .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); - if (preserveEC && sourceStatus.isErasureCoded() - && targetFS instanceof DistributedFileSystem) { - ErasureCodingPolicy ecPolicy = - ((HdfsFileStatus) sourceStatus).getErasureCodingPolicy(); - DistributedFileSystem dfs = (DistributedFileSystem) targetFS; - dfs.setErasureCodingPolicy(target, ecPolicy.getName()); + if (preserveEC && sourceStatus.isErasureCoded()) { + ErasureCodingPolicy ecPolicy = null; + if (sourceFs instanceof DistributedFileSystem) { + ecPolicy = ((HdfsFileStatus) sourceStatus).getErasureCodingPolicy(); + } else { + try { + String ecPolicyName = (String) getErasureCodingPolicyMethod( + sourceFs).invoke(sourceFs,sourceStatus); + ecPolicy = SystemErasureCodingPolicies.getByName(ecPolicyName); + } catch (NoSuchMethodException exception){ + return false; + } + } + + if (targetFS instanceof DistributedFileSystem) { + DistributedFileSystem dfs = (DistributedFileSystem) targetFS; + dfs.setErasureCodingPolicy(target, ecPolicy.getName()); + } else { + try { + setErasureCodingPolicyMethod(targetFS).invoke(targetFS, + ecPolicy.getName()); + } catch (NoSuchMethodException exception) { + return false; + } + } } return true; } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java index 84bb0010086373..f1b2cece9b5ef8 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java @@ -21,12 +21,14 @@ import java.io.BufferedOutputStream; import java.io.IOException; import java.io.OutputStream; +import java.lang.reflect.InvocationTargetException; import java.util.EnumSet; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; +import org.apache.hadoop.hdfs.protocol.SystemErasureCodingPolicies; import org.apache.hadoop.tools.DistCpOptions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -129,7 +131,7 @@ protected Object doExecute(Object... arguments) throws Exception { private long doCopy(CopyListingFileStatus source, Path target, Mapper.Context context, EnumSet fileAttributes, FileStatus sourceStatus) - throws IOException { + throws Exception { LOG.info("Copying {} to {}", source.getPath(), target); final boolean toAppend = action == FileAction.APPEND; @@ -151,8 +153,8 @@ private long doCopy(CopyListingFileStatus source, Path target, long offset = (action == FileAction.APPEND) ? targetFS.getFileStatus(target).getLen() : source.getChunkOffset(); - long bytesRead = copyToFile(targetPath, targetFS, source, - offset, context, fileAttributes, sourceChecksum, sourceStatus); + long bytesRead = copyToFile(targetPath, targetFS, source, offset, context, + fileAttributes, sourceChecksum, sourceStatus, sourceFS); if (!source.isSplit()) { DistCpUtils.compareFileLengthsAndChecksums(source.getLen(), sourceFS, @@ -195,8 +197,9 @@ private ChecksumOpt getChecksumOpt(EnumSet fileAttributes, private long copyToFile(Path targetPath, FileSystem targetFS, CopyListingFileStatus source, long sourceOffset, Mapper.Context context, EnumSet fileAttributes, final FileChecksum sourceChecksum, - FileStatus sourceStatus) - throws IOException { + FileStatus sourceStatus,FileSystem sourceFS) + throws IOException, NoSuchMethodException, IllegalAccessException, + InvocationTargetException { FsPermission permission = FsPermission.getFileDefault().applyUMask( FsPermission.getUMask(targetFS.getConf())); int copyBufferSize = context.getConfiguration().getInt( @@ -206,10 +209,14 @@ private long copyToFile(Path targetPath, FileSystem targetFS, .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); ErasureCodingPolicy ecPolicy = null; - if (preserveEC && sourceStatus.isErasureCoded() - && sourceStatus instanceof HdfsFileStatus - && targetFS instanceof DistributedFileSystem) { - ecPolicy = ((HdfsFileStatus) sourceStatus).getErasureCodingPolicy(); + if (preserveEC && sourceStatus.isErasureCoded()) { + if (sourceStatus instanceof HdfsFileStatus){ + ecPolicy = ((HdfsFileStatus) sourceStatus).getErasureCodingPolicy(); + } else { + String ecPolicyName = (String) getErasureCodingPolicyMethod(sourceFS) + .invoke(sourceFS,sourceStatus); + ecPolicy = SystemErasureCodingPolicies.getByName(ecPolicyName); + } } final OutputStream outStream; if (action == FileAction.OVERWRITE) { @@ -227,16 +234,22 @@ private long copyToFile(Path targetPath, FileSystem targetFS, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), copyBufferSize, repl, blockSize, context, checksumOpt); } else { - DistributedFileSystem dfs = (DistributedFileSystem) targetFS; - DistributedFileSystem.HdfsDataOutputStreamBuilder builder = - dfs.createFile(targetPath).permission(permission).create() - .overwrite(true).bufferSize(copyBufferSize).replication(repl) - .blockSize(blockSize).progress(context).recursive() - .ecPolicyName(ecPolicy.getName()); - if (checksumOpt != null) { - builder.checksumOpt(checksumOpt); + if (targetFS instanceof DistributedFileSystem) { + DistributedFileSystem dfs = (DistributedFileSystem) targetFS; + DistributedFileSystem.HdfsDataOutputStreamBuilder builder = + dfs.createFile(targetPath).permission(permission).create() + .overwrite(true).bufferSize(copyBufferSize).replication(repl) + .blockSize(blockSize).progress(context).recursive() + .ecPolicyName(ecPolicy.getName()); + if (checksumOpt != null) { + builder.checksumOpt(checksumOpt); + } + out = builder.build(); + } else { + out = (FSDataOutputStream) createECOutputStreamMethod(targetFS).invoke( + targetFS, targetPath, permission, copyBufferSize, repl, blockSize, + checksumOpt, ecPolicy.getName()); } - out = builder.build(); } outStream = new BufferedOutputStream(out); } else { diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java index 17a80adaeb8f55..fa505f876e7e7e 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java @@ -19,6 +19,11 @@ package org.apache.hadoop.tools.util; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Options; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.permission.FsPermission; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.io.retry.RetryPolicy; @@ -27,6 +32,10 @@ import org.apache.hadoop.util.ThreadUtil; import java.io.IOException; +import java.lang.reflect.Method; +import java.net.URL; +import java.security.CodeSource; +import java.security.ProtectionDomain; import java.util.concurrent.TimeUnit; /** @@ -111,4 +120,30 @@ public RetriableCommand setRetryPolicy(RetryPolicy retryHandler) { this.retryPolicy = retryHandler; return this; } + + protected static Method getErasureCodingPolicyMethod(FileSystem fs) + throws NoSuchMethodException { + return fs.getClass().getMethod("getErasureCodingPolicy", + FileStatus.class); + } + + protected static Method setErasureCodingPolicyMethod(FileSystem fs) + throws NoSuchMethodException { + ProtectionDomain protectionDomain = fs.getClass().getProtectionDomain(); + CodeSource codeSource = protectionDomain.getCodeSource(); + URL location = codeSource.getLocation(); + System.out.println("Loaded from: " + location); + return fs.getClass() + .getMethod("setErasureCodingPolicy", Path.class, String.class, + int.class, int.class, int.class); + } + + protected static Method createECOutputStreamMethod(FileSystem fs) + throws NoSuchMethodException { + return fs.getClass() + .getMethod("createECOutputStream", Path.class, FsPermission.class, + int.class, short.class, long.class, Options.ChecksumOpt.class, + String.class); + } + } From bb74b5c01a2da38e4a76d29f22c215ad9e9a269d Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Sat, 24 Feb 2024 00:34:14 +0530 Subject: [PATCH 02/13] create an interface for EC --- .../apache/hadoop/fs/WithErasureCoding.java | 53 +++++++++++++++++++ .../hadoop/hdfs/DistributedFileSystem.java | 25 ++++++++- .../hadoop/tools/mapred/CopyMapper.java | 3 +- .../RetriableDirectoryCreateCommand.java | 34 +++--------- .../mapred/RetriableFileCopyCommand.java | 41 +++++--------- 5 files changed, 100 insertions(+), 56 deletions(-) create mode 100644 hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java new file mode 100644 index 00000000000000..4fc1470c612f61 --- /dev/null +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java @@ -0,0 +1,53 @@ +/* + * 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; + +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.util.Progressable; + +import java.io.IOException; + +/** + * Filesystems that support EC can implement this interface. + */ +public interface WithErasureCoding { + + /** + * Get the EC Policy name of the given file + * @param fileStatus object of the file whose ecPolicy needs to be obtained + * @return the ec Policy name + */ + String getErasureCodingPolicyName(FileStatus fileStatus); + + /** + * Set the given ecPolicy on the path + * @param path on which the EC policy needs to be set + * @throws IOException if the set is not successful + */ + void setErasureCodingPolicy(Path path, String ecPolicyName) throws + IOException; + + /** + * A create file API but for EC files. + */ + FSDataOutputStreamBuilder createECFile(Path path, + FsPermission permission, boolean overwrite, int bufferSize, + short replication, long blockSize, Progressable prog, + Options.ChecksumOpt checksumOpt, String ecPolicy); +} diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java index 17c39f6c55b75c..5cd56272087db6 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java @@ -74,6 +74,7 @@ import org.apache.hadoop.fs.permission.AclStatus; import org.apache.hadoop.fs.permission.FsAction; import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.hdfs.DFSOpsCountStatistics.OpType; import org.apache.hadoop.hdfs.client.DfsPathCapabilities; import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys; @@ -146,7 +147,8 @@ @InterfaceAudience.LimitedPrivate({ "MapReduce", "HBase" }) @InterfaceStability.Unstable public class DistributedFileSystem extends FileSystem - implements KeyProviderTokenIssuer, BatchListingOperations, LeaseRecoverable, SafeMode { + implements KeyProviderTokenIssuer, BatchListingOperations, LeaseRecoverable, SafeMode, + WithErasureCoding { private Path workingDir; private URI uri; @@ -376,6 +378,27 @@ public FSDataInputStream open(PathHandle fd, int bufferSize) return dfs.createWrappedInputStream(dfsis); } + @Override + public String getErasureCodingPolicyName(FileStatus fileStatus) { + return ((HdfsFileStatus) fileStatus).getErasureCodingPolicy().getName(); + } + + @Override + public FSDataOutputStreamBuilder createECFile( + Path path, FsPermission permission, boolean overwrite, int bufferSize, + short replication, long blockSize, Progressable prog, + ChecksumOpt checksumOpt, String ecPolicy) { + HdfsDataOutputStreamBuilder builder = + createFile(path).permission(permission).create().overwrite(true) + .bufferSize(bufferSize).replication(replication) + .blockSize(blockSize).progress(prog).recursive() + .ecPolicyName(ecPolicy); + if (checksumOpt != null) { + builder.checksumOpt(checksumOpt); + } + return builder; + } + /** * Create a handle to an HDFS file. * @param st HdfsFileStatus instance from NameNode diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java index a07e3084d7156f..904b297c2b09a9 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java @@ -296,8 +296,7 @@ private void copyFileWithRetry(String description, } private void createTargetDirsWithRetry(String description, Path target, - Context context, FileStatus sourceStatus, FileSystem sourceFS) - throws IOException { + Context context, FileStatus sourceStatus, FileSystem sourceFS) throws IOException { try { new RetriableDirectoryCreateCommand(description).execute(target, context, sourceStatus, sourceFS); diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java index 748949253fc6d8..1a3ae1050b6db3 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java @@ -19,9 +19,9 @@ package org.apache.hadoop.tools.mapred; import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; -import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; import org.apache.hadoop.hdfs.protocol.SystemErasureCodingPolicies; import org.apache.hadoop.tools.DistCpOptions; import org.apache.hadoop.tools.util.RetriableCommand; @@ -67,31 +67,13 @@ protected Object doExecute(Object... arguments) throws Exception { boolean preserveEC = getFileAttributeSettings(context) .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); - if (preserveEC && sourceStatus.isErasureCoded()) { - ErasureCodingPolicy ecPolicy = null; - if (sourceFs instanceof DistributedFileSystem) { - ecPolicy = ((HdfsFileStatus) sourceStatus).getErasureCodingPolicy(); - } else { - try { - String ecPolicyName = (String) getErasureCodingPolicyMethod( - sourceFs).invoke(sourceFs,sourceStatus); - ecPolicy = SystemErasureCodingPolicies.getByName(ecPolicyName); - } catch (NoSuchMethodException exception){ - return false; - } - } - - if (targetFS instanceof DistributedFileSystem) { - DistributedFileSystem dfs = (DistributedFileSystem) targetFS; - dfs.setErasureCodingPolicy(target, ecPolicy.getName()); - } else { - try { - setErasureCodingPolicyMethod(targetFS).invoke(targetFS, - ecPolicy.getName()); - } catch (NoSuchMethodException exception) { - return false; - } - } + if (preserveEC && sourceStatus.isErasureCoded() + && targetFS instanceof WithErasureCoding) { + ErasureCodingPolicy ecPolicy = SystemErasureCodingPolicies.getByName( + ((WithErasureCoding) sourceFs).getErasureCodingPolicyName( + sourceStatus)); + WithErasureCoding ecFs = (DistributedFileSystem) targetFS; + ecFs.setErasureCodingPolicy(target,ecPolicy.getName()); } return true; } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java index f1b2cece9b5ef8..31e26c24cbabc9 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java @@ -21,23 +21,22 @@ import java.io.BufferedOutputStream; import java.io.IOException; import java.io.OutputStream; -import java.lang.reflect.InvocationTargetException; import java.util.EnumSet; import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; -import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; import org.apache.hadoop.hdfs.protocol.SystemErasureCodingPolicies; import org.apache.hadoop.tools.DistCpOptions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.CreateFlag; +import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileChecksum; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FSDataOutputStreamBuilder; import org.apache.hadoop.fs.Options.ChecksumOpt; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.permission.FsPermission; @@ -131,7 +130,7 @@ protected Object doExecute(Object... arguments) throws Exception { private long doCopy(CopyListingFileStatus source, Path target, Mapper.Context context, EnumSet fileAttributes, FileStatus sourceStatus) - throws Exception { + throws IOException { LOG.info("Copying {} to {}", source.getPath(), target); final boolean toAppend = action == FileAction.APPEND; @@ -198,8 +197,7 @@ private long copyToFile(Path targetPath, FileSystem targetFS, CopyListingFileStatus source, long sourceOffset, Mapper.Context context, EnumSet fileAttributes, final FileChecksum sourceChecksum, FileStatus sourceStatus,FileSystem sourceFS) - throws IOException, NoSuchMethodException, IllegalAccessException, - InvocationTargetException { + throws IOException { FsPermission permission = FsPermission.getFileDefault().applyUMask( FsPermission.getUMask(targetFS.getConf())); int copyBufferSize = context.getConfiguration().getInt( @@ -209,12 +207,12 @@ private long copyToFile(Path targetPath, FileSystem targetFS, .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); ErasureCodingPolicy ecPolicy = null; - if (preserveEC && sourceStatus.isErasureCoded()) { - if (sourceStatus instanceof HdfsFileStatus){ - ecPolicy = ((HdfsFileStatus) sourceStatus).getErasureCodingPolicy(); - } else { - String ecPolicyName = (String) getErasureCodingPolicyMethod(sourceFS) - .invoke(sourceFS,sourceStatus); + if (preserveEC && sourceStatus.isErasureCoded() + && sourceFS instanceof WithErasureCoding + && targetFS instanceof WithErasureCoding) { + String ecPolicyName = + ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); + if (ecPolicyName != null) { ecPolicy = SystemErasureCodingPolicies.getByName(ecPolicyName); } } @@ -234,22 +232,11 @@ private long copyToFile(Path targetPath, FileSystem targetFS, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), copyBufferSize, repl, blockSize, context, checksumOpt); } else { - if (targetFS instanceof DistributedFileSystem) { - DistributedFileSystem dfs = (DistributedFileSystem) targetFS; - DistributedFileSystem.HdfsDataOutputStreamBuilder builder = - dfs.createFile(targetPath).permission(permission).create() - .overwrite(true).bufferSize(copyBufferSize).replication(repl) - .blockSize(blockSize).progress(context).recursive() - .ecPolicyName(ecPolicy.getName()); - if (checksumOpt != null) { - builder.checksumOpt(checksumOpt); - } + FSDataOutputStreamBuilder builder = + ((WithErasureCoding) targetFS).createECFile(targetPath, permission, + true, copyBufferSize, repl, blockSize, context, checksumOpt, + ecPolicy.getName()); out = builder.build(); - } else { - out = (FSDataOutputStream) createECOutputStreamMethod(targetFS).invoke( - targetFS, targetPath, permission, copyBufferSize, repl, blockSize, - checksumOpt, ecPolicy.getName()); - } } outStream = new BufferedOutputStream(out); } else { From fb1c8fcdd7354cce2869c36ffde630e79b173286 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Sat, 24 Feb 2024 00:38:36 +0530 Subject: [PATCH 03/13] remove unwanted methods --- .../hadoop/tools/util/RetriableCommand.java | 26 ------------------- 1 file changed, 26 deletions(-) diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java index fa505f876e7e7e..87809c947f7538 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java @@ -120,30 +120,4 @@ public RetriableCommand setRetryPolicy(RetryPolicy retryHandler) { this.retryPolicy = retryHandler; return this; } - - protected static Method getErasureCodingPolicyMethod(FileSystem fs) - throws NoSuchMethodException { - return fs.getClass().getMethod("getErasureCodingPolicy", - FileStatus.class); - } - - protected static Method setErasureCodingPolicyMethod(FileSystem fs) - throws NoSuchMethodException { - ProtectionDomain protectionDomain = fs.getClass().getProtectionDomain(); - CodeSource codeSource = protectionDomain.getCodeSource(); - URL location = codeSource.getLocation(); - System.out.println("Loaded from: " + location); - return fs.getClass() - .getMethod("setErasureCodingPolicy", Path.class, String.class, - int.class, int.class, int.class); - } - - protected static Method createECOutputStreamMethod(FileSystem fs) - throws NoSuchMethodException { - return fs.getClass() - .getMethod("createECOutputStream", Path.class, FsPermission.class, - int.class, short.class, long.class, Options.ChecksumOpt.class, - String.class); - } - } From 76722aa4a23a5a403168c0600684aeaa4b15de10 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Sat, 24 Feb 2024 00:41:09 +0530 Subject: [PATCH 04/13] remove unused imports --- .../org/apache/hadoop/tools/util/RetriableCommand.java | 9 --------- 1 file changed, 9 deletions(-) diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java index 87809c947f7538..17a80adaeb8f55 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/RetriableCommand.java @@ -19,11 +19,6 @@ package org.apache.hadoop.tools.util; -import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Options; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.fs.permission.FsPermission; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.io.retry.RetryPolicy; @@ -32,10 +27,6 @@ import org.apache.hadoop.util.ThreadUtil; import java.io.IOException; -import java.lang.reflect.Method; -import java.net.URL; -import java.security.CodeSource; -import java.security.ProtectionDomain; import java.util.concurrent.TimeUnit; /** From 608c0b096299110d40f00d3dc826e7a3fd696ad9 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Fri, 30 Aug 2024 21:05:19 +0530 Subject: [PATCH 05/13] use createFile builder API --- .../main/java/org/apache/hadoop/fs/Options.java | 5 +++++ .../org/apache/hadoop/fs/WithErasureCoding.java | 11 ----------- .../hadoop/hdfs/DistributedFileSystem.java | 5 +++++ .../hadoop/hdfs/client/DfsPathCapabilities.java | 2 ++ .../mapred/RetriableDirectoryCreateCommand.java | 4 ++-- .../tools/mapred/RetriableFileCopyCommand.java | 16 ++++++++++++---- 6 files changed, 26 insertions(+), 17 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java index 9ef7de657dc152..eeaf53e7289486 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java @@ -641,5 +641,10 @@ private OpenFileOptions() { FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE) .collect(Collectors.toSet())); + /** + * EC policy to be set on the file that needs to be created. + */ + public static final String FS_OPTION_OPENFILE_EC_POLICY = + FS_OPTION_OPENFILE + "ec.policy"; } } diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java index 4fc1470c612f61..b6dc045f86abb8 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java @@ -18,9 +18,6 @@ package org.apache.hadoop.fs; -import org.apache.hadoop.fs.permission.FsPermission; -import org.apache.hadoop.util.Progressable; - import java.io.IOException; /** @@ -42,12 +39,4 @@ public interface WithErasureCoding { */ void setErasureCodingPolicy(Path path, String ecPolicyName) throws IOException; - - /** - * A create file API but for EC files. - */ - FSDataOutputStreamBuilder createECFile(Path path, - FsPermission permission, boolean overwrite, int bufferSize, - short replication, long blockSize, Progressable prog, - Options.ChecksumOpt checksumOpt, String ecPolicy); } diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java index 5cd56272087db6..eb0071e243b48b 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java @@ -3885,6 +3885,11 @@ protected EnumSet getFlags() { */ @Override public FSDataOutputStream build() throws IOException { + if (getOptionalKeys().contains( + Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { + ecPolicyName(getOptions().get( + Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)); + } if (getFlags().contains(CreateFlag.CREATE) || getFlags().contains(CreateFlag.OVERWRITE)) { if (isRecursive()) { diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/DfsPathCapabilities.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/DfsPathCapabilities.java index 612a9776303275..b779e42014f1c9 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/DfsPathCapabilities.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/DfsPathCapabilities.java @@ -21,6 +21,7 @@ import java.util.Optional; import org.apache.hadoop.fs.CommonPathCapabilities; +import org.apache.hadoop.fs.Options; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -54,6 +55,7 @@ public static Optional hasPathCapability(final Path path, case CommonPathCapabilities.FS_STORAGEPOLICY: case CommonPathCapabilities.FS_XATTRS: case CommonPathCapabilities.FS_TRUNCATE: + case Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY: return Optional.of(true); case CommonPathCapabilities.FS_SYMLINKS: return Optional.of(FileSystem.areSymlinksEnabled()); diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java index 1a3ae1050b6db3..653d91135a710e 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java @@ -72,8 +72,8 @@ protected Object doExecute(Object... arguments) throws Exception { ErasureCodingPolicy ecPolicy = SystemErasureCodingPolicies.getByName( ((WithErasureCoding) sourceFs).getErasureCodingPolicyName( sourceStatus)); - WithErasureCoding ecFs = (DistributedFileSystem) targetFS; - ecFs.setErasureCodingPolicy(target,ecPolicy.getName()); + WithErasureCoding ecFs = (WithErasureCoding) targetFS; + ecFs.setErasureCodingPolicy(target, ecPolicy.getName()); } return true; } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java index 31e26c24cbabc9..d445d9b8d5c907 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java @@ -55,6 +55,7 @@ import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY; import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL; +import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY; import static org.apache.hadoop.tools.mapred.CopyMapper.getFileAttributeSettings; import static org.apache.hadoop.util.functional.FutureIO.awaitFuture; @@ -232,10 +233,17 @@ private long copyToFile(Path targetPath, FileSystem targetFS, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), copyBufferSize, repl, blockSize, context, checksumOpt); } else { - FSDataOutputStreamBuilder builder = - ((WithErasureCoding) targetFS).createECFile(targetPath, permission, - true, copyBufferSize, repl, blockSize, context, checksumOpt, - ecPolicy.getName()); + String ecPolicyName = + ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); + FSDataOutputStreamBuilder builder = targetFS.createFile(targetPath) + .permission(permission) + .overwrite(true) + .bufferSize(copyBufferSize) + .replication(repl) + .blockSize(blockSize) + .progress(context) + .recursive(); + builder.opt(FS_OPTION_OPENFILE_EC_POLICY, ecPolicyName); out = builder.build(); } outStream = new BufferedOutputStream(out); From d030903d9afc4132f03b237cfc9263f50d6f87d4 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Fri, 30 Aug 2024 21:07:36 +0530 Subject: [PATCH 06/13] remove unused method --- .../hadoop/hdfs/DistributedFileSystem.java | 16 ---------------- 1 file changed, 16 deletions(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java index eb0071e243b48b..c443ae25628851 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java @@ -383,22 +383,6 @@ public String getErasureCodingPolicyName(FileStatus fileStatus) { return ((HdfsFileStatus) fileStatus).getErasureCodingPolicy().getName(); } - @Override - public FSDataOutputStreamBuilder createECFile( - Path path, FsPermission permission, boolean overwrite, int bufferSize, - short replication, long blockSize, Progressable prog, - ChecksumOpt checksumOpt, String ecPolicy) { - HdfsDataOutputStreamBuilder builder = - createFile(path).permission(permission).create().overwrite(true) - .bufferSize(bufferSize).replication(replication) - .blockSize(blockSize).progress(prog).recursive() - .ecPolicyName(ecPolicy); - if (checksumOpt != null) { - builder.checksumOpt(checksumOpt); - } - return builder; - } - /** * Create a handle to an HDFS file. * @param st HdfsFileStatus instance from NameNode From 292e110f05630c6df1e68912911a44e85f8300ef Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Sun, 1 Sep 2024 16:43:49 +0530 Subject: [PATCH 07/13] address comments and add tests --- .../java/org/apache/hadoop/fs/FileStatus.java | 8 +- .../apache/hadoop/fs/WithErasureCoding.java | 17 ++- .../hadoop/hdfs/DistributedFileSystem.java | 7 +- .../RetriableDirectoryCreateCommand.java | 36 +++++- .../mapred/RetriableFileCopyCommand.java | 34 ++--- .../hadoop/tools/TestDistCpWithRawXAttrs.java | 116 ++++++++++++++++++ 6 files changed, 189 insertions(+), 29 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java index 1d9458148e4a55..813dbc4b8ffda2 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java @@ -27,6 +27,7 @@ import java.util.EnumSet; import java.util.Set; +import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.fs.FSProtos.FileStatusProto; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; @@ -351,7 +352,12 @@ public String getGroup() { public Path getPath() { return path; } - + + @VisibleForTesting + public void setAttr(Set attr) { + this.attr = attr; + } + public void setPath(final Path p) { path = p; } diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java index b6dc045f86abb8..4ba7d2145fd4a5 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java @@ -26,16 +26,23 @@ public interface WithErasureCoding { /** - * Get the EC Policy name of the given file - * @param fileStatus object of the file whose ecPolicy needs to be obtained + * Get the EC Policy name of the given file's fileStatus. + * If the file is not erasure coded, this should return null. + * Callers will make sure to check if the FS schema of the fileStatus + * is that of an FS that implements this interface. + * If the call fails due to some error, return null. + * @param fileStatus object of the file whose ecPolicy needs to be obtained. * @return the ec Policy name */ String getErasureCodingPolicyName(FileStatus fileStatus); /** - * Set the given ecPolicy on the path - * @param path on which the EC policy needs to be set - * @throws IOException if the set is not successful + * Set the given ecPolicy on the path. + * The path & ecPolicyName should be valid (not null/empty, the + * implementing FS should support the supplied ecPolicy). + * implementations can throw IOException if these conditions are not met. + * @param path on which the EC policy needs to be set. + * @throws IOException if there is an error during the set op. */ void setErasureCodingPolicy(Path path, String ecPolicyName) throws IOException; diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java index c443ae25628851..51ffab77448b73 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java @@ -3869,10 +3869,9 @@ protected EnumSet getFlags() { */ @Override public FSDataOutputStream build() throws IOException { - if (getOptionalKeys().contains( - Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { - ecPolicyName(getOptions().get( - Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)); + String ecPolicy = getOptions().get(Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY, ""); + if (!ecPolicy.isEmpty()) { + ecPolicyName(ecPolicy); } if (getFlags().contains(CreateFlag.CREATE) || getFlags().contains(CreateFlag.OVERWRITE)) { diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java index 653d91135a710e..9d2748fc10241a 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java @@ -18,16 +18,20 @@ package org.apache.hadoop.tools.mapred; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.Options; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.WithErasureCoding; -import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; import org.apache.hadoop.hdfs.protocol.SystemErasureCodingPolicies; import org.apache.hadoop.tools.DistCpOptions; import org.apache.hadoop.tools.util.RetriableCommand; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.mapreduce.Mapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; import static org.apache.hadoop.tools.mapred.CopyMapper.getFileAttributeSettings; @@ -37,6 +41,9 @@ */ public class RetriableDirectoryCreateCommand extends RetriableCommand { + private static final Logger LOG = + LoggerFactory.getLogger(RetriableDirectoryCreateCommand.class); + /** * Constructor, taking a description of the action. * @param description Verbose description of the copy operation. @@ -68,13 +75,32 @@ protected Object doExecute(Object... arguments) throws Exception { boolean preserveEC = getFileAttributeSettings(context) .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); if (preserveEC && sourceStatus.isErasureCoded() - && targetFS instanceof WithErasureCoding) { + && checkFSSupportsEC(sourceStatus.getPath(), sourceFs) + && checkFSSupportsEC(target, targetFS)) { ErasureCodingPolicy ecPolicy = SystemErasureCodingPolicies.getByName( ((WithErasureCoding) sourceFs).getErasureCodingPolicyName( sourceStatus)); + if (LOG.isDebugEnabled()) { + LOG.debug("EC Policy for source path is {}", ecPolicy); + } WithErasureCoding ecFs = (WithErasureCoding) targetFS; - ecFs.setErasureCodingPolicy(target, ecPolicy.getName()); + if (ecPolicy != null) { + ecFs.setErasureCodingPolicy(target, ecPolicy.getName()); + } } return true; } + + /** + * Return true if the FS implements {@link WithErasureCoding} and + * supports EC_POLICY option in {@link Options.OpenFileOptions} + */ + boolean checkFSSupportsEC(Path path, FileSystem fs) throws IOException { + if (fs instanceof WithErasureCoding && fs.hasPathCapability(path, + Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { + return true; + } + LOG.warn("FS with scheme " + fs.getScheme() + " does not support EC"); + return false; + } } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java index d445d9b8d5c907..25bc7dd4964667 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java @@ -24,22 +24,21 @@ import java.util.EnumSet; import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; -import org.apache.hadoop.hdfs.protocol.SystemErasureCodingPolicies; import org.apache.hadoop.tools.DistCpOptions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.CreateFlag; -import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileChecksum; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FSDataOutputStreamBuilder; +import org.apache.hadoop.fs.Options; import org.apache.hadoop.fs.Options.ChecksumOpt; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.tools.CopyListingFileStatus; @@ -207,15 +206,11 @@ private long copyToFile(Path targetPath, FileSystem targetFS, boolean preserveEC = getFileAttributeSettings(context) .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); - ErasureCodingPolicy ecPolicy = null; + String ecPolicyName = null; if (preserveEC && sourceStatus.isErasureCoded() - && sourceFS instanceof WithErasureCoding - && targetFS instanceof WithErasureCoding) { - String ecPolicyName = - ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); - if (ecPolicyName != null) { - ecPolicy = SystemErasureCodingPolicies.getByName(ecPolicyName); - } + && checkFSSupportsEC(sourceStatus.getPath(), sourceFS) + && checkFSSupportsEC(targetPath, targetFS)) { + ecPolicyName = ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); } final OutputStream outStream; if (action == FileAction.OVERWRITE) { @@ -228,13 +223,11 @@ private long copyToFile(Path targetPath, FileSystem targetFS, targetFS, targetPath); FSDataOutputStream out; ChecksumOpt checksumOpt = getChecksumOpt(fileAttributes, sourceChecksum); - if (!preserveEC || ecPolicy == null) { + if (!preserveEC || ecPolicyName == null) { out = targetFS.create(targetPath, permission, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), copyBufferSize, repl, blockSize, context, checksumOpt); } else { - String ecPolicyName = - ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); FSDataOutputStreamBuilder builder = targetFS.createFile(targetPath) .permission(permission) .overwrite(true) @@ -410,6 +403,19 @@ private static long getBlockSize( .getDefaultBlockSize(tmpTargetPath); } + /** + * Return true if the FS implements {@link WithErasureCoding} and + * supports EC_POLICY option in {@link Options.OpenFileOptions} + */ + boolean checkFSSupportsEC(Path path, FileSystem fs) throws IOException { + if (fs instanceof WithErasureCoding && fs.hasPathCapability(path, + Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { + return true; + } + LOG.warn("FS with scheme " + fs.getScheme() + " does not support EC"); + return false; + } + /** * Special subclass of IOException. This is used to distinguish read-operation * failures from other kinds of IOExceptions. diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java index 841501869b5e4c..9711180fc011b6 100644 --- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java +++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java @@ -18,12 +18,20 @@ package org.apache.hadoop.tools; +import java.io.File; +import java.io.IOException; +import java.net.URI; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.LocalFileSystem; +import org.apache.hadoop.fs.Options; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.hdfs.DFSConfigKeys; import org.apache.hadoop.hdfs.DistributedFileSystem; import org.apache.hadoop.hdfs.MiniDFSCluster; @@ -41,8 +49,10 @@ import org.apache.hadoop.thirdparty.com.google.common.collect.Maps; +import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; /** @@ -68,12 +78,17 @@ public class TestDistCpWithRawXAttrs { private static final String rootedSrcName = "/src"; private static final String rawDestName = "/.reserved/raw/dest"; private static final String rawSrcName = "/.reserved/raw/src"; + private static final File base = + GenericTestUtils.getTestDir("work-dir/localfs"); + + private static final String TEST_ROOT_DIR = base.getAbsolutePath(); @BeforeClass public static void init() throws Exception { conf = new Configuration(); conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_XATTRS_ENABLED_KEY, true); conf.setInt(DFSConfigKeys.DFS_LIST_LIMIT, 2); + conf.setClass("fs.file.impl", DummyEcFs.class, FileSystem.class); cluster = new MiniDFSCluster.Builder(conf).numDataNodes(3).format(true) .build(); cluster.waitActive(); @@ -240,6 +255,107 @@ public void testPreserveAndNoPreserveEC() throws Exception { dfs.unsetErasureCodingPolicy(dir1); } + + @Test + public void testPreserveECAcrossFilesystems() throws Exception{ + // set EC policy on source (HDFS) + String[] args = {"-setPolicy", "-path", dir1.toString(), + "-policy", "XOR-2-1-1024k"}; + fs.delete(new Path("/dest"), true); + fs.mkdirs(subDir1); + DistributedFileSystem dfs = (DistributedFileSystem) fs; + dfs.enableErasureCodingPolicy("XOR-2-1-1024k"); + dfs.setErasureCodingPolicy(dir1, "XOR-2-1-1024k"); + fs.create(file1).close(); + int res = ToolRunner.run(conf, new ECAdmin(conf), args); + assertEquals("Unable to set EC policy on " + subDir1.toString(), res, 0); + String src = "/src/*"; + Path dest = new Path(TEST_ROOT_DIR, "dest"); + final Path dest2Dir1 = new Path(dest, "dir1"); + final Path dest2SubDir1 = new Path(dest2Dir1, "subdir1"); + + // copy source(HDFS) to target(DummyECFS) with preserveEC + + try (DummyEcFs dummyEcFs = (DummyEcFs)FileSystem.get(URI.create("file:///"), conf)) { + Path target = dummyEcFs.makeQualified(dest); + DistCpTestUtils.assertRunDistCp(DistCpConstants.SUCCESS, src, target.toString(), + "-pe", conf); + try { + FileStatus destDir1Status = dummyEcFs.getFileStatus(dest2Dir1); + FileStatus destSubDir1Status = dummyEcFs.getFileStatus(dest2SubDir1); + assertNotNull(destDir1Status); + assertNotNull(destSubDir1Status); + // check if target paths are erasure coded. + assertTrue(dummyEcFs.isPathErasureCoded(destDir1Status.getPath())); + assertTrue(dummyEcFs.isPathErasureCoded(destSubDir1Status.getPath())); + + // copy source(DummyECFS) to target (HDFS) + String dfsTarget = "/dest"; + DistCpTestUtils.assertRunDistCp(DistCpConstants.SUCCESS, + target.toString(), dfsTarget, "-pe", conf); + Path dfsTargetPath = new Path(dfsTarget); + Path dfsTargetDir1 = new Path(dfsTarget, "dir1"); + assertTrue(fs.exists(dfsTargetPath)); + assertTrue(fs.exists(dfsTargetDir1)); + assertTrue(fs.getFileStatus(dfsTargetDir1).isErasureCoded()); + fs.delete(dfsTargetPath, true); + } finally { + dummyEcFs.delete(new Path(base.getAbsolutePath()),true); + } + } + + } + + /** + * Dummy/Fake FS implementation that supports Erasure Coding. + */ + public static class DummyEcFs extends LocalFileSystem implements WithErasureCoding { + + private Set erasureCodedPaths; + public DummyEcFs() { + super(); + this.erasureCodedPaths = new HashSet<>(); + } + + public boolean isPathErasureCoded(Path p){ + return erasureCodedPaths.contains(p); + } + + + @Override + public boolean hasPathCapability(Path path, String capability) + throws IOException { + switch (validatePathCapabilityArgs(makeQualified(path), capability)) { + case Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY: + return true; + default: + return super.hasPathCapability(path, capability); + } + } + + @Override + public FileStatus getFileStatus(Path f) throws IOException { + FileStatus fileStatus = super.getFileStatus(f); + if (erasureCodedPaths.contains(f)) { + Set attrSet = new HashSet<>(); + attrSet.add(FileStatus.AttrFlags.HAS_EC); + fileStatus.setAttr(attrSet); + } + return fileStatus; + } + + @Override + public String getErasureCodingPolicyName(FileStatus fileStatus) { + return "XOR-2-1-1024k"; + } + + @Override + public void setErasureCodingPolicy(Path path, String ecPolicyName) + throws IOException { + erasureCodedPaths.add(path); + } + } + @Test public void testUseIterator() throws Exception { From f22360af04c4df7182297f472be768e5c6ae3be5 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Sun, 1 Sep 2024 21:00:25 +0530 Subject: [PATCH 08/13] fix test --- .../java/org/apache/hadoop/fs/WithErasureCoding.java | 3 ++- .../org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java | 9 +++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java index 4ba7d2145fd4a5..e84dc4a8b733f7 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java @@ -38,10 +38,11 @@ public interface WithErasureCoding { /** * Set the given ecPolicy on the path. - * The path & ecPolicyName should be valid (not null/empty, the + * The path and ecPolicyName should be valid (not null/empty, the * implementing FS should support the supplied ecPolicy). * implementations can throw IOException if these conditions are not met. * @param path on which the EC policy needs to be set. + * @param ecPolicyName the EC policy. * @throws IOException if there is an error during the set op. */ void setErasureCodingPolicy(Path path, String ecPolicyName) throws diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java index efc3b90b5a9706..10133934efc5fa 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java @@ -27,6 +27,7 @@ import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.fs.FileEncryptionInfo; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileStatus.AttrFlags; @@ -494,6 +495,14 @@ default FileStatus makeQualified(URI defaultUri, Path parent) { String getNamespace(); + /** + * See {@link FileStatus#setAttr(Set)}. + */ + @VisibleForTesting + default void setAttr(Set attr){ + // do nothing (this is used only for testing) + } + /** * Set redundant flags for compatibility with existing applications. */ From 79e99d3d964491ed0aa59a45ef3a9a7f2c775d4e Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Tue, 10 Sep 2024 17:20:50 +0530 Subject: [PATCH 09/13] address comments --- .../java/org/apache/hadoop/fs/FileStatus.java | 5 --- .../java/org/apache/hadoop/fs/FileUtil.java | 13 ++++++ .../java/org/apache/hadoop/fs/Options.java | 2 +- .../apache/hadoop/fs/WithErasureCoding.java | 8 ++-- .../hadoop/hdfs/DistributedFileSystem.java | 2 + .../hadoop/hdfs/protocol/HdfsFileStatus.java | 8 ---- .../RetriableDirectoryCreateCommand.java | 26 +++--------- .../mapred/RetriableFileCopyCommand.java | 19 ++------- .../hadoop/tools/TestDistCpWithRawXAttrs.java | 42 ++++++++++++------- 9 files changed, 56 insertions(+), 69 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java index 813dbc4b8ffda2..a7e1390cff880f 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java @@ -353,11 +353,6 @@ public Path getPath() { return path; } - @VisibleForTesting - public void setAttr(Set attr) { - this.attr = attr; - } - public void setPath(final Path p) { path = p; } diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java index fa87bb48aaa698..c2634af3bb8505 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java @@ -2108,4 +2108,17 @@ public static void maybeIgnoreMissingDirectory(FileSystem fs, LOG.info("Ignoring missing directory {}", path); LOG.debug("Directory missing", e); } + + /** + * Return true if the FS implements {@link WithErasureCoding} and + * supports EC_POLICY option in {@link Options.OpenFileOptions} + */ + public static boolean checkFSSupportsEC(FileSystem fs, Path path) throws IOException { + if (fs instanceof WithErasureCoding && + fs.hasPathCapability(path, Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { + return true; + } + LOG.warn("FS with scheme {} does not support EC", fs.getScheme()); + return false; + } } diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java index eeaf53e7289486..a8fb02248c540c 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java @@ -642,7 +642,7 @@ private OpenFileOptions() { .collect(Collectors.toSet())); /** - * EC policy to be set on the file that needs to be created. + * EC policy to be set on the file that needs to be created : {@value}. */ public static final String FS_OPTION_OPENFILE_EC_POLICY = FS_OPTION_OPENFILE + "ec.policy"; diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java index e84dc4a8b733f7..cf72af27564983 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java @@ -27,10 +27,10 @@ public interface WithErasureCoding { /** * Get the EC Policy name of the given file's fileStatus. - * If the file is not erasure coded, this should return null. - * Callers will make sure to check if the FS schema of the fileStatus - * is that of an FS that implements this interface. - * If the call fails due to some error, return null. + * If the file is not erasure coded, this shall return null. + * Callers will make sure to check if fileStatus isInstance of + * an FS that implements this interface. + * If the call fails due to some error, this shall return null. * @param fileStatus object of the file whose ecPolicy needs to be obtained. * @return the ec Policy name */ diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java index 51ffab77448b73..219888dcbec271 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java @@ -380,6 +380,8 @@ public FSDataInputStream open(PathHandle fd, int bufferSize) @Override public String getErasureCodingPolicyName(FileStatus fileStatus) { + if (!(fileStatus instanceof HdfsFileStatus)) + return null; return ((HdfsFileStatus) fileStatus).getErasureCodingPolicy().getName(); } diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java index 10133934efc5fa..ba3310fb55aa28 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java @@ -495,14 +495,6 @@ default FileStatus makeQualified(URI defaultUri, Path parent) { String getNamespace(); - /** - * See {@link FileStatus#setAttr(Set)}. - */ - @VisibleForTesting - default void setAttr(Set attr){ - // do nothing (this is used only for testing) - } - /** * Set redundant flags for compatibility with existing applications. */ diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java index 9d2748fc10241a..2b50d63e2f0cf7 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableDirectoryCreateCommand.java @@ -20,7 +20,6 @@ import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.Options; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.WithErasureCoding; import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy; @@ -28,11 +27,11 @@ import org.apache.hadoop.tools.DistCpOptions; import org.apache.hadoop.tools.util.RetriableCommand; import org.apache.hadoop.mapreduce.Mapper; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; - +import static org.apache.hadoop.fs.FileUtil.checkFSSupportsEC; import static org.apache.hadoop.tools.mapred.CopyMapper.getFileAttributeSettings; /** @@ -75,14 +74,12 @@ protected Object doExecute(Object... arguments) throws Exception { boolean preserveEC = getFileAttributeSettings(context) .contains(DistCpOptions.FileAttribute.ERASURECODINGPOLICY); if (preserveEC && sourceStatus.isErasureCoded() - && checkFSSupportsEC(sourceStatus.getPath(), sourceFs) - && checkFSSupportsEC(target, targetFS)) { + && checkFSSupportsEC(sourceFs, sourceStatus.getPath()) + && checkFSSupportsEC(targetFS, target)) { ErasureCodingPolicy ecPolicy = SystemErasureCodingPolicies.getByName( ((WithErasureCoding) sourceFs).getErasureCodingPolicyName( sourceStatus)); - if (LOG.isDebugEnabled()) { - LOG.debug("EC Policy for source path is {}", ecPolicy); - } + LOG.debug("EC Policy for source path is {}", ecPolicy); WithErasureCoding ecFs = (WithErasureCoding) targetFS; if (ecPolicy != null) { ecFs.setErasureCodingPolicy(target, ecPolicy.getName()); @@ -90,17 +87,4 @@ && checkFSSupportsEC(target, targetFS)) { } return true; } - - /** - * Return true if the FS implements {@link WithErasureCoding} and - * supports EC_POLICY option in {@link Options.OpenFileOptions} - */ - boolean checkFSSupportsEC(Path path, FileSystem fs) throws IOException { - if (fs instanceof WithErasureCoding && fs.hasPathCapability(path, - Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { - return true; - } - LOG.warn("FS with scheme " + fs.getScheme() + " does not support EC"); - return false; - } } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java index 25bc7dd4964667..48677850e1cb75 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java @@ -34,7 +34,6 @@ import org.apache.hadoop.fs.FileChecksum; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FSDataOutputStreamBuilder; -import org.apache.hadoop.fs.Options; import org.apache.hadoop.fs.Options.ChecksumOpt; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.permission.FsPermission; @@ -52,6 +51,7 @@ import org.apache.hadoop.classification.VisibleForTesting; +import static org.apache.hadoop.fs.FileUtil.checkFSSupportsEC; import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY; import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_SEQUENTIAL; import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY; @@ -208,8 +208,8 @@ private long copyToFile(Path targetPath, FileSystem targetFS, String ecPolicyName = null; if (preserveEC && sourceStatus.isErasureCoded() - && checkFSSupportsEC(sourceStatus.getPath(), sourceFS) - && checkFSSupportsEC(targetPath, targetFS)) { + && checkFSSupportsEC(sourceFS,sourceStatus.getPath()) + && checkFSSupportsEC(targetFS, targetPath)) { ecPolicyName = ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); } final OutputStream outStream; @@ -403,19 +403,6 @@ private static long getBlockSize( .getDefaultBlockSize(tmpTargetPath); } - /** - * Return true if the FS implements {@link WithErasureCoding} and - * supports EC_POLICY option in {@link Options.OpenFileOptions} - */ - boolean checkFSSupportsEC(Path path, FileSystem fs) throws IOException { - if (fs instanceof WithErasureCoding && fs.hasPathCapability(path, - Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { - return true; - } - LOG.warn("FS with scheme " + fs.getScheme() + " does not support EC"); - return false; - } - /** * Special subclass of IOException. This is used to distinguish read-operation * failures from other kinds of IOExceptions. diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java index 9711180fc011b6..6c6e5e78b9021f 100644 --- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java +++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java @@ -26,6 +26,7 @@ import java.util.Set; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.contract.ContractTestUtils; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.LocalFileSystem; @@ -268,7 +269,7 @@ public void testPreserveECAcrossFilesystems() throws Exception{ dfs.setErasureCodingPolicy(dir1, "XOR-2-1-1024k"); fs.create(file1).close(); int res = ToolRunner.run(conf, new ECAdmin(conf), args); - assertEquals("Unable to set EC policy on " + subDir1.toString(), res, 0); + assertEquals("Unable to set EC policy on " + subDir1.toString(), 0, res); String src = "/src/*"; Path dest = new Path(TEST_ROOT_DIR, "dest"); final Path dest2Dir1 = new Path(dest, "dir1"); @@ -283,11 +284,13 @@ public void testPreserveECAcrossFilesystems() throws Exception{ try { FileStatus destDir1Status = dummyEcFs.getFileStatus(dest2Dir1); FileStatus destSubDir1Status = dummyEcFs.getFileStatus(dest2SubDir1); - assertNotNull(destDir1Status); - assertNotNull(destSubDir1Status); + assertNotNull("FileStatus for path: " + dest2Dir1 + " is null", destDir1Status); + assertNotNull("FileStatus for path: " + dest2SubDir1 + " is null", destSubDir1Status); // check if target paths are erasure coded. - assertTrue(dummyEcFs.isPathErasureCoded(destDir1Status.getPath())); - assertTrue(dummyEcFs.isPathErasureCoded(destSubDir1Status.getPath())); + assertTrue("Path is not erasure coded : " + dest2Dir1, + dummyEcFs.isPathErasureCoded(destDir1Status.getPath())); + assertTrue("Path is not erasure coded : " + dest2SubDir1, + dummyEcFs.isPathErasureCoded(destSubDir1Status.getPath())); // copy source(DummyECFS) to target (HDFS) String dfsTarget = "/dest"; @@ -295,9 +298,13 @@ public void testPreserveECAcrossFilesystems() throws Exception{ target.toString(), dfsTarget, "-pe", conf); Path dfsTargetPath = new Path(dfsTarget); Path dfsTargetDir1 = new Path(dfsTarget, "dir1"); - assertTrue(fs.exists(dfsTargetPath)); - assertTrue(fs.exists(dfsTargetDir1)); - assertTrue(fs.getFileStatus(dfsTargetDir1).isErasureCoded()); + ContractTestUtils.assertPathExists(fs, + "Path doesn't exist:" + dfsTargetPath, dfsTargetPath); + ContractTestUtils.assertPathExists(fs, + "Path doesn't exist:" + dfsTargetDir1, dfsTargetDir1); + FileStatus targetDir1Status = fs.getFileStatus(dfsTargetDir1); + assertTrue("Path is not erasure coded : " + targetDir1Status, + targetDir1Status.isErasureCoded()); fs.delete(dfsTargetPath, true); } finally { dummyEcFs.delete(new Path(base.getAbsolutePath()),true); @@ -311,7 +318,7 @@ public void testPreserveECAcrossFilesystems() throws Exception{ */ public static class DummyEcFs extends LocalFileSystem implements WithErasureCoding { - private Set erasureCodedPaths; + private Set erasureCodedPaths; public DummyEcFs() { super(); this.erasureCodedPaths = new HashSet<>(); @@ -336,12 +343,19 @@ public boolean hasPathCapability(Path path, String capability) @Override public FileStatus getFileStatus(Path f) throws IOException { FileStatus fileStatus = super.getFileStatus(f); - if (erasureCodedPaths.contains(f)) { - Set attrSet = new HashSet<>(); - attrSet.add(FileStatus.AttrFlags.HAS_EC); - fileStatus.setAttr(attrSet); + if (!erasureCodedPaths.contains(f)) { + return fileStatus; } - return fileStatus; + Set attrSet = new HashSet<>(); + attrSet.add(FileStatus.AttrFlags.HAS_EC); + return new FileStatus(fileStatus.getLen(), fileStatus.isDirectory(), + fileStatus.getReplication(), fileStatus.getBlockSize(), + fileStatus.getModificationTime(), fileStatus.getAccessTime(), + fileStatus.getPermission(), fileStatus.getOwner(), + fileStatus.getGroup(), + fileStatus.isSymlink() ? fileStatus.getSymlink() : null, + fileStatus.getPath(), + attrSet); } @Override From 04e281f2a9040e9f987d42438caa7cae159f16c8 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Tue, 10 Sep 2024 17:26:41 +0530 Subject: [PATCH 10/13] left out changes --- .../src/main/java/org/apache/hadoop/fs/FileStatus.java | 3 +-- .../src/main/java/org/apache/hadoop/fs/WithErasureCoding.java | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java index a7e1390cff880f..1d9458148e4a55 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileStatus.java @@ -27,7 +27,6 @@ import java.util.EnumSet; import java.util.Set; -import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.fs.FSProtos.FileStatusProto; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; @@ -352,7 +351,7 @@ public String getGroup() { public Path getPath() { return path; } - + public void setPath(final Path p) { path = p; } diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java index cf72af27564983..5f8a7fbad6ea32 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/WithErasureCoding.java @@ -39,7 +39,7 @@ public interface WithErasureCoding { /** * Set the given ecPolicy on the path. * The path and ecPolicyName should be valid (not null/empty, the - * implementing FS should support the supplied ecPolicy). + * implementing FS shall support the supplied ecPolicy). * implementations can throw IOException if these conditions are not met. * @param path on which the EC policy needs to be set. * @param ecPolicyName the EC policy. From 44ee8e1d18df1bc1735397601cbd8c4453ff54a0 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Tue, 10 Sep 2024 17:33:34 +0530 Subject: [PATCH 11/13] remove import --- .../java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java | 1 - 1 file changed, 1 deletion(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java index ba3310fb55aa28..efc3b90b5a9706 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsFileStatus.java @@ -27,7 +27,6 @@ import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; -import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.fs.FileEncryptionInfo; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileStatus.AttrFlags; From 3c0b1fb6757a53640167af52c3826258bd7cf21b Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Wed, 11 Sep 2024 00:04:19 +0530 Subject: [PATCH 12/13] fix javadoc --- .../src/main/java/org/apache/hadoop/fs/FileUtil.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java index c2634af3bb8505..842d8071c39767 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java @@ -2112,6 +2112,10 @@ public static void maybeIgnoreMissingDirectory(FileSystem fs, /** * Return true if the FS implements {@link WithErasureCoding} and * supports EC_POLICY option in {@link Options.OpenFileOptions} + * @param fs filesystem + * @param path path + * @return true if the Filesystem supports EC + * @throws IOException if there is a failure in hasPathCapability call */ public static boolean checkFSSupportsEC(FileSystem fs, Path path) throws IOException { if (fs instanceof WithErasureCoding && From 6b056ffc4a183f7e6d128e81cf48209298bf2303 Mon Sep 17 00:00:00 2001 From: Sadanand Shenoy Date: Tue, 24 Sep 2024 21:36:24 +0530 Subject: [PATCH 13/13] address comments --- .../src/main/java/org/apache/hadoop/fs/FileUtil.java | 6 ++++-- .../java/org/apache/hadoop/hdfs/DistributedFileSystem.java | 3 ++- .../hadoop/tools/mapred/RetriableFileCopyCommand.java | 2 +- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java index 842d8071c39767..0d5cf48c8b20c4 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java @@ -2111,7 +2111,8 @@ public static void maybeIgnoreMissingDirectory(FileSystem fs, /** * Return true if the FS implements {@link WithErasureCoding} and - * supports EC_POLICY option in {@link Options.OpenFileOptions} + * supports EC_POLICY option in {@link Options.OpenFileOptions}. + * A message is logged when the filesystem does not support Erasure coding. * @param fs filesystem * @param path path * @return true if the Filesystem supports EC @@ -2122,7 +2123,8 @@ public static boolean checkFSSupportsEC(FileSystem fs, Path path) throws IOExcep fs.hasPathCapability(path, Options.OpenFileOptions.FS_OPTION_OPENFILE_EC_POLICY)) { return true; } - LOG.warn("FS with scheme {} does not support EC", fs.getScheme()); + LOG.warn("Filesystem with scheme {} does not support Erasure Coding" + + " at path {}", fs.getScheme(), path); return false; } } diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java index 219888dcbec271..dac205158d0f42 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java @@ -380,8 +380,9 @@ public FSDataInputStream open(PathHandle fd, int bufferSize) @Override public String getErasureCodingPolicyName(FileStatus fileStatus) { - if (!(fileStatus instanceof HdfsFileStatus)) + if (!(fileStatus instanceof HdfsFileStatus)) { return null; + } return ((HdfsFileStatus) fileStatus).getErasureCodingPolicy().getName(); } diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java index 48677850e1cb75..fc3109ee2cecd1 100644 --- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java +++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java @@ -208,7 +208,7 @@ private long copyToFile(Path targetPath, FileSystem targetFS, String ecPolicyName = null; if (preserveEC && sourceStatus.isErasureCoded() - && checkFSSupportsEC(sourceFS,sourceStatus.getPath()) + && checkFSSupportsEC(sourceFS, sourceStatus.getPath()) && checkFSSupportsEC(targetFS, targetPath)) { ecPolicyName = ((WithErasureCoding) sourceFS).getErasureCodingPolicyName(sourceStatus); }