From f3be380094ed07f602defebcfb0a30c79365380f Mon Sep 17 00:00:00 2001 From: "QI, Jiale" Date: Wed, 1 Jul 2026 18:47:04 +0800 Subject: [PATCH] HDFS: Improve FedBalance DistCp option handling --- .../hdfs/rbfbalance/RouterFedBalance.java | 69 +++++++ .../tools/fedbalance/DistCpProcedure.java | 131 +++++++++++++- .../hadoop/tools/fedbalance/FedBalance.java | 72 +++++++- .../tools/fedbalance/FedBalanceConfigs.java | 3 + .../tools/fedbalance/FedBalanceContext.java | 169 +++++++++++++++++- .../tools/fedbalance/FedBalanceOptions.java | 37 ++++ .../resources/hdfs-fedbalance-default.xml | 10 ++ .../site/markdown/HDFSFederationBalance.md | 7 +- .../tools/fedbalance/TestDistCpProcedure.java | 67 +++++++ .../tools/fedbalance/TestFedBalance.java | 61 ++++++- 10 files changed, 615 insertions(+), 11 deletions(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/rbfbalance/RouterFedBalance.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/rbfbalance/RouterFedBalance.java index 0cb4b54bfc431a..5d8e083f43e93b 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/rbfbalance/RouterFedBalance.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/rbfbalance/RouterFedBalance.java @@ -51,12 +51,18 @@ import static org.apache.hadoop.tools.fedbalance.FedBalance.FED_BALANCE_DEFAULT_XML; import static org.apache.hadoop.tools.fedbalance.FedBalance.FED_BALANCE_SITE_XML; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.CLI_OPTIONS; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DISTCP_STRATEGY; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.FORCE_CLOSE_OPEN; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.LIST_STATUS_THREADS; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.MAP; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.PRESERVE_TIMES; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.SKIP_ACL_PRESERVE; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.BANDWIDTH; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DELAY_DURATION; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DIFF_THRESHOLD; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.TRASH; +import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED; +import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED_DEFAULT; /** * Balance data in router-based federation cluster. From src sub-namespace to @@ -92,6 +98,14 @@ private class Builder { private long delayDuration = TimeUnit.SECONDS.toMillis(1); /* Specify the threshold of diff entries. */ private int diffThreshold = 0; + /* Whether to preserve ACLs in submitted DistCp jobs. */ + private boolean preserveAcl = true; + /* Whether to preserve modification/access times in submitted DistCp jobs. */ + private boolean preserveTimes = false; + /* DistCp copy strategy. */ + private String distCpStrategy; + /* Number of DistCp listStatus threads. */ + private int numListstatusThreads = 0; /* The source input. This specifies the source path. */ private final String inputSrc; /* The dst input. This specifies the dst path. */ @@ -156,6 +170,42 @@ public Builder setDiffThreshold(int value) { return this; } + /** + * Specify whether ACLs should be preserved by DistCp. + * @param value true if preserving ACLs. + */ + public Builder setPreserveAcl(boolean value) { + this.preserveAcl = value; + return this; + } + + /** + * Specify whether file times should be preserved by DistCp. + * @param value true if preserving file times. + */ + public Builder setPreserveTimes(boolean value) { + this.preserveTimes = value; + return this; + } + + /** + * Specify the DistCp copy strategy. + * @param value the DistCp copy strategy. + */ + public Builder setDistCpStrategy(String value) { + this.distCpStrategy = value; + return this; + } + + /** + * Specify the DistCp listStatus thread count. + * @param value the DistCp listStatus thread count. + */ + public Builder setNumListstatusThreads(int value) { + this.numListstatusThreads = value; + return this; + } + /** * Build the balance job. */ @@ -172,6 +222,11 @@ public BalanceJob build() throws IOException { .setForceCloseOpenFiles(forceCloseOpen).setUseMountReadOnly(true) .setMapNum(map).setBandwidthLimit(bandwidth).setTrash(trashOpt) .setDelayDuration(delayDuration).setDiffThreshold(diffThreshold) + .setPreserveAcl(preserveAcl && getConf().getBoolean( + PRESERVE_ACL_ENABLED, PRESERVE_ACL_ENABLED_DEFAULT)) + .setPreserveTimes(preserveTimes) + .setDistCpStrategy(distCpStrategy) + .setNumListstatusThreads(numListstatusThreads) .build(); LOG.info(context.toString()); @@ -281,6 +336,20 @@ private int submit(CommandLine command, String inputSrc, String inputDst) builder.setDiffThreshold(Integer.parseInt( command.getOptionValue(DIFF_THRESHOLD.getOpt()))); } + if (command.hasOption(SKIP_ACL_PRESERVE.getOpt())) { + builder.setPreserveAcl(false); + } + if (command.hasOption(PRESERVE_TIMES.getOpt())) { + builder.setPreserveTimes(true); + } + if (command.hasOption(DISTCP_STRATEGY.getOpt())) { + builder.setDistCpStrategy(command.getOptionValue( + DISTCP_STRATEGY.getOpt())); + } + if (command.hasOption(LIST_STATUS_THREADS.getOpt())) { + builder.setNumListstatusThreads(Integer.parseInt( + command.getOptionValue(LIST_STATUS_THREADS.getOpt()))); + } if (command.hasOption(TRASH.getOpt())) { String val = command.getOptionValue(TRASH.getOpt()); if (val.equalsIgnoreCase("skip")) { diff --git a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/DistCpProcedure.java b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/DistCpProcedure.java index 936a4e1d2649cd..942ac4679ce56b 100644 --- a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/DistCpProcedure.java +++ b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/DistCpProcedure.java @@ -25,6 +25,7 @@ import org.apache.hadoop.fs.permission.AclStatus; import org.apache.hadoop.fs.permission.FsPermission; import org.apache.hadoop.hdfs.DistributedFileSystem; +import org.apache.hadoop.hdfs.protocol.AclException; import org.apache.hadoop.hdfs.protocol.HdfsConstants; import org.apache.hadoop.hdfs.protocol.OpenFileEntry; import org.apache.hadoop.hdfs.protocol.OpenFilesIterator.OpenFilesType; @@ -91,9 +92,21 @@ public enum Stage { private boolean useMountReadOnly; /* The threshold of diff entries. */ private int diffThreshold; + /* Whether to preserve ACLs in submitted DistCp jobs. */ + private boolean preserveAcl; + /* Whether to preserve modification/access times in submitted DistCp jobs. */ + private boolean preserveTimes; + /* DistCp copy strategy. */ + private String distCpStrategy; + /* Number of DistCp listStatus threads. */ + private int numListstatusThreads; private FsPermission fPerm; // the permission of the src. private AclStatus acl; // the acl of the src. + /* Cached result of whether ACL is supported on srcFs. Null until checked. */ + private Boolean srcAclSupported; + /* Cached result of whether ACL is supported on dstFs. Null until checked. */ + private Boolean dstAclSupported; private JobClient client; private DistributedFileSystem srcFs; // fs of the src cluster. @@ -145,6 +158,10 @@ public DistCpProcedure(String name, String nextProcedure, long delayDuration, this.forceCloseOpenFiles = context.getForceCloseOpenFiles(); this.useMountReadOnly = context.getUseMountReadOnly(); this.diffThreshold = context.getDiffThreshold(); + this.preserveAcl = context.getPreserveAcl(); + this.preserveTimes = context.getPreserveTimes(); + this.distCpStrategy = context.getDistCpStrategy(); + this.numListstatusThreads = context.getNumListstatusThreads(); srcFs = (DistributedFileSystem) context.getSrc().getFileSystem(conf); dstFs = (DistributedFileSystem) context.getDst().getFileSystem(conf); } @@ -255,7 +272,7 @@ protected void disableWrite(FedBalanceContext fbcontext) throws IOException { // Save and cancel permission. FileStatus status = srcFs.getFileStatus(src); fPerm = status.getPermission(); - acl = srcFs.getAclStatus(src); + acl = getAclStatusIfSupported(); srcFs.setPermission(src, FsPermission.createImmutable((short) 0)); updateStage(Stage.FINAL_DISTCP); } @@ -273,15 +290,90 @@ protected void enableWrite() throws IOException { */ void restorePermission() throws IOException { // restore permission. - dstFs.removeAcl(dst); - if (acl != null) { - dstFs.modifyAclEntries(dst, acl.getEntries()); + removeAclIfSupported(); + if (preserveAcl && acl != null) { + try { + dstFs.modifyAclEntries(dst, acl.getEntries()); + } catch (AclException e) { + LOG.info("ACL is not enabled for {}. Skip ACL restore.", dst, e); + } } if (fPerm != null) { dstFs.setPermission(dst, fPerm); } } + private AclStatus getAclStatusIfSupported() throws IOException { + if (!shouldPreserveAcl()) { + return null; + } + return srcFs.getAclStatus(src); + } + + private void removeAclIfSupported() throws IOException { + if (!preserveAcl || acl == null || !isDstAclSupported()) { + return; + } + try { + dstFs.removeAcl(dst); + } catch (AclException e) { + LOG.info("ACL is not enabled for {}. Skip ACL restore.", dst, e); + } + } + + /** + * Whether ACL is supported on the source filesystem. The result is cached + * after the first check since ACL support doesn't change during a job. + */ + private boolean isSrcAclSupported() throws IOException { + if (srcAclSupported == null) { + try { + srcFs.getAclStatus(src); + srcAclSupported = true; + } catch (AclException e) { + LOG.info("ACL is not enabled for {}. Skip ACL preserve.", src, e); + srcAclSupported = false; + } + } + return srcAclSupported; + } + + /** + * Whether ACL is supported on the destination filesystem. The destination + * path may not exist before the initial DistCp, so probe the nearest + * existing parent path. + */ + private boolean isDstAclSupported() throws IOException { + if (dstAclSupported == null) { + Path probePath = getAclProbePath(dstFs, dst); + try { + dstFs.getAclStatus(probePath); + dstAclSupported = true; + } catch (AclException e) { + LOG.info("ACL is not enabled for {}. Skip ACL preserve.", probePath, + e); + dstAclSupported = false; + } + } + return dstAclSupported; + } + + private Path getAclProbePath(DistributedFileSystem fs, Path path) + throws IOException { + Path probePath = path; + while (probePath != null) { + if (fs.exists(probePath)) { + return probePath; + } + probePath = probePath.getParent(); + } + return new Path(Path.SEPARATOR); + } + + private boolean shouldPreserveAcl() throws IOException { + return preserveAcl && isSrcAclSupported() && isDstAclSupported(); + } + /** * Close all open files then submit the distcp with -diff. */ @@ -475,8 +567,8 @@ private void pathCheckBeforeInitDistcp() throws IOException { private String submitDistCpJob(String srcParam, String dstParam, boolean useSnapshotDiff) throws IOException { List command = new ArrayList<>(); - command.addAll(Arrays - .asList(new String[] {"-async", "-update", "-append", "-pruxgpcab"})); + command.addAll(Arrays.asList( + new String[] {"-async", "-update", "-append", getPreserveOption()})); if (useSnapshotDiff) { command.add("-diff"); command.add(LAST_SNAPSHOT_NAME); @@ -486,6 +578,14 @@ private String submitDistCpJob(String srcParam, String dstParam, command.add(mapNum + ""); command.add("-bandwidth"); command.add(bandWidth + ""); + if (numListstatusThreads > 0) { + command.add("-numListstatusThreads"); + command.add(numListstatusThreads + ""); + } + if (distCpStrategy != null && !distCpStrategy.isEmpty()) { + command.add("-strategy"); + command.add(distCpStrategy); + } command.add(srcParam); command.add(dstParam); @@ -505,6 +605,18 @@ private String submitDistCpJob(String srcParam, String dstParam, } } + @VisibleForTesting + String getPreserveOption() throws IOException { + StringBuilder preserve = new StringBuilder("-pruxgpcb"); + if (shouldPreserveAcl()) { + preserve.append('a'); + } + if (preserveTimes) { + preserve.append('t'); + } + return preserve.toString(); + } + @Override public void write(DataOutput out) throws IOException { super.write(out); @@ -564,6 +676,11 @@ public void readFields(DataInput in) throws IOException { bandWidth = context.getBandwidthLimit(); forceCloseOpenFiles = context.getForceCloseOpenFiles(); useMountReadOnly = context.getUseMountReadOnly(); + diffThreshold = context.getDiffThreshold(); + preserveAcl = context.getPreserveAcl(); + preserveTimes = context.getPreserveTimes(); + distCpStrategy = context.getDistCpStrategy(); + numListstatusThreads = context.getNumListstatusThreads(); this.client = new JobClient(conf); } @@ -666,4 +783,4 @@ public String getFailureInfo() throws IOException { } } } -} \ No newline at end of file +} diff --git a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalance.java b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalance.java index 1249138f955a33..b68f0187f3f9f6 100644 --- a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalance.java +++ b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalance.java @@ -38,8 +38,14 @@ import java.util.Collection; import java.util.concurrent.TimeUnit; +import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED; +import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED_DEFAULT; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DISTCP_STRATEGY; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.FORCE_CLOSE_OPEN; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.LIST_STATUS_THREADS; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.MAP; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.PRESERVE_TIMES; +import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.SKIP_ACL_PRESERVE; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.BANDWIDTH; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.TRASH; import static org.apache.hadoop.tools.fedbalance.FedBalanceOptions.DELAY_DURATION; @@ -83,6 +89,14 @@ private final class Builder { private long delayDuration = TimeUnit.SECONDS.toMillis(1); /* Specify the threshold of diff entries. */ private int diffThreshold = 0; + /* Whether to preserve ACLs in submitted DistCp jobs. */ + private boolean preserveAcl = true; + /* Whether to preserve modification/access times in submitted DistCp jobs. */ + private boolean preserveTimes = false; + /* DistCp copy strategy. */ + private String distCpStrategy; + /* Number of DistCp listStatus threads. */ + private int numListstatusThreads = 0; /* The source input. This specifies the source path. */ private final String inputSrc; /* The dst input. This specifies the dst path. */ @@ -147,6 +161,42 @@ public Builder setDiffThreshold(int value) { return this; } + /** + * Specify whether ACLs should be preserved by DistCp. + * @param value true if preserving ACLs. + */ + public Builder setPreserveAcl(boolean value) { + this.preserveAcl = value; + return this; + } + + /** + * Specify whether file times should be preserved by DistCp. + * @param value true if preserving file times. + */ + public Builder setPreserveTimes(boolean value) { + this.preserveTimes = value; + return this; + } + + /** + * Specify the DistCp copy strategy. + * @param value the DistCp copy strategy. + */ + public Builder setDistCpStrategy(String value) { + this.distCpStrategy = value; + return this; + } + + /** + * Specify the DistCp listStatus thread count. + * @param value the DistCp listStatus thread count. + */ + public Builder setNumListstatusThreads(int value) { + this.numListstatusThreads = value; + return this; + } + /** * Build the balance job. */ @@ -164,7 +214,13 @@ public BalanceJob build() throws IOException { context = new FedBalanceContext.Builder(src, dst, NO_MOUNT, getConf()) .setForceCloseOpenFiles(forceCloseOpen).setUseMountReadOnly(false) .setMapNum(map).setBandwidthLimit(bandwidth).setTrash(trashOpt) - .setDiffThreshold(diffThreshold).build(); + .setDiffThreshold(diffThreshold) + .setPreserveAcl(preserveAcl && getConf().getBoolean( + PRESERVE_ACL_ENABLED, PRESERVE_ACL_ENABLED_DEFAULT)) + .setPreserveTimes(preserveTimes) + .setDistCpStrategy(distCpStrategy) + .setNumListstatusThreads(numListstatusThreads) + .build(); LOG.info(context.toString()); // Construct the balance job. @@ -268,6 +324,20 @@ private int submit(CommandLine command, String inputSrc, String inputDst) builder.setDiffThreshold(Integer.parseInt( command.getOptionValue(DIFF_THRESHOLD.getOpt()))); } + if (command.hasOption(SKIP_ACL_PRESERVE.getOpt())) { + builder.setPreserveAcl(false); + } + if (command.hasOption(PRESERVE_TIMES.getOpt())) { + builder.setPreserveTimes(true); + } + if (command.hasOption(DISTCP_STRATEGY.getOpt())) { + builder.setDistCpStrategy(command.getOptionValue( + DISTCP_STRATEGY.getOpt())); + } + if (command.hasOption(LIST_STATUS_THREADS.getOpt())) { + builder.setNumListstatusThreads(Integer.parseInt( + command.getOptionValue(LIST_STATUS_THREADS.getOpt()))); + } if (command.hasOption(TRASH.getOpt())) { String val = command.getOptionValue(TRASH.getOpt()); if (val.equalsIgnoreCase("skip")) { diff --git a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceConfigs.java b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceConfigs.java index efe906bbc78016..8596d07cfac4a7 100644 --- a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceConfigs.java +++ b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceConfigs.java @@ -44,6 +44,9 @@ public enum TrashOption { "hdfs.fedbalance.procedure.scheduler.journal.uri"; public static final String JOB_PREFIX = "JOB-"; public static final String TMP_TAIL = ".tmp"; + public static final String PRESERVE_ACL_ENABLED = + "hdfs.fedbalance.preserve.acl.enabled"; + public static final boolean PRESERVE_ACL_ENABLED_DEFAULT = true; private FedBalanceConfigs(){} } diff --git a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceContext.java b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceContext.java index 2a49ecc9e60b6e..c9bed76b95cd52 100644 --- a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceContext.java +++ b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceContext.java @@ -56,47 +56,141 @@ public class FedBalanceContext implements Writable { private long delayDuration; /* The threshold of diff entries. */ private int diffThreshold; + /* Whether to preserve ACLs in submitted DistCp jobs. */ + private boolean preserveAcl = true; + /* Whether to preserve modification/access times in submitted DistCp jobs. */ + private boolean preserveTimes; + /* DistCp copy strategy. */ + private String distCpStrategy; + /* Number of DistCp listStatus threads. */ + private int numListstatusThreads; private Configuration conf; public FedBalanceContext() {} + /** + * Get the configuration used by this context. + * + * @return configuration used by this context. + */ public Configuration getConf() { return conf; } + /** + * Get the source path. + * + * @return source path. + */ public Path getSrc() { return src; } + /** + * Get the destination path. + * + * @return destination path. + */ public Path getDst() { return dst; } + /** + * Get the mount point. + * + * @return mount point. + */ public String getMount() { return mount; } + /** + * Get whether open files should be force closed. + * + * @return true if open files should be force closed. + */ public boolean getForceCloseOpenFiles() { return forceCloseOpenFiles; } + /** + * Get whether the mount point should be made read-only. + * + * @return true if the mount point should be made read-only. + */ public boolean getUseMountReadOnly() { return useMountReadOnly; } + /** + * Get the maximum number of maps for submitted DistCp jobs. + * + * @return maximum number of maps for submitted DistCp jobs. + */ public int getMapNum() { return mapNum; } + /** + * Get the per-map bandwidth limit in MB. + * + * @return per-map bandwidth limit in MB. + */ public int getBandwidthLimit() { return bandwidthLimit; } + /** + * Get the snapshot diff threshold. + * + * @return snapshot diff threshold. + */ public int getDiffThreshold() { return diffThreshold; } + /** + * Get whether ACLs should be preserved when supported. + * + * @return true if ACLs should be preserved when supported. + */ + public boolean getPreserveAcl() { + return preserveAcl; + } + + /** + * Get whether file timestamps should be preserved. + * + * @return true if file timestamps should be preserved. + */ + public boolean getPreserveTimes() { + return preserveTimes; + } + + /** + * Get the DistCp copy strategy. + * + * @return DistCp copy strategy, or null to use the DistCp default. + */ + public String getDistCpStrategy() { + return distCpStrategy; + } + + /** + * Get the number of DistCp listStatus threads. + * + * @return number of DistCp listStatus threads, or 0 for DistCp default. + */ + public int getNumListstatusThreads() { + return numListstatusThreads; + } + + /** + * Get the source trash behavior. + * + * @return source trash behavior. + */ public TrashOption getTrashOpt() { return trashOpt; } @@ -114,6 +208,10 @@ public void write(DataOutput out) throws IOException { out.writeInt(trashOpt.ordinal()); out.writeLong(delayDuration); out.writeInt(diffThreshold); + out.writeBoolean(preserveAcl); + out.writeBoolean(preserveTimes); + Text.writeString(out, distCpStrategy == null ? "" : distCpStrategy); + out.writeInt(numListstatusThreads); } @Override @@ -130,6 +228,14 @@ public void readFields(DataInput in) throws IOException { trashOpt = TrashOption.values()[in.readInt()]; delayDuration = in.readLong(); diffThreshold = in.readInt(); + preserveAcl = in.readBoolean(); + preserveTimes = in.readBoolean(); + distCpStrategy = emptyToNull(Text.readString(in)); + numListstatusThreads = in.readInt(); + } + + private static String emptyToNull(String value) { + return value == null || value.isEmpty() ? null : value; } @Override @@ -155,6 +261,10 @@ public boolean equals(Object obj) { .append(trashOpt, bc.trashOpt) .append(delayDuration, bc.delayDuration) .append(diffThreshold, bc.diffThreshold) + .append(preserveAcl, bc.preserveAcl) + .append(preserveTimes, bc.preserveTimes) + .append(distCpStrategy, bc.distCpStrategy) + .append(numListstatusThreads, bc.numListstatusThreads) .isEquals(); } @@ -171,6 +281,10 @@ public int hashCode() { .append(trashOpt) .append(delayDuration) .append(diffThreshold) + .append(preserveAcl) + .append(preserveTimes) + .append(distCpStrategy) + .append(numListstatusThreads) .build(); } @@ -204,6 +318,11 @@ public String toString() { break; } builder.append(" Delay duration is ").append(delayDuration).append("ms."); + builder.append(" Preserve ACL is ").append(preserveAcl).append("."); + builder.append(" Preserve times is ").append(preserveTimes).append("."); + builder.append(" DistCp strategy is ").append(distCpStrategy).append("."); + builder.append(" DistCp listStatus threads is ") + .append(numListstatusThreads).append("."); return builder.toString(); } @@ -219,6 +338,10 @@ public static class Builder { private TrashOption trashOpt; private long delayDuration; private int diffThreshold; + private boolean preserveAcl = true; + private boolean preserveTimes; + private String distCpStrategy; + private int numListstatusThreads; /** * This class helps building the FedBalanceContext. @@ -305,6 +428,46 @@ public Builder setDiffThreshold(int value) { return this; } + /** + * Specify whether ACLs should be preserved by DistCp. + * @param value true if preserving ACLs. + * @return the builder. + */ + public Builder setPreserveAcl(boolean value) { + this.preserveAcl = value; + return this; + } + + /** + * Specify whether file times should be preserved by DistCp. + * @param value true if preserving file times. + * @return the builder. + */ + public Builder setPreserveTimes(boolean value) { + this.preserveTimes = value; + return this; + } + + /** + * Specify the DistCp copy strategy. + * @param value the DistCp copy strategy. + * @return the builder. + */ + public Builder setDistCpStrategy(String value) { + this.distCpStrategy = emptyToNull(value); + return this; + } + + /** + * Specify the DistCp listStatus thread count. + * @param value the DistCp listStatus thread count. + * @return the builder. + */ + public Builder setNumListstatusThreads(int value) { + this.numListstatusThreads = value; + return this; + } + /** * Build the FedBalanceContext. * @@ -323,7 +486,11 @@ public FedBalanceContext build() { context.trashOpt = this.trashOpt; context.delayDuration = this.delayDuration; context.diffThreshold = this.diffThreshold; + context.preserveAcl = this.preserveAcl; + context.preserveTimes = this.preserveTimes; + context.distCpStrategy = this.distCpStrategy; + context.numListstatusThreads = this.numListstatusThreads; return context; } } -} \ No newline at end of file +} diff --git a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceOptions.java b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceOptions.java index 4df3f50ebfffe5..3be82c6d2c77bb 100644 --- a/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceOptions.java +++ b/hadoop-tools/hadoop-federation-balance/src/main/java/org/apache/hadoop/tools/fedbalance/FedBalanceOptions.java @@ -82,6 +82,38 @@ private FedBalanceOptions() {} + " used. If the trash is disabled in the server side, the default" + " trash interval 60 minutes is used."); + /** + * Do not preserve ACLs in the DistCp jobs submitted by FedBalance. + */ + public final static Option SKIP_ACL_PRESERVE = + new Option("skipAclPreserve", false, + "Do not preserve ACLs in the DistCp jobs submitted by FedBalance."); + + /** + * Preserve file timestamps in the DistCp jobs submitted by FedBalance. + */ + public final static Option PRESERVE_TIMES = + new Option("preserveTimes", false, + "Preserve file timestamps in the DistCp jobs submitted by " + + "FedBalance."); + + /** + * Set the DistCp copy strategy for FedBalance submitted jobs. + */ + public final static Option DISTCP_STRATEGY = + new Option("distcpStrategy", true, + "Set the DistCp copy strategy for FedBalance submitted jobs."); + + /** + * Set DistCp listStatus threads for FedBalance submitted jobs. + */ + public final static Option LIST_STATUS_THREADS = + new Option("numListstatusThreads", true, + "Set DistCp listStatus threads for FedBalance submitted jobs."); + + /** + * FedBalance command line options. + */ public final static Options CLI_OPTIONS = new Options(); static { @@ -90,5 +122,10 @@ private FedBalanceOptions() {} CLI_OPTIONS.addOption(BANDWIDTH); CLI_OPTIONS.addOption(DELAY_DURATION); CLI_OPTIONS.addOption(TRASH); + CLI_OPTIONS.addOption(DIFF_THRESHOLD); + CLI_OPTIONS.addOption(SKIP_ACL_PRESERVE); + CLI_OPTIONS.addOption(PRESERVE_TIMES); + CLI_OPTIONS.addOption(DISTCP_STRATEGY); + CLI_OPTIONS.addOption(LIST_STATUS_THREADS); } } diff --git a/hadoop-tools/hadoop-federation-balance/src/main/resources/hdfs-fedbalance-default.xml b/hadoop-tools/hadoop-federation-balance/src/main/resources/hdfs-fedbalance-default.xml index d769832273378b..e76071a260ae3c 100644 --- a/hadoop-tools/hadoop-federation-balance/src/main/resources/hdfs-fedbalance-default.xml +++ b/hadoop-tools/hadoop-federation-balance/src/main/resources/hdfs-fedbalance-default.xml @@ -38,4 +38,14 @@ + + hdfs.fedbalance.preserve.acl.enabled + true + + Whether FedBalance should preserve ACLs in the submitted DistCp + jobs when ACLs are supported by the source and destination + filesystems. + + + diff --git a/hadoop-tools/hadoop-federation-balance/src/site/markdown/HDFSFederationBalance.md b/hadoop-tools/hadoop-federation-balance/src/site/markdown/HDFSFederationBalance.md index e665528f7b9a23..8b250e3e10df9e 100644 --- a/hadoop-tools/hadoop-federation-balance/src/site/markdown/HDFSFederationBalance.md +++ b/hadoop-tools/hadoop-federation-balance/src/site/markdown/HDFSFederationBalance.md @@ -100,6 +100,10 @@ Command `submit` has the following options: | -delay | Specify the delayed duration(millie seconds) when the job needs to retry. | 1000 | | -moveToTrash | This options has 3 values: `trash` (move the source path to trash), `delete` (delete the source path directly) and `skip` (skip both trash and deletion). By default the server side trash interval is used. If the trash is disabled in the server side, the default trash interval 60 minutes is used. | trash | | -diffThreshold | Specify the threshold of the diff entries that used in incremental copy stage. If the diff entries size is no greater than the threshold and the open files check is satisfied(no open files or force close all open files), the fedBalance will go to the final round of distcp. Setting to 0 means waiting until there is no diff.| 0 | +| -skipAclPreserve | Do not preserve ACLs in the DistCp jobs submitted by FedBalance. | ACLs are preserved when supported by the source filesystem. | +| -preserveTimes | Preserve file timestamps in the DistCp jobs submitted by FedBalance. | File timestamps are not preserved. | +| -distcpStrategy | Set the DistCp copy strategy for FedBalance submitted jobs. | DistCp default | +| -numListstatusThreads | Set DistCp listStatus threads for FedBalance submitted jobs. | DistCp default | ### Configuration Options -------------------- @@ -110,6 +114,7 @@ Set configuration options at hdfs-fedbalance-site.xml. | ------------------------------ | ------------------------------------ | ------- | | hdfs.fedbalance.procedure.work.thread.num | The worker threads number of the BalanceProcedureScheduler. BalanceProcedureScheduler is responsible for scheduling a balance job, including submit, run, delay and recover. | 10 | | hdfs.fedbalance.procedure.scheduler.journal.uri | The uri of the journal, the journal file is used for handling the job persistence and recover. | hdfs://localhost:8020/tmp/procedure | +| hdfs.fedbalance.preserve.acl.enabled | Whether FedBalance should preserve ACLs in the submitted DistCp jobs when ACLs are supported by the source and destination filesystems. | true | Architecture of HDFS Federation Balance ---------------------- @@ -167,4 +172,4 @@ Architecture of HDFS Federation Balance * TrashProcedure: This procedure moves the source path to trash. - After all 3 procedures finish, the balance job is done. \ No newline at end of file + After all 3 procedures finish, the balance job is done. diff --git a/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestDistCpProcedure.java b/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestDistCpProcedure.java index cf3a4ce104dc31..e9114bad5b481a 100644 --- a/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestDistCpProcedure.java +++ b/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestDistCpProcedure.java @@ -20,10 +20,12 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.permission.AclStatus; import org.apache.hadoop.fs.permission.FsPermission; import org.apache.hadoop.hdfs.DFSConfigKeys; import org.apache.hadoop.hdfs.DFSTestUtil; import org.apache.hadoop.hdfs.DistributedFileSystem; +import org.apache.hadoop.hdfs.protocol.AclException; import org.apache.hadoop.hdfs.protocol.HdfsConstants; import org.apache.hadoop.ipc.RemoteException; import org.apache.hadoop.tools.fedbalance.DistCpProcedure.Stage; @@ -31,6 +33,7 @@ import org.apache.hadoop.tools.fedbalance.procedure.BalanceJob; import org.apache.hadoop.tools.fedbalance.procedure.BalanceProcedure.RetryException; import org.apache.hadoop.tools.fedbalance.procedure.BalanceProcedureScheduler; +import org.apache.hadoop.test.Whitebox; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -57,6 +60,9 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; /** * Test DistCpProcedure. @@ -200,6 +206,39 @@ public void testDiffThreshold() throws Exception { cleanup(fs, new Path(testRoot)); } + @Test + public void testDiffThresholdAfterRecovery() throws Exception { + String testRoot = nnUri + "/user/foo/testdir." + getMethodName(); + DistributedFileSystem fs = + (DistributedFileSystem) FileSystem.get(URI.create(nnUri), conf); + createFiles(fs, testRoot, srcfiles); + Path src = new Path(testRoot, SRCDAT); + Path dst = new Path(testRoot, DSTDAT); + + FedBalanceContext context = buildContext(src, dst, MOUNT, 10); + DistCpProcedure dcProcedure = + new DistCpProcedure("distcp-procedure", null, 1000, context); + executeProcedure(dcProcedure, Stage.DIFF_DISTCP, + () -> dcProcedure.initDistCp()); + DistCpProcedure recoveredProcedure = serializeProcedure(dcProcedure); + + Path lastPath = new Path(src, "a"); + for (int i = 0; i < 5; i++) { + Path newPath = new Path(src, "a-" + i); + fs.rename(lastPath, newPath); + lastPath = newPath; + } + assertTrue(recoveredProcedure.diffDistCpStageDone()); + cleanup(fs, new Path(testRoot)); + } + + @Test + public void testPreserveAclRequiresDestinationAclSupport() + throws Exception { + assertEquals("-pruxgpcba", getPreserveOption(true)); + assertEquals("-pruxgpcb", getPreserveOption(false)); + } + @Test public void testDiffDistCp() throws Exception { String testRoot = nnUri + "/user/foo/testdir." + getMethodName(); @@ -391,6 +430,34 @@ private FedBalanceContext buildContext(Path src, Path dst, String mount, .setDiffThreshold(diffThreshold).build(); } + private String getPreserveOption(boolean dstAclSupported) throws IOException { + DistCpProcedure dcProcedure = new DistCpProcedure(); + DistributedFileSystem srcFs = mock(DistributedFileSystem.class); + DistributedFileSystem dstFs = mock(DistributedFileSystem.class); + Path src = new Path("/src"); + Path dst = new Path("/dst"); + Path dstRoot = new Path(Path.SEPARATOR); + AclStatus aclStatus = new AclStatus.Builder().owner("owner") + .group("group").build(); + + doReturn(aclStatus).when(srcFs).getAclStatus(src); + doReturn(false).when(dstFs).exists(dst); + doReturn(true).when(dstFs).exists(dstRoot); + if (dstAclSupported) { + doReturn(aclStatus).when(dstFs).getAclStatus(dstRoot); + } else { + doThrow(new AclException("ACL disabled")).when(dstFs) + .getAclStatus(dstRoot); + } + + Whitebox.setInternalState(dcProcedure, "src", src); + Whitebox.setInternalState(dcProcedure, "dst", dst); + Whitebox.setInternalState(dcProcedure, "srcFs", srcFs); + Whitebox.setInternalState(dcProcedure, "dstFs", dstFs); + Whitebox.setInternalState(dcProcedure, "preserveAcl", true); + return dcProcedure.getPreserveOption(); + } + protected interface Call { void execute() throws IOException, RetryException; } diff --git a/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestFedBalance.java b/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestFedBalance.java index 39cdb1c69d46ab..2170a6e67c8952 100644 --- a/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestFedBalance.java +++ b/hadoop-tools/hadoop-federation-balance/src/test/java/org/apache/hadoop/tools/fedbalance/TestFedBalance.java @@ -18,11 +18,23 @@ package org.apache.hadoop.tools.fedbalance; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; import org.junit.jupiter.api.Test; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.PRESERVE_ACL_ENABLED; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.SCHEDULER_JOURNAL_URI; +import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.TrashOption; import static org.apache.hadoop.tools.fedbalance.FedBalanceConfigs.WORK_THREAD_NUM; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; public class TestFedBalance { @Test @@ -30,5 +42,52 @@ public void testLoadFedBalanceDefaultConf() { Configuration conf = FedBalance.getDefaultConf(); assertNotNull(conf.get(SCHEDULER_JOURNAL_URI)); assertNotNull(conf.get(WORK_THREAD_NUM)); + assertTrue(conf.getBoolean(PRESERVE_ACL_ENABLED, false)); + } + + @Test + public void testFedBalanceOptionsRegistered() { + assertTrue(FedBalanceOptions.CLI_OPTIONS.hasOption( + FedBalanceOptions.DIFF_THRESHOLD.getOpt())); + assertTrue(FedBalanceOptions.CLI_OPTIONS.hasOption( + FedBalanceOptions.SKIP_ACL_PRESERVE.getOpt())); + assertTrue(FedBalanceOptions.CLI_OPTIONS.hasOption( + FedBalanceOptions.PRESERVE_TIMES.getOpt())); + assertTrue(FedBalanceOptions.CLI_OPTIONS.hasOption( + FedBalanceOptions.DISTCP_STRATEGY.getOpt())); + assertTrue(FedBalanceOptions.CLI_OPTIONS.hasOption( + FedBalanceOptions.LIST_STATUS_THREADS.getOpt())); + } + + @Test + public void testFedBalanceContextDistCpOptionSerialization() + throws IOException { + Configuration conf = new Configuration(false); + FedBalanceContext context = new FedBalanceContext.Builder( + new Path("hdfs://src/data"), new Path("hdfs://dst/data"), + "mount", conf) + .setMapNum(1) + .setBandwidthLimit(1) + .setTrash(TrashOption.SKIP) + .setDelayDuration(1) + .setDiffThreshold(3) + .setPreserveAcl(false) + .setPreserveTimes(true) + .setDistCpStrategy("dynamic") + .setNumListstatusThreads(8) + .build(); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + context.write(new DataOutputStream(out)); + + FedBalanceContext recovered = new FedBalanceContext(); + recovered.readFields(new DataInputStream( + new ByteArrayInputStream(out.toByteArray()))); + + assertEquals(3, recovered.getDiffThreshold()); + assertFalse(recovered.getPreserveAcl()); + assertTrue(recovered.getPreserveTimes()); + assertEquals("dynamic", recovered.getDistCpStrategy()); + assertEquals(8, recovered.getNumListstatusThreads()); } -} \ No newline at end of file +}