You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by st...@apache.org on 2018/03/15 18:06:54 UTC
[4/4] hadoop git commit: HADOOP-15209. DistCp to eliminate needless
deletion of files under already-deleted directories. Contributed by Steve
Loughran.
HADOOP-15209. DistCp to eliminate needless deletion of files under already-deleted directories.
Contributed by Steve Loughran.
(cherry picked from commit 1976e0066e9ae8852715fa69d8aea3769330e933)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8f845598
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8f845598
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8f845598
Branch: refs/heads/branch-3.1
Commit: 8f845598193cc15ea3b6a7f6e7a9785e5f1a1cd0
Parents: 13f2ee0
Author: Steve Loughran <st...@apache.org>
Authored: Thu Mar 15 18:06:32 2018 +0000
Committer: Steve Loughran <st...@apache.org>
Committed: Thu Mar 15 18:06:32 2018 +0000
----------------------------------------------------------------------
.../hadoop/fs/contract/ContractTestUtils.java | 32 +-
.../fs/contract/s3a/ITestS3AContractDistCp.java | 17 +
hadoop-tools/hadoop-azure-datalake/pom.xml | 19 +
.../fs/adl/live/TestAdlContractDistCpLive.java | 36 ++
.../org/apache/hadoop/tools/CopyListing.java | 3 +-
.../hadoop/tools/CopyListingFileStatus.java | 13 +-
.../apache/hadoop/tools/DistCpConstants.java | 20 +-
.../apache/hadoop/tools/DistCpOptionSwitch.java | 20 +-
.../org/apache/hadoop/tools/DistCpOptions.java | 23 +
.../org/apache/hadoop/tools/OptionsParser.java | 6 +
.../hadoop/tools/mapred/CopyCommitter.java | 219 +++++++--
.../hadoop/tools/mapred/DeletedDirTracker.java | 181 ++++++++
.../apache/hadoop/tools/util/DistCpUtils.java | 41 +-
.../contract/AbstractContractDistCpTest.java | 453 +++++++++++++++++--
.../tools/contract/TestLocalContractDistCp.java | 36 ++
.../hadoop/tools/mapred/TestCopyCommitter.java | 165 +++----
.../tools/mapred/TestDeletedDirTracker.java | 250 ++++++++++
.../hadoop/tools/util/TestDistCpUtils.java | 46 +-
.../src/test/resources/contract/localfs.xml | 128 ++++++
.../src/test/resources/log4j.properties | 13 +-
20 files changed, 1510 insertions(+), 211 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
index d2cbca0..54d015a 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
@@ -228,9 +228,9 @@ public class ContractTestUtils extends Assert {
public static void verifyFileContents(FileSystem fs,
Path path,
byte[] original) throws IOException {
+ assertIsFile(fs, path);
FileStatus stat = fs.getFileStatus(path);
String statText = stat.toString();
- assertTrue("not a file " + statText, stat.isFile());
assertEquals("wrong length " + statText, original.length, stat.getLen());
byte[] bytes = readDataset(fs, path, original.length);
compareByteArrays(original, bytes, original.length);
@@ -854,6 +854,36 @@ public class ContractTestUtils extends Assert {
}
/**
+ * Assert that a varargs list of paths exist.
+ * @param fs filesystem
+ * @param message message for exceptions
+ * @param paths paths
+ * @throws IOException IO failure
+ */
+ public static void assertPathsExist(FileSystem fs,
+ String message,
+ Path... paths) throws IOException {
+ for (Path path : paths) {
+ assertPathExists(fs, message, path);
+ }
+ }
+
+ /**
+ * Assert that a varargs list of paths do not exist.
+ * @param fs filesystem
+ * @param message message for exceptions
+ * @param paths paths
+ * @throws IOException IO failure
+ */
+ public static void assertPathsDoNotExist(FileSystem fs,
+ String message,
+ Path... paths) throws IOException {
+ for (Path path : paths) {
+ assertPathDoesNotExist(fs, message, path);
+ }
+ }
+
+ /**
* Create a dataset for use in the tests; all data is in the range
* base to (base+modulo-1) inclusive.
* @param len length of data
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractDistCp.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractDistCp.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractDistCp.java
index 8da8b6a..b3d511e 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractDistCp.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractDistCp.java
@@ -18,11 +18,15 @@
package org.apache.hadoop.fs.contract.s3a;
+import java.io.IOException;
+
import static org.apache.hadoop.fs.s3a.Constants.*;
import static org.apache.hadoop.fs.s3a.S3ATestConstants.SCALE_TEST_TIMEOUT_MILLIS;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.maybeEnableS3Guard;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.FailureInjectionPolicy;
import org.apache.hadoop.tools.contract.AbstractContractDistCpTest;
/**
@@ -57,4 +61,17 @@ public class ITestS3AContractDistCp extends AbstractContractDistCpTest {
protected S3AContract createContract(Configuration conf) {
return new S3AContract(conf);
}
+
+ /**
+ * Always inject the delay path in, so if the destination is inconsistent,
+ * and uses this key, inconsistency triggered.
+ * @param filepath path string in
+ * @return path on the remote FS for distcp
+ * @throws IOException IO failure
+ */
+ @Override
+ protected Path path(final String filepath) throws IOException {
+ Path path = super.path(filepath);
+ return new Path(path, FailureInjectionPolicy.DEFAULT_DELAY_KEY_SUBSTRING);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-azure-datalake/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/pom.xml b/hadoop-tools/hadoop-azure-datalake/pom.xml
index 9b94924..b857abc 100644
--- a/hadoop-tools/hadoop-azure-datalake/pom.xml
+++ b/hadoop-tools/hadoop-azure-datalake/pom.xml
@@ -147,5 +147,24 @@
<version>${okhttp.version}</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-distcp</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-distcp</artifactId>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-azure-datalake/src/test/java/org/apache/hadoop/fs/adl/live/TestAdlContractDistCpLive.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure-datalake/src/test/java/org/apache/hadoop/fs/adl/live/TestAdlContractDistCpLive.java b/hadoop-tools/hadoop-azure-datalake/src/test/java/org/apache/hadoop/fs/adl/live/TestAdlContractDistCpLive.java
new file mode 100644
index 0000000..dae6e80
--- /dev/null
+++ b/hadoop-tools/hadoop-azure-datalake/src/test/java/org/apache/hadoop/fs/adl/live/TestAdlContractDistCpLive.java
@@ -0,0 +1,36 @@
+/*
+ * 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.adl.live;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.contract.AbstractFSContract;
+import org.apache.hadoop.tools.contract.AbstractContractDistCpTest;
+
+/**
+ * Test DistCP operations.
+ */
+public class TestAdlContractDistCpLive extends AbstractContractDistCpTest {
+
+ @Override
+ protected AbstractFSContract createContract(Configuration configuration) {
+ return new AdlStorageContract(configuration);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListing.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListing.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListing.java
index 908b558..9a40a49 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListing.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListing.java
@@ -143,7 +143,6 @@ public abstract class CopyListing extends Configured {
throws DuplicateFileException, IOException {
Configuration config = getConf();
- FileSystem fs = pathToListFile.getFileSystem(config);
final boolean splitLargeFile = context.splitLargeFile();
@@ -153,7 +152,7 @@ public abstract class CopyListing extends Configured {
// <chunkOffset, chunkLength> is continuous.
//
Path checkPath = splitLargeFile?
- pathToListFile : DistCpUtils.sortListing(fs, config, pathToListFile);
+ pathToListFile : DistCpUtils.sortListing(config, pathToListFile);
SequenceFile.Reader reader = new SequenceFile.Reader(
config, SequenceFile.Reader.file(checkPath));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListingFileStatus.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListingFileStatus.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListingFileStatus.java
index 138b491..e01a6b1 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListingFileStatus.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListingFileStatus.java
@@ -27,6 +27,7 @@ import java.util.Map;
import java.util.Map.Entry;
import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.AclEntry;
@@ -46,8 +47,18 @@ import com.google.common.collect.Maps;
/**
* CopyListingFileStatus is a view of {@link FileStatus}, recording additional
* data members useful to distcp.
+ *
+ * This is the datastructure persisted in the sequence files generated
+ * in the CopyCommitter when deleting files.
+ * Any tool working with these generated files needs to be aware of an
+ * important stability guarantee: there is none; expect it to change
+ * across minor Hadoop releases without any support for reading the files of
+ * different versions.
+ * Tools parsing the listings must be built and tested against the point
+ * release of Hadoop which they intend to support.
*/
-@InterfaceAudience.Private
+@InterfaceAudience.LimitedPrivate("Distcp support tools")
+@InterfaceStability.Unstable
public final class CopyListingFileStatus implements Writable {
private static final byte NO_ACL_ENTRIES = -1;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java
index 0bae5d5..212256c 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java
@@ -18,12 +18,19 @@
package org.apache.hadoop.tools;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.Path;
/**
* Utility class to hold commonly used constants.
*/
-public class DistCpConstants {
+@InterfaceAudience.LimitedPrivate("Distcp support tools")
+@InterfaceStability.Evolving
+public final class DistCpConstants {
+
+ private DistCpConstants() {
+ }
/* Default number of threads to use for building file listing */
public static final int DEFAULT_LISTSTATUS_THREADS = 1;
@@ -52,6 +59,8 @@ public class DistCpConstants {
"distcp.preserve.rawxattrs";
public static final String CONF_LABEL_SYNC_FOLDERS = "distcp.sync.folders";
public static final String CONF_LABEL_DELETE_MISSING = "distcp.delete.missing.source";
+ public static final String CONF_LABEL_TRACK_MISSING =
+ "distcp.track.missing.source";
public static final String CONF_LABEL_LISTSTATUS_THREADS = "distcp.liststatus.threads";
public static final String CONF_LABEL_MAX_MAPS = "distcp.max.maps";
public static final String CONF_LABEL_SOURCE_LISTING = "distcp.source.listing";
@@ -148,4 +157,13 @@ public class DistCpConstants {
static final String HDFS_DISTCP_DIFF_DIRECTORY_NAME = ".distcp.diff.tmp";
public static final int COPY_BUFFER_SIZE_DEFAULT = 8 * 1024;
+
+ /** Filename of sorted files in when tracking saves them. */
+ public static final String SOURCE_SORTED_FILE = "source_sorted.seq";
+
+ /** Filename of unsorted target listing. */
+ public static final String TARGET_LISTING_FILE = "target_listing.seq";
+
+ /** Filename of sorted target listing. */
+ public static final String TARGET_SORTED_FILE = "target_sorted.seq";
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptionSwitch.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptionSwitch.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptionSwitch.java
index faef7e6..3ce12b2 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptionSwitch.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptionSwitch.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.tools;
import org.apache.commons.cli.Option;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
/**
@@ -63,10 +64,10 @@ public enum DistCpOptionSwitch {
*/
SYNC_FOLDERS(DistCpConstants.CONF_LABEL_SYNC_FOLDERS,
new Option("update", false, "Update target, copying only missing" +
- "files or directories")),
+ " files or directories")),
/**
- * Deletes missing files in target that are missing from source
+ * Deletes missing files in target that are missing from source.
* This allows the target to be in sync with the source contents
* Typically used in conjunction with SYNC_FOLDERS
* Incompatible with ATOMIC_COMMIT
@@ -74,6 +75,21 @@ public enum DistCpOptionSwitch {
DELETE_MISSING(DistCpConstants.CONF_LABEL_DELETE_MISSING,
new Option("delete", false, "Delete from target, " +
"files missing in source. Delete is applicable only with update or overwrite options")),
+
+ /**
+ * Track missing files in target that are missing from source
+ * This allows for other applications to complete the synchronization,
+ * possibly with object-store-specific delete algorithms.
+ * Typically used in conjunction with SYNC_FOLDERS
+ * Incompatible with ATOMIC_COMMIT
+ */
+ @InterfaceStability.Unstable
+ TRACK_MISSING(DistCpConstants.CONF_LABEL_TRACK_MISSING,
+ new Option("xtrack", true,
+ "Save information about missing source files to the"
+ + " specified directory")),
+
+
/**
* Number of threads for building source file listing (before map-reduce
* phase, max one listStatus per thread at a time).
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptions.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptions.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptions.java
index f33f7fd..ea99016 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptions.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpOptions.java
@@ -24,6 +24,8 @@ import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.tools.util.DistCpUtils;
@@ -43,6 +45,8 @@ import java.util.Set;
*
* This class is immutable.
*/
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
public final class DistCpOptions {
private static final Logger LOG = LoggerFactory.getLogger(Builder.class);
public static final int MAX_NUM_LISTSTATUS_THREADS = 40;
@@ -68,6 +72,9 @@ public final class DistCpOptions {
/** Whether source and target folder contents be sync'ed up. */
private final boolean syncFolder;
+ /** Path to save source/dest sequence files to, if non-null. */
+ private final Path trackPath;
+
/** Whether files only present in target should be deleted. */
private boolean deleteMissing;
@@ -208,6 +215,7 @@ public final class DistCpOptions {
this.copyBufferSize = builder.copyBufferSize;
this.verboseLog = builder.verboseLog;
+ this.trackPath = builder.trackPath;
}
public Path getSourceFileListing() {
@@ -331,6 +339,10 @@ public final class DistCpOptions {
return verboseLog;
}
+ public Path getTrackPath() {
+ return trackPath;
+ }
+
/**
* Add options to configuration. These will be used in the Mapper/committer
*
@@ -371,6 +383,11 @@ public final class DistCpOptions {
String.valueOf(copyBufferSize));
DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.VERBOSE_LOG,
String.valueOf(verboseLog));
+ if (trackPath != null) {
+ DistCpOptionSwitch.addToConf(conf, DistCpOptionSwitch.TRACK_MISSING,
+ String.valueOf(trackPath));
+ }
+
}
/**
@@ -441,6 +458,7 @@ public final class DistCpOptions {
private String filtersFile;
private Path logPath;
+ private Path trackPath;
private String copyStrategy = DistCpConstants.UNIFORMSIZE;
private int numListstatusThreads = 0; // 0 indicates that flag is not set.
@@ -641,6 +659,11 @@ public final class DistCpOptions {
return this;
}
+ public Builder withTrackMissing(Path path) {
+ this.trackPath = path;
+ return this;
+ }
+
public Builder withCopyStrategy(String newCopyStrategy) {
this.copyStrategy = newCopyStrategy;
return this;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/OptionsParser.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/OptionsParser.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/OptionsParser.java
index 606ed32..668b594 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/OptionsParser.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/OptionsParser.java
@@ -145,6 +145,12 @@ public class OptionsParser {
builder.withAtomicWorkPath(new Path(workPath));
}
}
+ if (command.hasOption(DistCpOptionSwitch.TRACK_MISSING.getSwitch())) {
+ builder.withTrackMissing(
+ new Path(getVal(
+ command,
+ DistCpOptionSwitch.TRACK_MISSING.getSwitch())));
+ }
if (command.hasOption(DistCpOptionSwitch.BANDWIDTH.getSwitch())) {
try {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
index 81c2be7..07eacb0 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java
@@ -18,8 +18,9 @@
package org.apache.hadoop.tools.mapred;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -49,6 +50,8 @@ import java.util.EnumSet;
import java.util.LinkedList;
import java.util.List;
+import static org.apache.hadoop.tools.DistCpConstants.*;
+
/**
* The CopyCommitter class is DistCp's OutputCommitter implementation. It is
* responsible for handling the completion/cleanup of the DistCp run.
@@ -62,7 +65,8 @@ import java.util.List;
* 5. Cleanup of any partially copied files, from previous, failed attempts.
*/
public class CopyCommitter extends FileOutputCommitter {
- private static final Log LOG = LogFactory.getLog(CopyCommitter.class);
+ private static final Logger LOG =
+ LoggerFactory.getLogger(CopyCommitter.class);
private final TaskAttemptContext taskAttemptContext;
private boolean syncFolder = false;
@@ -111,6 +115,9 @@ public class CopyCommitter extends FileOutputCommitter {
deleteMissing(conf);
} else if (conf.getBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, false)) {
commitData(conf);
+ } else if (conf.get(CONF_LABEL_TRACK_MISSING) != null) {
+ // save missing information to a directory
+ trackMissing(conf);
}
taskAttemptContext.setStatus("Commit Successful");
}
@@ -334,40 +341,64 @@ public class CopyCommitter extends FileOutputCommitter {
LOG.info("Preserved status on " + preservedEntries + " dir entries on target");
}
- // This method deletes "extra" files from the target, if they're not
- // available at the source.
+ /**
+ * Track all the missing files by saving the listings to the tracking
+ * directory.
+ * This is the same as listing phase of the
+ * {@link #deleteMissing(Configuration)} operation.
+ * @param conf configuration to read options from, and for FS instantiation.
+ * @throws IOException IO failure
+ */
+ private void trackMissing(Configuration conf) throws IOException {
+ // destination directory for all output files
+ Path trackDir = new Path(
+ conf.get(DistCpConstants.CONF_LABEL_TRACK_MISSING));
+
+ // where is the existing source listing?
+ Path sourceListing = new Path(
+ conf.get(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH));
+ LOG.info("Tracking file changes to directory {}", trackDir);
+
+ // the destination path is under the track directory
+ Path sourceSortedListing = new Path(trackDir,
+ DistCpConstants.SOURCE_SORTED_FILE);
+ LOG.info("Source listing {}", sourceSortedListing);
+
+ DistCpUtils.sortListing(conf, sourceListing, sourceSortedListing);
+
+ // Similarly, create the listing of target-files. Sort alphabetically.
+ // target listing will be deleted after the sort
+ Path targetListing = new Path(trackDir, TARGET_LISTING_FILE);
+ Path sortedTargetListing = new Path(trackDir, TARGET_SORTED_FILE);
+ // list the target
+ listTargetFiles(conf, targetListing, sortedTargetListing);
+ LOG.info("Target listing {}", sortedTargetListing);
+
+ targetListing.getFileSystem(conf).delete(targetListing, false);
+ }
+
+ /**
+ * Deletes "extra" files and directories from the target, if they're not
+ * available at the source.
+ * @param conf configuration to read options from, and for FS instantiation.
+ * @throws IOException IO failure
+ */
private void deleteMissing(Configuration conf) throws IOException {
LOG.info("-delete option is enabled. About to remove entries from " +
"target that are missing in source");
+ long listingStart = System.currentTimeMillis();
// Sort the source-file listing alphabetically.
Path sourceListing = new Path(conf.get(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH));
FileSystem clusterFS = sourceListing.getFileSystem(conf);
- Path sortedSourceListing = DistCpUtils.sortListing(clusterFS, conf, sourceListing);
+ Path sortedSourceListing = DistCpUtils.sortListing(conf, sourceListing);
// Similarly, create the listing of target-files. Sort alphabetically.
Path targetListing = new Path(sourceListing.getParent(), "targetListing.seq");
- CopyListing target = new GlobbedCopyListing(new Configuration(conf), null);
+ Path sortedTargetListing = new Path(targetListing.toString() + "_sorted");
- List<Path> targets = new ArrayList<Path>(1);
- Path targetFinalPath = new Path(conf.get(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH));
- targets.add(targetFinalPath);
- Path resultNonePath = Path.getPathWithoutSchemeAndAuthority(targetFinalPath)
- .toString().startsWith(DistCpConstants.HDFS_RESERVED_RAW_DIRECTORY_NAME)
- ? DistCpConstants.RAW_NONE_PATH : DistCpConstants.NONE_PATH;
- //
- // Set up options to be the same from the CopyListing.buildListing's perspective,
- // so to collect similar listings as when doing the copy
- //
- DistCpOptions options = new DistCpOptions.Builder(targets, resultNonePath)
- .withOverwrite(overwrite)
- .withSyncFolder(syncFolder)
- .build();
- DistCpContext distCpContext = new DistCpContext(options);
- distCpContext.setTargetPathExists(targetPathExists);
-
- target.buildListing(targetListing, distCpContext);
- Path sortedTargetListing = DistCpUtils.sortListing(clusterFS, conf, targetListing);
+ Path targetFinalPath = listTargetFiles(conf,
+ targetListing, sortedTargetListing);
long totalLen = clusterFS.getFileStatus(sortedTargetListing).getLen();
SequenceFile.Reader sourceReader = new SequenceFile.Reader(conf,
@@ -377,41 +408,153 @@ public class CopyCommitter extends FileOutputCommitter {
// Walk both source and target file listings.
// Delete all from target that doesn't also exist on source.
+ long deletionStart = System.currentTimeMillis();
+ LOG.info("Listing completed in {}",
+ formatDuration(deletionStart - listingStart));
+
long deletedEntries = 0;
+ long filesDeleted = 0;
+ long missingDeletes = 0;
+ long failedDeletes = 0;
+ long skippedDeletes = 0;
+ long deletedDirectories = 0;
+ // this is an arbitrary constant.
+ final DeletedDirTracker tracker = new DeletedDirTracker(1000);
try {
CopyListingFileStatus srcFileStatus = new CopyListingFileStatus();
Text srcRelPath = new Text();
CopyListingFileStatus trgtFileStatus = new CopyListingFileStatus();
Text trgtRelPath = new Text();
- FileSystem targetFS = targetFinalPath.getFileSystem(conf);
+ final FileSystem targetFS = targetFinalPath.getFileSystem(conf);
+ boolean showProgress;
boolean srcAvailable = sourceReader.next(srcRelPath, srcFileStatus);
while (targetReader.next(trgtRelPath, trgtFileStatus)) {
// Skip sources that don't exist on target.
while (srcAvailable && trgtRelPath.compareTo(srcRelPath) > 0) {
srcAvailable = sourceReader.next(srcRelPath, srcFileStatus);
}
+ Path targetEntry = trgtFileStatus.getPath();
+ LOG.debug("Comparing {} and {}",
+ srcFileStatus.getPath(), targetEntry);
if (srcAvailable && trgtRelPath.equals(srcRelPath)) continue;
- // Target doesn't exist at source. Delete.
- boolean result = targetFS.delete(trgtFileStatus.getPath(), true)
- || !targetFS.exists(trgtFileStatus.getPath());
- if (result) {
- LOG.info("Deleted " + trgtFileStatus.getPath() + " - Missing at source");
- deletedEntries++;
+ // Target doesn't exist at source. Try to delete it.
+ if (tracker.shouldDelete(trgtFileStatus)) {
+ showProgress = true;
+ try {
+ if (targetFS.delete(targetEntry, true)) {
+ // the delete worked. Unless the file is actually missing, this is the
+ LOG.info("Deleted " + targetEntry + " - missing at source");
+ deletedEntries++;
+ if (trgtFileStatus.isDirectory()) {
+ deletedDirectories++;
+ } else {
+ filesDeleted++;
+ }
+ } else {
+ // delete returned false.
+ // For all the filestores which implement the FS spec properly,
+ // this means "the file wasn't there".
+ // so track but don't worry about it.
+ LOG.info("delete({}) returned false ({})",
+ targetEntry, trgtFileStatus);
+ missingDeletes++;
+ }
+ } catch (IOException e) {
+ if (!ignoreFailures) {
+ throw e;
+ } else {
+ // failed to delete, but ignoring errors. So continue
+ LOG.info("Failed to delete {}, ignoring exception {}",
+ targetEntry, e.toString());
+ LOG.debug("Failed to delete {}", targetEntry, e);
+ // count and break out the loop
+ failedDeletes++;
+ }
+ }
} else {
- throw new IOException("Unable to delete " + trgtFileStatus.getPath());
+ LOG.debug("Skipping deletion of {}", targetEntry);
+ skippedDeletes++;
+ showProgress = false;
+ }
+ if (showProgress) {
+ // update progress if there's been any FS IO/files deleted.
+ taskAttemptContext.progress();
+ taskAttemptContext.setStatus("Deleting removed files from target. [" +
+ targetReader.getPosition() * 100 / totalLen + "%]");
}
- taskAttemptContext.progress();
- taskAttemptContext.setStatus("Deleting missing files from target. [" +
- targetReader.getPosition() * 100 / totalLen + "%]");
}
+ // if the FS toString() call prints statistics, they get logged here
+ LOG.info("Completed deletion of files from {}", targetFS);
} finally {
IOUtils.closeStream(sourceReader);
IOUtils.closeStream(targetReader);
}
- LOG.info("Deleted " + deletedEntries + " from target: " + targets.get(0));
+ long deletionEnd = System.currentTimeMillis();
+ long deletedFileCount = deletedEntries - deletedDirectories;
+ LOG.info("Deleted from target: files: {} directories: {};"
+ + " skipped deletions {}; deletions already missing {};"
+ + " failed deletes {}",
+ deletedFileCount, deletedDirectories, skippedDeletes,
+ missingDeletes, failedDeletes);
+ LOG.info("Number of tracked deleted directories {}", tracker.size());
+ LOG.info("Duration of deletions: {}",
+ formatDuration(deletionEnd - deletionStart));
+ LOG.info("Total duration of deletion operation: {}",
+ formatDuration(deletionEnd - listingStart));
+ }
+
+ /**
+ * Take a duration and return a human-readable duration of
+ * hours:minutes:seconds.millis.
+ * @param duration to process
+ * @return a string for logging.
+ */
+ private String formatDuration(long duration) {
+
+ long seconds = duration > 0 ? (duration / 1000) : 0;
+ long minutes = (seconds / 60);
+ long hours = (minutes / 60);
+ return String.format("%d:%02d:%02d.%03d",
+ hours, minutes % 60, seconds % 60, duration % 1000);
+ }
+
+ /**
+ * Build a listing of the target files, sorted and unsorted.
+ * @param conf configuration to work with
+ * @param targetListing target listing
+ * @param sortedTargetListing sorted version of the listing
+ * @return the target path of the operation
+ * @throws IOException IO failure.
+ */
+ private Path listTargetFiles(final Configuration conf,
+ final Path targetListing,
+ final Path sortedTargetListing) throws IOException {
+ CopyListing target = new GlobbedCopyListing(new Configuration(conf), null);
+ Path targetFinalPath = new Path(
+ conf.get(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH));
+ List<Path> targets = new ArrayList<>(1);
+ targets.add(targetFinalPath);
+ Path resultNonePath = Path.getPathWithoutSchemeAndAuthority(targetFinalPath)
+ .toString().startsWith(DistCpConstants.HDFS_RESERVED_RAW_DIRECTORY_NAME)
+ ? DistCpConstants.RAW_NONE_PATH
+ : DistCpConstants.NONE_PATH;
+ //
+ // Set up options to be the same from the CopyListing.buildListing's
+ // perspective, so to collect similar listings as when doing the copy
+ //
+ DistCpOptions options = new DistCpOptions.Builder(targets, resultNonePath)
+ .withOverwrite(overwrite)
+ .withSyncFolder(syncFolder)
+ .build();
+ DistCpContext distCpContext = new DistCpContext(options);
+ distCpContext.setTargetPathExists(targetPathExists);
+
+ target.buildListing(targetListing, distCpContext);
+ DistCpUtils.sortListing(conf, targetListing, sortedTargetListing);
+ return targetFinalPath;
}
private void commitData(Configuration conf) throws IOException {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/DeletedDirTracker.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/DeletedDirTracker.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/DeletedDirTracker.java
new file mode 100644
index 0000000..64431f7
--- /dev/null
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/DeletedDirTracker.java
@@ -0,0 +1,181 @@
+/*
+ * 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.tools.mapred;
+
+import com.google.common.base.Preconditions;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.tools.CopyListingFileStatus;
+
+/**
+ * Track deleted directories and support queries to
+ * check for add them.
+ *
+ * Assumptions.
+ * <ol>
+ * <liA sorted list of deletions are processed, where directories come
+ * before their children/descendants.</li>
+ * <li>Deep directory trees are being deleted.</li>
+ * <li>The total number of directories deleted is very much
+ * less than the number of files.</li>
+ * <li>Most deleted files are in directories which have
+ * been deleted.</li>
+ * <li>The cost of issuing a delete() call is less than that that
+ * of creating Path entries for parent directories and looking them
+ * up in a hash table.</li>
+ * <li>That a modest cache is sufficient to identify whether or not
+ * a parent directory has been deleted./li>
+ * <li>And that if a path has been evicted from a path, the cost of
+ * the extra deletions incurred is not significant.</li>
+ * </ol>
+ *
+ * The directory structure this algorithm is intended to optimize for is
+ * the deletion of datasets partitioned/bucketed into a directory tree,
+ * and deleted in bulk.
+ *
+ * The ordering of deletions comes from the merge sort of the copy listings;
+ * we rely on this placing a path "/dir1" ahead of "/dir1/file1",
+ * "/dir1/dir2/file2", and other descendants.
+ * We do not rely on parent entries being added immediately before children,
+ * as sorting may place "/dir12" between "/dir1" and its descendants.
+ *
+ * Algorithm
+ *
+ * <ol>
+ * <li>
+ * Before deleting a directory or file, a check is made to see if an
+ * ancestor is in the cache of deleted directories.
+ * </li>
+ * <li>
+ * If an ancestor is found is: skip the delete.
+ * </li>
+ * <li>
+ * If an ancestor is not foundI: delete the file/dir.
+ * </li>
+ * <li>
+ * When the entry probed is a directory, it is always added to the cache of
+ * directories, irrespective of the search for an ancestor.
+ * This is to speed up scans of files directly underneath the path.
+ * </li>
+ * </ol>
+ *
+ *
+ */
+final class DeletedDirTracker {
+
+ /**
+ * An LRU cache of directories.
+ */
+ private final Cache<Path, Path> directories;
+
+ /**
+ * Maximum size of the cache.
+ */
+ private final int cacheSize;
+
+ /**
+ * Create an instance.
+ * @param cacheSize maximum cache size.
+ */
+ DeletedDirTracker(int cacheSize) {
+ this.cacheSize = cacheSize;
+ directories = CacheBuilder.newBuilder()
+ .maximumSize(this.cacheSize)
+ .build();
+ }
+
+ /**
+ * Recursive scan for a directory being in the cache of deleted paths.
+ * @param dir directory to look for.
+ * @return true iff the path or a parent is in the cache.
+ */
+ boolean isDirectoryOrAncestorDeleted(Path dir) {
+ if (dir == null) {
+ // at root
+ return false;
+ } else if (isContained(dir)) {
+ // cache hit
+ return true;
+ } else {
+ // cache miss, check parent
+ return isDirectoryOrAncestorDeleted(dir.getParent());
+ }
+ }
+
+ /**
+ * Probe for a path being deleted by virtue of the fact that an
+ * ancestor dir has already been deleted.
+ * @param path path to check
+ * @return true if the parent dir is deleted.
+ */
+ private boolean isInDeletedDirectory(Path path) {
+ Preconditions.checkArgument(!path.isRoot(), "Root Dir");
+ return isDirectoryOrAncestorDeleted(path.getParent());
+ }
+
+ /**
+ * Should a file or directory be deleted?
+ * The cache of deleted directories will be updated with the path
+ * of the status if it references a directory.
+ * @param status file/path to check
+ * @return true if the path should be deleted.
+ */
+ boolean shouldDelete(CopyListingFileStatus status) {
+ Path path = status.getPath();
+ Preconditions.checkArgument(!path.isRoot(), "Root Dir");
+ if (status.isDirectory()) {
+ boolean deleted = isDirectoryOrAncestorDeleted(path);
+ // even if an ancestor has been deleted, add this entry as
+ // a deleted directory.
+ directories.put(path, path);
+ return !deleted;
+ } else {
+ return !isInDeletedDirectory(path);
+ }
+ }
+
+ /**
+ * Is a path directly contained in the set of deleted directories.
+ * @param dir directory to probe
+ * @return true if this directory is recorded as being deleted.
+ */
+ boolean isContained(Path dir) {
+ return directories.getIfPresent(dir) != null;
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder(
+ "DeletedDirTracker{");
+ sb.append("maximum size=").append(cacheSize);
+ sb.append("; current size=").append(directories.size());
+ sb.append('}');
+ return sb.toString();
+ }
+
+ /**
+ * Return the current size of the tracker, as in #of entries in the cache.
+ * @return tracker size.
+ */
+ long size() {
+ return directories.size();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/DistCpUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/DistCpUtils.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/DistCpUtils.java
index 2b3b529..b6302c0 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/DistCpUtils.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/DistCpUtils.java
@@ -433,24 +433,45 @@ public class DistCpUtils {
}
/**
- * Sort sequence file containing FileStatus and Text as key and value respecitvely
+ * Sort sequence file containing FileStatus and Text as key and value
+ * respectively.
*
- * @param fs - File System
* @param conf - Configuration
* @param sourceListing - Source listing file
* @return Path of the sorted file. Is source file with _sorted appended to the name
* @throws IOException - Any exception during sort.
*/
- public static Path sortListing(FileSystem fs, Configuration conf, Path sourceListing)
+ public static Path sortListing(Configuration conf,
+ Path sourceListing)
throws IOException {
+ Path output = new Path(sourceListing.toString() + "_sorted");
+ sortListing(conf, sourceListing, output);
+ return output;
+ }
+
+ /**
+ * Sort sequence file containing FileStatus and Text as key and value
+ * respectively, saving the result to the {@code output} path, which
+ * will be deleted first.
+ *
+ * @param conf - Configuration
+ * @param sourceListing - Source listing file
+ * @param output output path
+ * @throws IOException - Any exception during sort.
+ */
+
+ public static void sortListing(final Configuration conf,
+ final Path sourceListing,
+ final Path output) throws IOException {
+ FileSystem fs = sourceListing.getFileSystem(conf);
+ // force verify that the destination FS matches the input
+ fs.makeQualified(output);
SequenceFile.Sorter sorter = new SequenceFile.Sorter(fs, Text.class,
CopyListingFileStatus.class, conf);
- Path output = new Path(sourceListing.toString() + "_sorted");
fs.delete(output, false);
sorter.sort(sourceListing, output);
- return output;
}
/**
@@ -547,9 +568,13 @@ public class DistCpUtils {
throws IOException {
FileChecksum targetChecksum = null;
try {
- sourceChecksum = sourceChecksum != null ? sourceChecksum : sourceFS
- .getFileChecksum(source);
- targetChecksum = targetFS.getFileChecksum(target);
+ sourceChecksum = sourceChecksum != null
+ ? sourceChecksum
+ : sourceFS.getFileChecksum(source);
+ if (sourceChecksum != null) {
+ // iff there's a source checksum, look for one at the destination.
+ targetChecksum = targetFS.getFileChecksum(target);
+ }
} catch (IOException e) {
LOG.error("Unable to retrieve checksum for " + source + " or " + target, e);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java
index fb1a64d..a5e0a03 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/AbstractContractDistCpTest.java
@@ -20,22 +20,35 @@ package org.apache.hadoop.tools.contract;
import static org.apache.hadoop.fs.contract.ContractTestUtils.*;
+import java.io.IOException;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.contract.AbstractFSContractTestBase;
import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.tools.CopyListingFileStatus;
import org.apache.hadoop.tools.DistCp;
+import org.apache.hadoop.tools.DistCpConstants;
import org.apache.hadoop.tools.DistCpOptions;
+import org.apache.hadoop.tools.mapred.CopyMapper;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Contract test suite covering a file system's integration with DistCp. The
@@ -48,13 +61,70 @@ import org.junit.rules.TestName;
public abstract class AbstractContractDistCpTest
extends AbstractFSContractTestBase {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AbstractContractDistCpTest.class);
+
+ public static final String SCALE_TEST_DISTCP_FILE_SIZE_KB
+ = "scale.test.distcp.file.size.kb";
+
+ public static final int DEFAULT_DISTCP_SIZE_KB = 1024;
+
+ protected static final int MB = 1024 * 1024;
+
@Rule
public TestName testName = new TestName();
+ /**
+ * The timeout value is extended over the default so that large updates
+ * are allowed to take time, especially to remote stores.
+ * @return the current test timeout
+ */
+ protected int getTestTimeoutMillis() {
+ return 15 * 60 * 1000;
+ }
+
private Configuration conf;
private FileSystem localFS, remoteFS;
private Path localDir, remoteDir;
+ private Path inputDir;
+
+ private Path inputSubDir1;
+
+ private Path inputSubDir2;
+
+ private Path inputSubDir4;
+
+ private Path inputFile1;
+
+ private Path inputFile2;
+
+ private Path inputFile3;
+
+ private Path inputFile4;
+
+ private Path inputFile5;
+
+ private Path outputDir;
+
+ private Path outputSubDir1;
+
+ private Path outputSubDir2;
+
+ private Path outputSubDir4;
+
+ private Path outputFile1;
+
+ private Path outputFile2;
+
+ private Path outputFile3;
+
+ private Path outputFile4;
+
+ private Path outputFile5;
+
+ private Path inputDirUnderOutputDir;
+
@Override
protected Configuration createConfiguration() {
Configuration newConf = new Configuration();
@@ -73,20 +143,307 @@ public abstract class AbstractContractDistCpTest
// All paths are fully qualified including scheme (not taking advantage of
// default file system), so if something fails, the messages will make it
// clear which paths are local and which paths are remote.
- Path testSubDir = new Path(getClass().getSimpleName(),
- testName.getMethodName());
- localDir = localFS.makeQualified(new Path(new Path(
- GenericTestUtils.getTestDir().toURI()), testSubDir));
+ String className = getClass().getSimpleName();
+ String testSubDir = className + "/" + testName.getMethodName();
+ localDir =
+ localFS.makeQualified(new Path(new Path(
+ GenericTestUtils.getTestDir().toURI()), testSubDir + "/local"));
mkdirs(localFS, localDir);
- remoteDir = remoteFS.makeQualified(
- new Path(getContract().getTestPath(), testSubDir));
+ remoteDir = path(testSubDir + "/remote");
mkdirs(remoteFS, remoteDir);
+ // test teardown does this, but IDE-based test debugging can skip
+ // that teardown; this guarantees the initial state is clean
+ remoteFS.delete(remoteDir, true);
+ localFS.delete(localDir, true);
+ }
+
+ /**
+ * Set up both input and output fields.
+ * @param src source tree
+ * @param dest dest tree
+ */
+ protected void initPathFields(final Path src, final Path dest) {
+ initInputFields(src);
+ initOutputFields(dest);
+ }
+
+ /**
+ * Output field setup.
+ * @param path path to set up
+ */
+ protected void initOutputFields(final Path path) {
+ outputDir = new Path(path, "outputDir");
+ inputDirUnderOutputDir = new Path(outputDir, "inputDir");
+ outputFile1 = new Path(inputDirUnderOutputDir, "file1");
+ outputSubDir1 = new Path(inputDirUnderOutputDir, "subDir1");
+ outputFile2 = new Path(outputSubDir1, "file2");
+ outputSubDir2 = new Path(inputDirUnderOutputDir, "subDir2/subDir2");
+ outputFile3 = new Path(outputSubDir2, "file3");
+ outputSubDir4 = new Path(inputDirUnderOutputDir, "subDir4/subDir4");
+ outputFile4 = new Path(outputSubDir4, "file4");
+ outputFile5 = new Path(outputSubDir4, "file5");
+ }
+
+ /**
+ * this path setup is used across different methods (copy, update, track)
+ * so they are set up as fields.
+ * @param srcDir source directory for these to go under.
+ */
+ protected void initInputFields(final Path srcDir) {
+ inputDir = new Path(srcDir, "inputDir");
+ inputFile1 = new Path(inputDir, "file1");
+ inputSubDir1 = new Path(inputDir, "subDir1");
+ inputFile2 = new Path(inputSubDir1, "file2");
+ inputSubDir2 = new Path(inputDir, "subDir2/subDir2");
+ inputFile3 = new Path(inputSubDir2, "file3");
+ inputSubDir4 = new Path(inputDir, "subDir4/subDir4");
+ inputFile4 = new Path(inputSubDir4, "file4");
+ inputFile5 = new Path(inputSubDir4, "file5");
+ }
+
+ protected FileSystem getLocalFS() {
+ return localFS;
+ }
+
+ protected FileSystem getRemoteFS() {
+ return remoteFS;
+ }
+
+ protected Path getLocalDir() {
+ return localDir;
+ }
+
+ protected Path getRemoteDir() {
+ return remoteDir;
+ }
+
+ @Test
+ public void testUpdateDeepDirectoryStructureToRemote() throws Exception {
+ describe("update a deep directory structure from local to remote");
+ distCpDeepDirectoryStructure(localFS, localDir, remoteFS, remoteDir);
+ distCpUpdateDeepDirectoryStructure(inputDirUnderOutputDir);
+ }
+
+ @Test
+ public void testUpdateDeepDirectoryStructureNoChange() throws Exception {
+ describe("update an unchanged directory structure"
+ + " from local to remote; expect no copy");
+ Path target = distCpDeepDirectoryStructure(localFS, localDir, remoteFS,
+ remoteDir);
+ describe("\nExecuting Update\n");
+ Job job = distCpUpdate(localDir, target);
+ assertCounterInRange(job, CopyMapper.Counter.SKIP, 1, -1);
+ assertCounterInRange(job, CopyMapper.Counter.BYTESCOPIED, 0, 0);
+ }
+
+ /**
+ * Assert that a counter is in a range; min and max values are inclusive.
+ * @param job job to query
+ * @param counter counter to examine
+ * @param min min value, if negative "no minimum"
+ * @param max max value, if negative "no maximum"
+ * @throws IOException IO problem
+ */
+ void assertCounterInRange(Job job, Enum<?> counter, long min, long max)
+ throws IOException {
+ Counter c = job.getCounters().findCounter(counter);
+ long value = c.getValue();
+ String description =
+ String.format("%s value %s", c.getDisplayName(), value);
+
+ if (min >= 0) {
+ assertTrue(description + " too below minimum " + min,
+ value >= min);
+ }
+ if (max >= 0) {
+ assertTrue(description + " above maximum " + max,
+ value <= max);
+ }
}
+ /**
+ * Do a distcp from the local source to the destination filesystem.
+ * This is executed as part of
+ * {@link #testUpdateDeepDirectoryStructureToRemote()}; it's designed to be
+ * overidden or wrapped by subclasses which wish to add more assertions.
+ *
+ * Life is complicated here by the way that the src/dest paths
+ * on a distcp is different with -update.
+ * @param destDir output directory used by the initial distcp
+ * @return the distcp job
+ */
+ protected Job distCpUpdateDeepDirectoryStructure(final Path destDir)
+ throws Exception {
+ describe("Now do an incremental update with deletion of missing files");
+ Path srcDir = inputDir;
+ LOG.info("Source directory = {}, dest={}", srcDir, destDir);
+
+ ContractTestUtils.assertPathsExist(localFS,
+ "Paths for test are wrong",
+ inputFile1, inputFile2, inputFile3, inputFile4, inputFile5);
+
+ modifySourceDirectories();
+
+ Job job = distCpUpdate(srcDir, destDir);
+
+ Path outputFileNew1 = new Path(outputSubDir2, "newfile1");
+
+ lsR("Updated Remote", remoteFS, destDir);
+
+ ContractTestUtils.assertPathDoesNotExist(remoteFS,
+ " deleted from " + inputFile1, outputFile1);
+ ContractTestUtils.assertIsFile(remoteFS, outputFileNew1);
+ ContractTestUtils.assertPathsDoNotExist(remoteFS,
+ "DistCP should have deleted",
+ outputFile3, outputFile4, outputSubDir4);
+ assertCounterInRange(job, CopyMapper.Counter.COPY, 1, 1);
+ assertCounterInRange(job, CopyMapper.Counter.SKIP, 1, -1);
+ return job;
+ }
+
+ /**
+ * Run distcp -update srcDir destDir.
+ * @param srcDir local source directory
+ * @param destDir remote destination directory.
+ * @return the completed job
+ * @throws Exception any failure.
+ */
+ private Job distCpUpdate(final Path srcDir, final Path destDir)
+ throws Exception {
+ describe("\nDistcp -update from " + srcDir + " to " + destDir);
+ lsR("Local to update", localFS, srcDir);
+ lsR("Remote before update", remoteFS, destDir);
+ return runDistCp(buildWithStandardOptions(
+ new DistCpOptions.Builder(
+ Collections.singletonList(srcDir), destDir)
+ .withDeleteMissing(true)
+ .withSyncFolder(true)
+ .withCRC(true)
+ .withOverwrite(false)));
+ }
+
+ /**
+ * Update the source directories as various tests expect,
+ * including adding a new file.
+ * @return the path to the newly created file
+ * @throws IOException IO failure
+ */
+ private Path modifySourceDirectories() throws IOException {
+ localFS.delete(inputFile1, false);
+ localFS.delete(inputFile3, false);
+ // delete all of subdir4, so input/output file 4 & 5 will go
+ localFS.delete(inputSubDir4, true);
+ // add one new file
+ Path inputFileNew1 = new Path(inputSubDir2, "newfile1");
+ ContractTestUtils.touch(localFS, inputFileNew1);
+ return inputFileNew1;
+ }
+
+
@Test
- public void deepDirectoryStructureToRemote() throws Exception {
+ public void testTrackDeepDirectoryStructureToRemote() throws Exception {
describe("copy a deep directory structure from local to remote");
- deepDirectoryStructure(localFS, localDir, remoteFS, remoteDir);
+
+ Path destDir = distCpDeepDirectoryStructure(localFS, localDir, remoteFS,
+ remoteDir);
+ ContractTestUtils.assertIsDirectory(remoteFS, destDir);
+
+ describe("Now do an incremental update and save of missing files");
+ Path srcDir = inputDir;
+ // same path setup as in deepDirectoryStructure()
+ Path trackDir = new Path(localDir, "trackDir");
+
+
+ describe("\nDirectories\n");
+ lsR("Local to update", localFS, srcDir);
+ lsR("Remote before update", remoteFS, destDir);
+
+
+ ContractTestUtils.assertPathsExist(localFS,
+ "Paths for test are wrong",
+ inputFile2, inputFile3, inputFile4, inputFile5);
+
+ Path inputFileNew1 = modifySourceDirectories();
+
+ // Distcp set to track but not delete
+ runDistCp(buildWithStandardOptions(
+ new DistCpOptions.Builder(
+ Collections.singletonList(srcDir),
+ inputDirUnderOutputDir)
+ .withTrackMissing(trackDir)
+ .withSyncFolder(true)
+ .withOverwrite(false)));
+
+ lsR("tracked udpate", remoteFS, destDir);
+ // new file went over
+ Path outputFileNew1 = new Path(outputSubDir2, "newfile1");
+ ContractTestUtils.assertIsFile(remoteFS, outputFileNew1);
+
+ ContractTestUtils.assertPathExists(localFS, "tracking directory",
+ trackDir);
+
+ // now read in the listings
+ Path sortedSourceListing = new Path(trackDir,
+ DistCpConstants.SOURCE_SORTED_FILE);
+ ContractTestUtils.assertIsFile(localFS, sortedSourceListing);
+ Path sortedTargetListing = new Path(trackDir,
+ DistCpConstants.TARGET_SORTED_FILE);
+ ContractTestUtils.assertIsFile(localFS, sortedTargetListing);
+ // deletion didn't happen
+ ContractTestUtils.assertPathsExist(remoteFS,
+ "DistCP should have retained",
+ outputFile2, outputFile3, outputFile4, outputSubDir4);
+
+ // now scan the table and see that things are there.
+ Map<String, Path> sourceFiles = new HashMap<>(10);
+ Map<String, Path> targetFiles = new HashMap<>(10);
+
+ try (SequenceFile.Reader sourceReader = new SequenceFile.Reader(conf,
+ SequenceFile.Reader.file(sortedSourceListing));
+ SequenceFile.Reader targetReader = new SequenceFile.Reader(conf,
+ SequenceFile.Reader.file(sortedTargetListing))) {
+ CopyListingFileStatus copyStatus = new CopyListingFileStatus();
+ Text name = new Text();
+ while(sourceReader.next(name, copyStatus)) {
+ String key = name.toString();
+ Path path = copyStatus.getPath();
+ LOG.info("{}: {}", key, path);
+ sourceFiles.put(key, path);
+ }
+ while(targetReader.next(name, copyStatus)) {
+ String key = name.toString();
+ Path path = copyStatus.getPath();
+ LOG.info("{}: {}", key, path);
+ targetFiles.put(name.toString(), copyStatus.getPath());
+ }
+ }
+
+ // look for the new file in both lists
+ assertTrue("No " + outputFileNew1 + " in source listing",
+ sourceFiles.containsValue(inputFileNew1));
+ assertTrue("No " + outputFileNew1 + " in target listing",
+ targetFiles.containsValue(outputFileNew1));
+ assertTrue("No " + outputSubDir4 + " in target listing",
+ targetFiles.containsValue(outputSubDir4));
+ assertFalse("Found " + inputSubDir4 + " in source listing",
+ sourceFiles.containsValue(inputSubDir4));
+
+ }
+
+ public void lsR(final String description,
+ final FileSystem fs,
+ final Path dir) throws IOException {
+ RemoteIterator<LocatedFileStatus> files = fs.listFiles(dir, true);
+ LOG.info("{}: {}:", description, dir);
+ StringBuilder sb = new StringBuilder();
+ while(files.hasNext()) {
+ LocatedFileStatus status = files.next();
+ sb.append(String.format(" %s; type=%s; length=%d",
+ status.getPath(),
+ status.isDirectory()? "dir" : "file",
+ status.getLen()));
+ }
+ LOG.info("{}", sb);
}
@Test
@@ -96,34 +453,35 @@ public abstract class AbstractContractDistCpTest
}
@Test
- public void deepDirectoryStructureFromRemote() throws Exception {
+ public void testDeepDirectoryStructureFromRemote() throws Exception {
describe("copy a deep directory structure from remote to local");
- deepDirectoryStructure(remoteFS, remoteDir, localFS, localDir);
+ distCpDeepDirectoryStructure(remoteFS, remoteDir, localFS, localDir);
}
@Test
- public void largeFilesFromRemote() throws Exception {
+ public void testLargeFilesFromRemote() throws Exception {
describe("copy multiple large files from remote to local");
largeFiles(remoteFS, remoteDir, localFS, localDir);
}
/**
- * Executes a test using a file system sub-tree with multiple nesting levels.
+ * Executes a DistCp using a file system sub-tree with multiple nesting
+ * levels.
+ * The filenames are those of the fields initialized in setup.
*
* @param srcFS source FileSystem
* @param srcDir source directory
* @param dstFS destination FileSystem
* @param dstDir destination directory
+ * @return the target directory of the copy
* @throws Exception if there is a failure
*/
- private void deepDirectoryStructure(FileSystem srcFS, Path srcDir,
- FileSystem dstFS, Path dstDir) throws Exception {
- Path inputDir = new Path(srcDir, "inputDir");
- Path inputSubDir1 = new Path(inputDir, "subDir1");
- Path inputSubDir2 = new Path(inputDir, "subDir2/subDir3");
- Path inputFile1 = new Path(inputDir, "file1");
- Path inputFile2 = new Path(inputSubDir1, "file2");
- Path inputFile3 = new Path(inputSubDir2, "file3");
+ private Path distCpDeepDirectoryStructure(FileSystem srcFS,
+ Path srcDir,
+ FileSystem dstFS,
+ Path dstDir) throws Exception {
+ initPathFields(srcDir, dstDir);
+
mkdirs(srcFS, inputSubDir1);
mkdirs(srcFS, inputSubDir2);
byte[] data1 = dataset(100, 33, 43);
@@ -132,14 +490,18 @@ public abstract class AbstractContractDistCpTest
createFile(srcFS, inputFile2, true, data2);
byte[] data3 = dataset(300, 53, 63);
createFile(srcFS, inputFile3, true, data3);
+ createFile(srcFS, inputFile4, true, dataset(400, 53, 63));
+ createFile(srcFS, inputFile5, true, dataset(500, 53, 63));
Path target = new Path(dstDir, "outputDir");
runDistCp(inputDir, target);
ContractTestUtils.assertIsDirectory(dstFS, target);
+ lsR("Destination tree after distcp", dstFS, target);
verifyFileContents(dstFS, new Path(target, "inputDir/file1"), data1);
verifyFileContents(dstFS,
new Path(target, "inputDir/subDir1/file2"), data2);
verifyFileContents(dstFS,
- new Path(target, "inputDir/subDir2/subDir3/file3"), data3);
+ new Path(target, "inputDir/subDir2/subDir2/file3"), data3);
+ return target;
}
/**
@@ -153,20 +515,21 @@ public abstract class AbstractContractDistCpTest
*/
private void largeFiles(FileSystem srcFS, Path srcDir, FileSystem dstFS,
Path dstDir) throws Exception {
- Path inputDir = new Path(srcDir, "inputDir");
- Path inputFile1 = new Path(inputDir, "file1");
- Path inputFile2 = new Path(inputDir, "file2");
- Path inputFile3 = new Path(inputDir, "file3");
+ initPathFields(srcDir, dstDir);
+ Path largeFile1 = new Path(inputDir, "file1");
+ Path largeFile2 = new Path(inputDir, "file2");
+ Path largeFile3 = new Path(inputDir, "file3");
mkdirs(srcFS, inputDir);
- int fileSizeKb = conf.getInt("scale.test.distcp.file.size.kb", 10 * 1024);
+ int fileSizeKb = conf.getInt(SCALE_TEST_DISTCP_FILE_SIZE_KB,
+ DEFAULT_DISTCP_SIZE_KB);
int fileSizeMb = fileSizeKb / 1024;
getLog().info("{} with file size {}", testName.getMethodName(), fileSizeMb);
- byte[] data1 = dataset((fileSizeMb + 1) * 1024 * 1024, 33, 43);
- createFile(srcFS, inputFile1, true, data1);
- byte[] data2 = dataset((fileSizeMb + 2) * 1024 * 1024, 43, 53);
- createFile(srcFS, inputFile2, true, data2);
- byte[] data3 = dataset((fileSizeMb + 3) * 1024 * 1024, 53, 63);
- createFile(srcFS, inputFile3, true, data3);
+ byte[] data1 = dataset((fileSizeMb + 1) * MB, 33, 43);
+ createFile(srcFS, largeFile1, true, data1);
+ byte[] data2 = dataset((fileSizeMb + 2) * MB, 43, 53);
+ createFile(srcFS, largeFile2, true, data2);
+ byte[] data3 = dataset((fileSizeMb + 3) * MB, 53, 63);
+ createFile(srcFS, largeFile3, true, data3);
Path target = new Path(dstDir, "outputDir");
runDistCp(inputDir, target);
ContractTestUtils.assertIsDirectory(dstFS, target);
@@ -183,12 +546,34 @@ public abstract class AbstractContractDistCpTest
* @throws Exception if there is a failure
*/
private void runDistCp(Path src, Path dst) throws Exception {
- DistCpOptions options = new DistCpOptions.Builder(
- Collections.singletonList(src), dst).build();
+ runDistCp(buildWithStandardOptions(
+ new DistCpOptions.Builder(Collections.singletonList(src), dst)));
+ }
+
+ /**
+ * Run the distcp job.
+ * @param optons distcp options
+ * @return the job. It will have already completed.
+ * @throws Exception failure
+ */
+ private Job runDistCp(final DistCpOptions options) throws Exception {
Job job = new DistCp(conf, options).execute();
assertNotNull("Unexpected null job returned from DistCp execution.", job);
assertTrue("DistCp job did not complete.", job.isComplete());
assertTrue("DistCp job did not complete successfully.", job.isSuccessful());
+ return job;
+ }
+
+ /**
+ * Add any standard options and then build.
+ * @param builder DistCp option builder
+ * @return the build options
+ */
+ private DistCpOptions buildWithStandardOptions(
+ DistCpOptions.Builder builder) {
+ return builder
+ .withNumListstatusThreads(8)
+ .build();
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/TestLocalContractDistCp.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/TestLocalContractDistCp.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/TestLocalContractDistCp.java
new file mode 100644
index 0000000..1a054c5
--- /dev/null
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/contract/TestLocalContractDistCp.java
@@ -0,0 +1,36 @@
+/*
+ * 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.tools.contract;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.contract.AbstractFSContract;
+import org.apache.hadoop.fs.contract.localfs.LocalFSContract;
+
+/**
+ * Verifies that the local FS passes all the tests in
+ * {@link AbstractContractDistCpTest}.
+ * As such, it acts as an in-module validation of this contract test itself.
+ */
+public class TestLocalContractDistCp extends AbstractContractDistCpTest {
+
+ @Override
+ protected AbstractFSContract createContract(Configuration conf) {
+ return new LocalFSContract(conf);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8f845598/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
index bf151cd..e00241b 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/mapred/TestCopyCommitter.java
@@ -43,6 +43,9 @@ import org.junit.*;
import java.io.IOException;
import java.util.*;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.*;
+import static org.apache.hadoop.tools.util.TestDistCpUtils.*;
+
public class TestCopyCommitter {
private static final Log LOG = LogFactory.getLog(TestCopyCommitter.class);
@@ -80,56 +83,42 @@ public class TestCopyCommitter {
}
@Before
- public void createMetaFolder() {
+ public void createMetaFolder() throws IOException {
config.set(DistCpConstants.CONF_LABEL_META_FOLDER, "/meta");
// Unset listing file path since the config is shared by
// multiple tests, and some test doesn't set it, such as
// testNoCommitAction, but the distcp code will check it.
config.set(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH, "");
Path meta = new Path("/meta");
- try {
- cluster.getFileSystem().mkdirs(meta);
- } catch (IOException e) {
- LOG.error("Exception encountered while creating meta folder", e);
- Assert.fail("Unable to create meta folder");
- }
+ cluster.getFileSystem().mkdirs(meta);
}
@After
- public void cleanupMetaFolder() {
+ public void cleanupMetaFolder() throws IOException {
Path meta = new Path("/meta");
- try {
- if (cluster.getFileSystem().exists(meta)) {
- cluster.getFileSystem().delete(meta, true);
- Assert.fail("Expected meta folder to be deleted");
- }
- } catch (IOException e) {
- LOG.error("Exception encountered while cleaning up folder", e);
- Assert.fail("Unable to clean up meta folder");
+ if (cluster.getFileSystem().exists(meta)) {
+ cluster.getFileSystem().delete(meta, true);
+ Assert.fail("Expected meta folder to be deleted");
}
}
@Test
- public void testNoCommitAction() {
+ public void testNoCommitAction() throws IOException {
TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
- JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
+ JobContext jobContext = new JobContextImpl(
+ taskAttemptContext.getConfiguration(),
taskAttemptContext.getTaskAttemptID().getJobID());
- try {
- OutputCommitter committer = new CopyCommitter(null, taskAttemptContext);
- committer.commitJob(jobContext);
- Assert.assertEquals(taskAttemptContext.getStatus(), "Commit Successful");
+ OutputCommitter committer = new CopyCommitter(null, taskAttemptContext);
+ committer.commitJob(jobContext);
+ Assert.assertEquals("Commit Successful", taskAttemptContext.getStatus());
- //Test for idempotent commit
- committer.commitJob(jobContext);
- Assert.assertEquals(taskAttemptContext.getStatus(), "Commit Successful");
- } catch (IOException e) {
- LOG.error("Exception encountered ", e);
- Assert.fail("Commit failed");
- }
+ //Test for idempotent commit
+ committer.commitJob(jobContext);
+ Assert.assertEquals("Commit Successful", taskAttemptContext.getStatus());
}
@Test
- public void testPreserveStatus() {
+ public void testPreserveStatus() throws IOException {
TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
taskAttemptContext.getTaskAttemptID().getJobID());
@@ -161,19 +150,12 @@ public class TestCopyCommitter {
conf.set(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH, targetBase);
committer.commitJob(jobContext);
- if (!checkDirectoryPermissions(fs, targetBase, sourcePerm)) {
- Assert.fail("Permission don't match");
- }
+ checkDirectoryPermissions(fs, targetBase, sourcePerm);
//Test for idempotent commit
committer.commitJob(jobContext);
- if (!checkDirectoryPermissions(fs, targetBase, sourcePerm)) {
- Assert.fail("Permission don't match");
- }
+ checkDirectoryPermissions(fs, targetBase, sourcePerm);
- } catch (IOException e) {
- LOG.error("Exception encountered while testing for preserve status", e);
- Assert.fail("Preserve status failure");
} finally {
TestDistCpUtils.delete(fs, "/tmp1");
conf.unset(DistCpConstants.CONF_LABEL_PRESERVE_STATUS);
@@ -182,7 +164,7 @@ public class TestCopyCommitter {
}
@Test
- public void testDeleteMissing() {
+ public void testDeleteMissing() throws IOException {
TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
taskAttemptContext.getTaskAttemptID().getJobID());
@@ -213,24 +195,13 @@ public class TestCopyCommitter {
conf.set(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH, targetBase);
committer.commitJob(jobContext);
- if (!TestDistCpUtils.checkIfFoldersAreInSync(fs, targetBase, sourceBase)) {
- Assert.fail("Source and target folders are not in sync");
- }
- if (!TestDistCpUtils.checkIfFoldersAreInSync(fs, sourceBase, targetBase)) {
- Assert.fail("Source and target folders are not in sync");
- }
+ verifyFoldersAreInSync(fs, targetBase, sourceBase);
+ verifyFoldersAreInSync(fs, sourceBase, targetBase);
//Test for idempotent commit
committer.commitJob(jobContext);
- if (!TestDistCpUtils.checkIfFoldersAreInSync(fs, targetBase, sourceBase)) {
- Assert.fail("Source and target folders are not in sync");
- }
- if (!TestDistCpUtils.checkIfFoldersAreInSync(fs, sourceBase, targetBase)) {
- Assert.fail("Source and target folders are not in sync");
- }
- } catch (Throwable e) {
- LOG.error("Exception encountered while testing for delete missing", e);
- Assert.fail("Delete missing failure");
+ verifyFoldersAreInSync(fs, targetBase, sourceBase);
+ verifyFoldersAreInSync(fs, sourceBase, targetBase);
} finally {
TestDistCpUtils.delete(fs, "/tmp1");
conf.set(DistCpConstants.CONF_LABEL_DELETE_MISSING, "false");
@@ -238,7 +209,7 @@ public class TestCopyCommitter {
}
@Test
- public void testDeleteMissingFlatInterleavedFiles() {
+ public void testDeleteMissingFlatInterleavedFiles() throws IOException {
TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
taskAttemptContext.getTaskAttemptID().getJobID());
@@ -253,20 +224,20 @@ public class TestCopyCommitter {
fs = FileSystem.get(conf);
sourceBase = "/tmp1/" + String.valueOf(rand.nextLong());
targetBase = "/tmp1/" + String.valueOf(rand.nextLong());
- TestDistCpUtils.createFile(fs, sourceBase + "/1");
- TestDistCpUtils.createFile(fs, sourceBase + "/3");
- TestDistCpUtils.createFile(fs, sourceBase + "/4");
- TestDistCpUtils.createFile(fs, sourceBase + "/5");
- TestDistCpUtils.createFile(fs, sourceBase + "/7");
- TestDistCpUtils.createFile(fs, sourceBase + "/8");
- TestDistCpUtils.createFile(fs, sourceBase + "/9");
-
- TestDistCpUtils.createFile(fs, targetBase + "/2");
- TestDistCpUtils.createFile(fs, targetBase + "/4");
- TestDistCpUtils.createFile(fs, targetBase + "/5");
- TestDistCpUtils.createFile(fs, targetBase + "/7");
- TestDistCpUtils.createFile(fs, targetBase + "/9");
- TestDistCpUtils.createFile(fs, targetBase + "/A");
+ createFile(fs, sourceBase + "/1");
+ createFile(fs, sourceBase + "/3");
+ createFile(fs, sourceBase + "/4");
+ createFile(fs, sourceBase + "/5");
+ createFile(fs, sourceBase + "/7");
+ createFile(fs, sourceBase + "/8");
+ createFile(fs, sourceBase + "/9");
+
+ createFile(fs, targetBase + "/2");
+ createFile(fs, targetBase + "/4");
+ createFile(fs, targetBase + "/5");
+ createFile(fs, targetBase + "/7");
+ createFile(fs, targetBase + "/9");
+ createFile(fs, targetBase + "/A");
final DistCpOptions options = new DistCpOptions.Builder(
Collections.singletonList(new Path(sourceBase)), new Path("/out"))
@@ -282,20 +253,13 @@ public class TestCopyCommitter {
conf.set(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH, targetBase);
committer.commitJob(jobContext);
- if (!TestDistCpUtils.checkIfFoldersAreInSync(fs, targetBase, sourceBase)) {
- Assert.fail("Source and target folders are not in sync");
- }
- Assert.assertEquals(fs.listStatus(new Path(targetBase)).length, 4);
+ verifyFoldersAreInSync(fs, targetBase, sourceBase);
+ Assert.assertEquals(4, fs.listStatus(new Path(targetBase)).length);
//Test for idempotent commit
committer.commitJob(jobContext);
- if (!TestDistCpUtils.checkIfFoldersAreInSync(fs, targetBase, sourceBase)) {
- Assert.fail("Source and target folders are not in sync");
- }
- Assert.assertEquals(fs.listStatus(new Path(targetBase)).length, 4);
- } catch (IOException e) {
- LOG.error("Exception encountered while testing for delete missing", e);
- Assert.fail("Delete missing failure");
+ verifyFoldersAreInSync(fs, targetBase, sourceBase);
+ Assert.assertEquals(4, fs.listStatus(new Path(targetBase)).length);
} finally {
TestDistCpUtils.delete(fs, "/tmp1");
conf.set(DistCpConstants.CONF_LABEL_DELETE_MISSING, "false");
@@ -304,7 +268,7 @@ public class TestCopyCommitter {
}
@Test
- public void testAtomicCommitMissingFinal() {
+ public void testAtomicCommitMissingFinal() throws IOException {
TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
taskAttemptContext.getTaskAttemptID().getJobID());
@@ -322,19 +286,16 @@ public class TestCopyCommitter {
conf.set(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH, finalPath);
conf.setBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, true);
- Assert.assertTrue(fs.exists(new Path(workPath)));
- Assert.assertFalse(fs.exists(new Path(finalPath)));
+ assertPathExists(fs, "Work path", new Path(workPath));
+ assertPathDoesNotExist(fs, "Final path", new Path(finalPath));
committer.commitJob(jobContext);
- Assert.assertFalse(fs.exists(new Path(workPath)));
- Assert.assertTrue(fs.exists(new Path(finalPath)));
+ assertPathDoesNotExist(fs, "Work path", new Path(workPath));
+ assertPathExists(fs, "Final path", new Path(finalPath));
//Test for idempotent commit
committer.commitJob(jobContext);
- Assert.assertFalse(fs.exists(new Path(workPath)));
- Assert.assertTrue(fs.exists(new Path(finalPath)));
- } catch (IOException e) {
- LOG.error("Exception encountered while testing for preserve status", e);
- Assert.fail("Atomic commit failure");
+ assertPathDoesNotExist(fs, "Work path", new Path(workPath));
+ assertPathExists(fs, "Final path", new Path(finalPath));
} finally {
TestDistCpUtils.delete(fs, workPath);
TestDistCpUtils.delete(fs, finalPath);
@@ -343,7 +304,7 @@ public class TestCopyCommitter {
}
@Test
- public void testAtomicCommitExistingFinal() {
+ public void testAtomicCommitExistingFinal() throws IOException {
TaskAttemptContext taskAttemptContext = getTaskAttemptContext(config);
JobContext jobContext = new JobContextImpl(taskAttemptContext.getConfiguration(),
taskAttemptContext.getTaskAttemptID().getJobID());
@@ -363,20 +324,17 @@ public class TestCopyCommitter {
conf.set(DistCpConstants.CONF_LABEL_TARGET_FINAL_PATH, finalPath);
conf.setBoolean(DistCpConstants.CONF_LABEL_ATOMIC_COPY, true);
- Assert.assertTrue(fs.exists(new Path(workPath)));
- Assert.assertTrue(fs.exists(new Path(finalPath)));
+ assertPathExists(fs, "Work path", new Path(workPath));
+ assertPathExists(fs, "Final path", new Path(finalPath));
try {
committer.commitJob(jobContext);
Assert.fail("Should not be able to atomic-commit to pre-existing path.");
} catch(Exception exception) {
- Assert.assertTrue(fs.exists(new Path(workPath)));
- Assert.assertTrue(fs.exists(new Path(finalPath)));
+ assertPathExists(fs, "Work path", new Path(workPath));
+ assertPathExists(fs, "Final path", new Path(finalPath));
LOG.info("Atomic-commit Test pass.");
}
- } catch (IOException e) {
- LOG.error("Exception encountered while testing for atomic commit.", e);
- Assert.fail("Atomic commit failure");
} finally {
TestDistCpUtils.delete(fs, workPath);
TestDistCpUtils.delete(fs, finalPath);
@@ -389,11 +347,11 @@ public class TestCopyCommitter {
new TaskAttemptID("200707121733", 1, TaskType.MAP, 1, 1));
}
- private boolean checkDirectoryPermissions(FileSystem fs, String targetBase,
- FsPermission sourcePerm) throws IOException {
+ private void checkDirectoryPermissions(FileSystem fs, String targetBase,
+ FsPermission sourcePerm) throws IOException {
Path base = new Path(targetBase);
- Stack<Path> stack = new Stack<Path>();
+ Stack<Path> stack = new Stack<>();
stack.push(base);
while (!stack.isEmpty()) {
Path file = stack.pop();
@@ -404,11 +362,10 @@ public class TestCopyCommitter {
for (FileStatus status : fStatus) {
if (status.isDirectory()) {
stack.push(status.getPath());
- Assert.assertEquals(status.getPermission(), sourcePerm);
+ Assert.assertEquals(sourcePerm, status.getPermission());
}
}
}
- return true;
}
private static class NullInputFormat extends InputFormat {
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org