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 tu...@apache.org on 2014/08/29 00:06:57 UTC
[02/11] HDFS-6134 and HADOOP-10150 subtasks.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm
index 41b381a..3e8de4f 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm
@@ -191,6 +191,26 @@ $H3 Update and Overwrite
If `-update` is used, `1` is overwritten as well.
+$H3 raw Namespace Extended Attribute Preservation
+
+ This section only applies to HDFS.
+
+ If the target and all of the source pathnames are in the /.reserved/raw
+ hierarchy, then 'raw' namespace extended attributes will be preserved.
+ 'raw' xattrs are used by the system for internal functions such as encryption
+ meta data. They are only visible to users when accessed through the
+ /.reserved/raw hierarchy.
+
+ raw xattrs are preserved based solely on whether /.reserved/raw prefixes are
+ supplied. The -p (preserve, see below) flag does not impact preservation of
+ raw xattrs.
+
+ To prevent raw xattrs from being preserved, simply do not use the
+ /.reserved/raw prefix on any of the source and target paths.
+
+ If the /.reserved/raw prefix is specified on only a subset of the source and
+ target paths, an error will be displayed and a non-0 exit code returned.
+
Command Line Options
--------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMerger.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMerger.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMerger.java
index 1aea500..c5ab420 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMerger.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestMerger.java
@@ -24,14 +24,16 @@ import static org.mockito.Mockito.doAnswer;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import org.apache.hadoop.fs.FSDataInputStream;
import org.junit.Assert;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
@@ -51,10 +53,16 @@ import org.apache.hadoop.mapred.RawKeyValueIterator;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.MRConfig;
+import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskID;
import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.security.TokenCache;
+import org.apache.hadoop.mapreduce.CryptoUtils;
import org.apache.hadoop.mapreduce.task.reduce.MergeManagerImpl;
+import org.apache.hadoop.mapreduce.task.reduce.MergeManagerImpl.CompressAwarePath;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Progress;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
@@ -63,40 +71,48 @@ import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import com.google.common.collect.Lists;
+
public class TestMerger {
private Configuration conf;
private JobConf jobConf;
private FileSystem fs;
-
+
@Before
public void setup() throws IOException {
conf = new Configuration();
jobConf = new JobConf();
fs = FileSystem.getLocal(conf);
}
-
- @After
- public void cleanup() throws IOException {
- fs.delete(new Path(jobConf.getLocalDirs()[0]), true);
+
+
+ @Test
+ public void testEncryptedMerger() throws Throwable {
+ jobConf.setBoolean(MRJobConfig.MR_ENCRYPTED_INTERMEDIATE_DATA, true);
+ conf.setBoolean(MRJobConfig.MR_ENCRYPTED_INTERMEDIATE_DATA, true);
+ Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials();
+ TokenCache.setShuffleSecretKey(new byte[16], credentials);
+ UserGroupInformation.getCurrentUser().addCredentials(credentials);
+ testInMemoryAndOnDiskMerger();
}
-
+
@Test
- public void testInMemoryMerger() throws Throwable {
+ public void testInMemoryAndOnDiskMerger() throws Throwable {
JobID jobId = new JobID("a", 0);
- TaskAttemptID reduceId = new TaskAttemptID(
+ TaskAttemptID reduceId1 = new TaskAttemptID(
new TaskID(jobId, TaskType.REDUCE, 0), 0);
TaskAttemptID mapId1 = new TaskAttemptID(
new TaskID(jobId, TaskType.MAP, 1), 0);
TaskAttemptID mapId2 = new TaskAttemptID(
new TaskID(jobId, TaskType.MAP, 2), 0);
-
+
LocalDirAllocator lda = new LocalDirAllocator(MRConfig.LOCAL_DIR);
-
+
MergeManagerImpl<Text, Text> mergeManager = new MergeManagerImpl<Text, Text>(
- reduceId, jobConf, fs, lda, Reporter.NULL, null, null, null, null, null,
+ reduceId1, jobConf, fs, lda, Reporter.NULL, null, null, null, null, null,
null, null, new Progress(), new MROutputFiles());
-
+
// write map outputs
Map<String, String> map1 = new TreeMap<String, String>();
map1.put("apple", "disgusting");
@@ -113,32 +129,88 @@ public class TestMerger {
mapOutputBytes1.length);
System.arraycopy(mapOutputBytes2, 0, mapOutput2.getMemory(), 0,
mapOutputBytes2.length);
-
+
// create merger and run merge
MergeThread<InMemoryMapOutput<Text, Text>, Text, Text> inMemoryMerger =
mergeManager.createInMemoryMerger();
- List<InMemoryMapOutput<Text, Text>> mapOutputs =
+ List<InMemoryMapOutput<Text, Text>> mapOutputs1 =
new ArrayList<InMemoryMapOutput<Text, Text>>();
- mapOutputs.add(mapOutput1);
- mapOutputs.add(mapOutput2);
-
- inMemoryMerger.merge(mapOutputs);
-
+ mapOutputs1.add(mapOutput1);
+ mapOutputs1.add(mapOutput2);
+
+ inMemoryMerger.merge(mapOutputs1);
+
Assert.assertEquals(1, mergeManager.onDiskMapOutputs.size());
- Path outPath = mergeManager.onDiskMapOutputs.iterator().next();
-
+
+ TaskAttemptID reduceId2 = new TaskAttemptID(
+ new TaskID(jobId, TaskType.REDUCE, 3), 0);
+ TaskAttemptID mapId3 = new TaskAttemptID(
+ new TaskID(jobId, TaskType.MAP, 4), 0);
+ TaskAttemptID mapId4 = new TaskAttemptID(
+ new TaskID(jobId, TaskType.MAP, 5), 0);
+ // write map outputs
+ Map<String, String> map3 = new TreeMap<String, String>();
+ map3.put("apple", "awesome");
+ map3.put("carrot", "amazing");
+ Map<String, String> map4 = new TreeMap<String, String>();
+ map4.put("banana", "bla");
+ byte[] mapOutputBytes3 = writeMapOutput(conf, map3);
+ byte[] mapOutputBytes4 = writeMapOutput(conf, map4);
+ InMemoryMapOutput<Text, Text> mapOutput3 = new InMemoryMapOutput<Text, Text>(
+ conf, mapId3, mergeManager, mapOutputBytes3.length, null, true);
+ InMemoryMapOutput<Text, Text> mapOutput4 = new InMemoryMapOutput<Text, Text>(
+ conf, mapId4, mergeManager, mapOutputBytes4.length, null, true);
+ System.arraycopy(mapOutputBytes3, 0, mapOutput3.getMemory(), 0,
+ mapOutputBytes3.length);
+ System.arraycopy(mapOutputBytes4, 0, mapOutput4.getMemory(), 0,
+ mapOutputBytes4.length);
+
+// // create merger and run merge
+ MergeThread<InMemoryMapOutput<Text, Text>, Text, Text> inMemoryMerger2 =
+ mergeManager.createInMemoryMerger();
+ List<InMemoryMapOutput<Text, Text>> mapOutputs2 =
+ new ArrayList<InMemoryMapOutput<Text, Text>>();
+ mapOutputs2.add(mapOutput3);
+ mapOutputs2.add(mapOutput4);
+
+ inMemoryMerger2.merge(mapOutputs2);
+
+ Assert.assertEquals(2, mergeManager.onDiskMapOutputs.size());
+
+ List<CompressAwarePath> paths = new ArrayList<CompressAwarePath>();
+ Iterator<CompressAwarePath> iterator = mergeManager.onDiskMapOutputs.iterator();
List<String> keys = new ArrayList<String>();
List<String> values = new ArrayList<String>();
- readOnDiskMapOutput(conf, fs, outPath, keys, values);
- Assert.assertEquals(keys, Arrays.asList("apple", "banana", "carrot"));
- Assert.assertEquals(values, Arrays.asList("disgusting", "pretty good", "delicious"));
+ while (iterator.hasNext()) {
+ CompressAwarePath next = iterator.next();
+ readOnDiskMapOutput(conf, fs, next, keys, values);
+ paths.add(next);
+ }
+ Assert.assertEquals(keys, Arrays.asList("apple", "banana", "carrot", "apple", "banana", "carrot"));
+ Assert.assertEquals(values, Arrays.asList("awesome", "bla", "amazing", "disgusting", "pretty good", "delicious"));
+ mergeManager.close();
+
+ mergeManager = new MergeManagerImpl<Text, Text>(
+ reduceId2, jobConf, fs, lda, Reporter.NULL, null, null, null, null, null,
+ null, null, new Progress(), new MROutputFiles());
+
+ MergeThread<CompressAwarePath,Text,Text> onDiskMerger = mergeManager.createOnDiskMerger();
+ onDiskMerger.merge(paths);
+
+ Assert.assertEquals(1, mergeManager.onDiskMapOutputs.size());
+
+ keys = new ArrayList<String>();
+ values = new ArrayList<String>();
+ readOnDiskMapOutput(conf, fs, mergeManager.onDiskMapOutputs.iterator().next(), keys, values);
+ Assert.assertEquals(keys, Arrays.asList("apple", "apple", "banana", "banana", "carrot", "carrot"));
+ Assert.assertEquals(values, Arrays.asList("awesome", "disgusting", "pretty good", "bla", "amazing", "delicious"));
mergeManager.close();
Assert.assertEquals(0, mergeManager.inMemoryMapOutputs.size());
Assert.assertEquals(0, mergeManager.inMemoryMergedMapOutputs.size());
Assert.assertEquals(0, mergeManager.onDiskMapOutputs.size());
}
-
+
private byte[] writeMapOutput(Configuration conf, Map<String, String> keysToValues)
throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
@@ -152,11 +224,13 @@ public class TestMerger {
writer.close();
return baos.toByteArray();
}
-
+
private void readOnDiskMapOutput(Configuration conf, FileSystem fs, Path path,
List<String> keys, List<String> values) throws IOException {
- IFile.Reader<Text, Text> reader = new IFile.Reader<Text, Text>(conf, fs,
- path, null, null);
+ FSDataInputStream in = CryptoUtils.wrapIfNecessary(conf, fs.open(path));
+
+ IFile.Reader<Text, Text> reader = new IFile.Reader<Text, Text>(conf, in,
+ fs.getFileStatus(path).getLen(), null, null);
DataInputBuffer keyBuff = new DataInputBuffer();
DataInputBuffer valueBuff = new DataInputBuffer();
Text key = new Text();
@@ -169,17 +243,17 @@ public class TestMerger {
values.add(value.toString());
}
}
-
+
@Test
public void testCompressed() throws IOException {
testMergeShouldReturnProperProgress(getCompressedSegments());
- }
-
+}
+
@Test
public void testUncompressed() throws IOException {
testMergeShouldReturnProperProgress(getUncompressedSegments());
}
-
+
@SuppressWarnings( { "deprecation", "unchecked" })
public void testMergeShouldReturnProperProgress(
List<Segment<Text, Text>> segments) throws IOException {
@@ -212,7 +286,7 @@ public class TestMerger {
}
return segments;
}
-
+
private List<Segment<Text, Text>> getCompressedSegments() throws IOException {
List<Segment<Text, Text>> segments = new ArrayList<Segment<Text, Text>>();
for (int i = 1; i < 1; i++) {
@@ -220,7 +294,7 @@ public class TestMerger {
}
return segments;
}
-
+
private Segment<Text, Text> getUncompressedSegment(int i) throws IOException {
return new Segment<Text, Text>(getReader(i), false);
}
@@ -258,7 +332,7 @@ public class TestMerger {
}
};
}
-
+
private Answer<?> getValueAnswer(final String segmentName) {
return new Answer<Void>() {
int i = 0;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestIFile.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestIFile.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestIFile.java
index e3c7253..a314fc1 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestIFile.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestIFile.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.mapred;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
@@ -42,7 +44,7 @@ public class TestIFile {
DefaultCodec codec = new GzipCodec();
codec.setConf(conf);
IFile.Writer<Text, Text> writer =
- new IFile.Writer<Text, Text>(conf, rfs, path, Text.class, Text.class,
+ new IFile.Writer<Text, Text>(conf, rfs.create(path), Text.class, Text.class,
codec, null);
writer.close();
}
@@ -56,12 +58,15 @@ public class TestIFile {
Path path = new Path(new Path("build/test.ifile"), "data");
DefaultCodec codec = new GzipCodec();
codec.setConf(conf);
+ FSDataOutputStream out = rfs.create(path);
IFile.Writer<Text, Text> writer =
- new IFile.Writer<Text, Text>(conf, rfs, path, Text.class, Text.class,
+ new IFile.Writer<Text, Text>(conf, out, Text.class, Text.class,
codec, null);
writer.close();
+ FSDataInputStream in = rfs.open(path);
IFile.Reader<Text, Text> reader =
- new IFile.Reader<Text, Text>(conf, rfs, path, codec, null);
+ new IFile.Reader<Text, Text>(conf, in, rfs.getFileStatus(path).getLen(),
+ codec, null);
reader.close();
// test check sum
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRIntermediateDataEncryption.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRIntermediateDataEncryption.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRIntermediateDataEncryption.java
new file mode 100644
index 0000000..ebc32ad
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestMRIntermediateDataEncryption.java
@@ -0,0 +1,254 @@
+/**
+ * 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.mapred;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+
+import static org.junit.Assert.*;
+
+@SuppressWarnings(value={"unchecked", "deprecation"})
+/**
+ * This test tests the support for a merge operation in Hadoop. The input files
+ * are already sorted on the key. This test implements an external
+ * MapOutputCollector implementation that just copies the records to different
+ * partitions while maintaining the sort order in each partition. The Hadoop
+ * framework's merge on the reduce side will merge the partitions created to
+ * generate the final output which is sorted on the key.
+ */
+public class TestMRIntermediateDataEncryption {
+ // Where MR job's input will reside.
+ private static final Path INPUT_DIR = new Path("/test/input");
+ // Where output goes.
+ private static final Path OUTPUT = new Path("/test/output");
+
+ @Test
+ public void testSingleReducer() throws Exception {
+ doEncryptionTest(3, 1, 2);
+ }
+
+ @Test
+ public void testMultipleMapsPerNode() throws Exception {
+ doEncryptionTest(8, 1, 2);
+ }
+
+ @Test
+ public void testMultipleReducers() throws Exception {
+ doEncryptionTest(2, 4, 2);
+ }
+
+ public void doEncryptionTest(int numMappers, int numReducers, int numNodes) throws Exception {
+ doEncryptionTest(numMappers, numReducers, numNodes, 1000);
+ }
+
+ public void doEncryptionTest(int numMappers, int numReducers, int numNodes, int numLines) throws Exception {
+ MiniDFSCluster dfsCluster = null;
+ MiniMRClientCluster mrCluster = null;
+ FileSystem fileSystem = null;
+ try {
+ Configuration conf = new Configuration();
+ // Start the mini-MR and mini-DFS clusters
+
+ dfsCluster = new MiniDFSCluster.Builder(conf)
+ .numDataNodes(numNodes).build();
+ fileSystem = dfsCluster.getFileSystem();
+ mrCluster = MiniMRClientClusterFactory.create(this.getClass(),
+ numNodes, conf);
+ // Generate input.
+ createInput(fileSystem, numMappers, numLines);
+ // Run the test.
+ runMergeTest(new JobConf(mrCluster.getConfig()), fileSystem, numMappers, numReducers, numLines);
+ } finally {
+ if (dfsCluster != null) {
+ dfsCluster.shutdown();
+ }
+ if (mrCluster != null) {
+ mrCluster.stop();
+ }
+ }
+ }
+
+ private void createInput(FileSystem fs, int numMappers, int numLines) throws Exception {
+ fs.delete(INPUT_DIR, true);
+ for (int i = 0; i < numMappers; i++) {
+ OutputStream os = fs.create(new Path(INPUT_DIR, "input_" + i + ".txt"));
+ Writer writer = new OutputStreamWriter(os);
+ for (int j = 0; j < numLines; j++) {
+ // Create sorted key, value pairs.
+ int k = j + 1;
+ String formattedNumber = String.format("%09d", k);
+ writer.write(formattedNumber + " " + formattedNumber + "\n");
+ }
+ writer.close();
+ }
+ }
+
+ private void runMergeTest(JobConf job, FileSystem fileSystem, int numMappers, int numReducers, int numLines)
+ throws Exception {
+ fileSystem.delete(OUTPUT, true);
+ job.setJobName("Test");
+ JobClient client = new JobClient(job);
+ RunningJob submittedJob = null;
+ FileInputFormat.setInputPaths(job, INPUT_DIR);
+ FileOutputFormat.setOutputPath(job, OUTPUT);
+ job.set("mapreduce.output.textoutputformat.separator", " ");
+ job.setInputFormat(TextInputFormat.class);
+ job.setMapOutputKeyClass(Text.class);
+ job.setMapOutputValueClass(Text.class);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(Text.class);
+ job.setMapperClass(MyMapper.class);
+ job.setPartitionerClass(MyPartitioner.class);
+ job.setOutputFormat(TextOutputFormat.class);
+ job.setNumReduceTasks(numReducers);
+
+ job.setInt("mapreduce.map.maxattempts", 1);
+ job.setInt("mapreduce.reduce.maxattempts", 1);
+ job.setInt("mapred.test.num_lines", numLines);
+ job.setBoolean(MRJobConfig.MR_ENCRYPTED_INTERMEDIATE_DATA, true);
+ try {
+ submittedJob = client.submitJob(job);
+ try {
+ if (! client.monitorAndPrintJob(job, submittedJob)) {
+ throw new IOException("Job failed!");
+ }
+ } catch(InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ } catch(IOException ioe) {
+ System.err.println("Job failed with: " + ioe);
+ } finally {
+ verifyOutput(submittedJob, fileSystem, numMappers, numLines);
+ }
+ }
+
+ private void verifyOutput(RunningJob submittedJob, FileSystem fileSystem, int numMappers, int numLines)
+ throws Exception {
+ FSDataInputStream dis = null;
+ long numValidRecords = 0;
+ long numInvalidRecords = 0;
+ String prevKeyValue = "000000000";
+ Path[] fileList =
+ FileUtil.stat2Paths(fileSystem.listStatus(OUTPUT,
+ new Utils.OutputFileUtils.OutputFilesFilter()));
+ for (Path outFile : fileList) {
+ try {
+ dis = fileSystem.open(outFile);
+ String record;
+ while((record = dis.readLine()) != null) {
+ // Split the line into key and value.
+ int blankPos = record.indexOf(" ");
+ String keyString = record.substring(0, blankPos);
+ String valueString = record.substring(blankPos+1);
+ // Check for sorted output and correctness of record.
+ if (keyString.compareTo(prevKeyValue) >= 0
+ && keyString.equals(valueString)) {
+ prevKeyValue = keyString;
+ numValidRecords++;
+ } else {
+ numInvalidRecords++;
+ }
+ }
+ } finally {
+ if (dis != null) {
+ dis.close();
+ dis = null;
+ }
+ }
+ }
+ // Make sure we got all input records in the output in sorted order.
+ assertEquals((long)(numMappers * numLines), numValidRecords);
+ // Make sure there is no extraneous invalid record.
+ assertEquals(0, numInvalidRecords);
+ }
+
+ /**
+ * A mapper implementation that assumes that key text contains valid integers
+ * in displayable form.
+ */
+ public static class MyMapper extends MapReduceBase
+ implements Mapper<LongWritable, Text, Text, Text> {
+ private Text keyText;
+ private Text valueText;
+
+ public MyMapper() {
+ keyText = new Text();
+ valueText = new Text();
+ }
+
+ @Override
+ public void map(LongWritable key, Text value,
+ OutputCollector<Text, Text> output,
+ Reporter reporter) throws IOException {
+ String record = value.toString();
+ int blankPos = record.indexOf(" ");
+ keyText.set(record.substring(0, blankPos));
+ valueText.set(record.substring(blankPos+1));
+ output.collect(keyText, valueText);
+ }
+
+ public void close() throws IOException {
+ }
+ }
+
+ /**
+ * Partitioner implementation to make sure that output is in total sorted
+ * order. We basically route key ranges to different reducers such that
+ * key values monotonically increase with the partition number. For example,
+ * in this test, the keys are numbers from 1 to 1000 in the form "000000001"
+ * to "000001000" in each input file. The keys "000000001" to "000000250" are
+ * routed to partition 0, "000000251" to "000000500" are routed to partition 1
+ * and so on since we have 4 reducers.
+ */
+ static class MyPartitioner implements Partitioner<Text, Text> {
+
+ private JobConf job;
+
+ public MyPartitioner() {
+ }
+
+ public void configure(JobConf job) {
+ this.job = job;
+ }
+
+ public int getPartition(Text key, Text value, int numPartitions) {
+ int keyValue = 0;
+ try {
+ keyValue = Integer.parseInt(key.toString());
+ } catch(NumberFormatException nfe) {
+ keyValue = 0;
+ }
+ int partitionNumber = (numPartitions*(Math.max(0, keyValue-1)))/job.getInt("mapred.test.num_lines", 10000);
+ return partitionNumber;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestReduceTask.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestReduceTask.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestReduceTask.java
index d3a0844..43fd948 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestReduceTask.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestReduceTask.java
@@ -80,7 +80,7 @@ public class TestReduceTask extends TestCase {
FileSystem rfs = ((LocalFileSystem)localFs).getRaw();
Path path = new Path(tmpDir, "data.in");
IFile.Writer<Text, Text> writer =
- new IFile.Writer<Text, Text>(conf, rfs, path, Text.class, Text.class,
+ new IFile.Writer<Text, Text>(conf, rfs.create(path), Text.class, Text.class,
codec, null);
for(Pair p: vals) {
writer.append(new Text(p.key), new Text(p.value));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/pipes/TestPipeApplication.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/pipes/TestPipeApplication.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/pipes/TestPipeApplication.java
index 69994f3..f447ebc 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/pipes/TestPipeApplication.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/pipes/TestPipeApplication.java
@@ -95,9 +95,9 @@ public class TestPipeApplication {
new Counters.Counter(), new Progress());
FileSystem fs = new RawLocalFileSystem();
fs.setConf(conf);
- Writer<IntWritable, Text> wr = new Writer<IntWritable, Text>(conf, fs,
- new Path(workSpace + File.separator + "outfile"), IntWritable.class,
- Text.class, null, null);
+ Writer<IntWritable, Text> wr = new Writer<IntWritable, Text>(conf, fs.create(
+ new Path(workSpace + File.separator + "outfile")), IntWritable.class,
+ Text.class, null, null, true);
output.setWriter(wr);
// stub for client
File fCommand = getFileCommand("org.apache.hadoop.mapred.pipes.PipeApplicationRunnableStub");
@@ -177,9 +177,9 @@ public class TestPipeApplication {
new Progress());
FileSystem fs = new RawLocalFileSystem();
fs.setConf(conf);
- Writer<IntWritable, Text> wr = new Writer<IntWritable, Text>(conf, fs,
- new Path(workSpace.getAbsolutePath() + File.separator + "outfile"),
- IntWritable.class, Text.class, null, null);
+ Writer<IntWritable, Text> wr = new Writer<IntWritable, Text>(conf, fs.create(
+ new Path(workSpace.getAbsolutePath() + File.separator + "outfile")),
+ IntWritable.class, Text.class, null, null, true);
output.setWriter(wr);
conf.set(Submitter.PRESERVE_COMMANDFILE, "true");
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-project-dist/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-project-dist/pom.xml b/hadoop-project-dist/pom.xml
index 94af108..3b7a8ad 100644
--- a/hadoop-project-dist/pom.xml
+++ b/hadoop-project-dist/pom.xml
@@ -41,6 +41,8 @@
<hadoop.component>UNDEF</hadoop.component>
<bundle.snappy>false</bundle.snappy>
<bundle.snappy.in.bin>false</bundle.snappy.in.bin>
+ <bundle.openssl>false</bundle.openssl>
+ <bundle.openssl.in.bin>false</bundle.openssl.in.bin>
</properties>
<dependencies>
@@ -351,6 +353,10 @@
cd "${snappy.lib}"
$$TAR *snappy* | (cd $${TARGET_DIR}/; $$UNTAR)
fi
+ if [ "${bundle.openssl}" = "true" ] ; then
+ cd "${openssl.lib}"
+ $$TAR *crypto* | (cd $${TARGET_DIR}/; $$UNTAR)
+ fi
fi
BIN_DIR="${BUILD_DIR}/bin"
if [ -d $${BIN_DIR} ] ; then
@@ -364,6 +370,12 @@
$$TAR *snappy* | (cd $${TARGET_BIN_DIR}/; $$UNTAR)
fi
fi
+ if [ "${bundle.openssl.in.bin}" = "true" ] ; then
+ if [ "${bundle.openssl}" = "true" ] ; then
+ cd "${openssl.lib}"
+ $$TAR *crypto* | (cd $${TARGET_BIN_DIR}/; $$UNTAR)
+ fi
+ fi
fi
</echo>
<exec executable="sh" dir="${project.build.directory}" failonerror="true">
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-project/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-project/pom.xml b/hadoop-project/pom.xml
index e24992d..98e72a4 100644
--- a/hadoop-project/pom.xml
+++ b/hadoop-project/pom.xml
@@ -1108,6 +1108,7 @@
<!-- attempt to open a file at this path. -->
<java.security.egd>file:/dev/urandom</java.security.egd>
<bundle.snappy.in.bin>true</bundle.snappy.in.bin>
+ <bundle.openssl.in.bin>true</bundle.openssl.in.bin>
</properties>
<build>
<plugins>
@@ -1118,6 +1119,7 @@
<environmentVariables>
<!-- Specify where to look for the native DLL on Windows -->
<PATH>${env.PATH};${hadoop.common.build.dir}/bin;${snappy.lib}</PATH>
+ <PATH>${env.PATH};${hadoop.common.build.dir}/bin;${openssl.lib}</PATH>
</environmentVariables>
</configuration>
</plugin>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-project/src/site/site.xml
----------------------------------------------------------------------
diff --git a/hadoop-project/src/site/site.xml b/hadoop-project/src/site/site.xml
index 8cb0168..a42aff0 100644
--- a/hadoop-project/src/site/site.xml
+++ b/hadoop-project/src/site/site.xml
@@ -91,6 +91,7 @@
<item name="HDFS NFS Gateway" href="hadoop-project-dist/hadoop-hdfs/HdfsNfsGateway.html"/>
<item name="HDFS Rolling Upgrade" href="hadoop-project-dist/hadoop-hdfs/HdfsRollingUpgrade.html"/>
<item name="Extended Attributes" href="hadoop-project-dist/hadoop-hdfs/ExtendedAttributes.html"/>
+ <item name="Transparent Encryption" href="hadoop-project-dist/hadoop-hdfs/TransparentEncryption.html"/>
<item name="HDFS Support for Multihoming" href="hadoop-project-dist/hadoop-hdfs/HdfsMultihoming.html"/>
</menu>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/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 d1dba19..7e71096 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
@@ -42,6 +42,8 @@ public class DistCpConstants {
public static final String CONF_LABEL_LOG_PATH = "distcp.log.path";
public static final String CONF_LABEL_IGNORE_FAILURES = "distcp.ignore.failures";
public static final String CONF_LABEL_PRESERVE_STATUS = "distcp.preserve.status";
+ public static final String CONF_LABEL_PRESERVE_RAWXATTRS =
+ "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_SSL_CONF = "distcp.keystore.resource";
@@ -128,4 +130,8 @@ public class DistCpConstants {
public static final int MIN_RECORDS_PER_CHUNK_DEFAULT = 5;
public static final int SPLIT_RATIO_DEFAULT = 2;
+ /**
+ * Value of reserved raw HDFS directory when copying raw.* xattrs.
+ */
+ static final String HDFS_RESERVED_RAW_DIRECTORY_NAME = "/.reserved/raw";
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/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 e77b6e1..c544813 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
@@ -48,7 +48,11 @@ public enum DistCpOptionSwitch {
new Option("p", true, "preserve status (rbugpcax)(replication, " +
"block-size, user, group, permission, checksum-type, ACL, XATTR). " +
"If -p is specified with no <arg>, then preserves replication, " +
- "block size, user, group, permission and checksum type.")),
+ "block size, user, group, permission and checksum type." +
+ "raw.* xattrs are preserved when both the source and destination " +
+ "paths are in the /.reserved/raw hierarchy (HDFS only). raw.* xattr" +
+ "preservation is independent of the -p flag." +
+ "Refer to the DistCp documentation for more details.")),
/**
* Update target location by copying only files that are missing
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/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 1ed9ccd..a2a5be3 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
@@ -52,6 +52,8 @@ public class DistCpOptions {
private EnumSet<FileAttribute> preserveStatus = EnumSet.noneOf(FileAttribute.class);
+ private boolean preserveRawXattrs;
+
private Path atomicWorkPath;
private Path logPath;
@@ -123,6 +125,7 @@ public class DistCpOptions {
this.sslConfigurationFile = that.getSslConfigurationFile();
this.copyStrategy = that.copyStrategy;
this.preserveStatus = that.preserveStatus;
+ this.preserveRawXattrs = that.preserveRawXattrs;
this.atomicWorkPath = that.getAtomicWorkPath();
this.logPath = that.getLogPath();
this.sourceFileListing = that.getSourceFileListing();
@@ -345,7 +348,7 @@ public class DistCpOptions {
}
/**
- * Checks if the input attibute should be preserved or not
+ * Checks if the input attribute should be preserved or not
*
* @param attribute - Attribute to check
* @return True if attribute should be preserved, false otherwise
@@ -369,6 +372,21 @@ public class DistCpOptions {
preserveStatus.add(fileAttribute);
}
+ /**
+ * Return true if raw.* xattrs should be preserved.
+ * @return true if raw.* xattrs should be preserved.
+ */
+ public boolean shouldPreserveRawXattrs() {
+ return preserveRawXattrs;
+ }
+
+ /**
+ * Indicate that raw.* xattrs should be preserved
+ */
+ public void preserveRawXattrs() {
+ preserveRawXattrs = true;
+ }
+
/** Get work path for atomic commit. If null, the work
* path would be parentOf(targetPath) + "/._WIP_" + nameOf(targetPath)
*
@@ -565,6 +583,7 @@ public class DistCpOptions {
", sourcePaths=" + sourcePaths +
", targetPath=" + targetPath +
", targetPathExists=" + targetPathExists +
+ ", preserveRawXattrs=" + preserveRawXattrs +
'}';
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/SimpleCopyListing.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/SimpleCopyListing.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/SimpleCopyListing.java
index ad29942..f9cfc86 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/SimpleCopyListing.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/SimpleCopyListing.java
@@ -37,6 +37,9 @@ import com.google.common.annotations.VisibleForTesting;
import java.io.*;
import java.util.Stack;
+import static org.apache.hadoop.tools.DistCpConstants
+ .HDFS_RESERVED_RAW_DIRECTORY_NAME;
+
/**
* The SimpleCopyListing is responsible for making the exhaustive list of
* all files/directories under its specified list of input-paths.
@@ -67,6 +70,10 @@ public class SimpleCopyListing extends CopyListing {
Path targetPath = options.getTargetPath();
FileSystem targetFS = targetPath.getFileSystem(getConf());
boolean targetIsFile = targetFS.isFile(targetPath);
+ targetPath = targetFS.makeQualified(targetPath);
+ final boolean targetIsReservedRaw =
+ Path.getPathWithoutSchemeAndAuthority(targetPath).toString().
+ startsWith(HDFS_RESERVED_RAW_DIRECTORY_NAME);
//If target is a file, then source has to be single file
if (targetIsFile) {
@@ -93,6 +100,27 @@ public class SimpleCopyListing extends CopyListing {
if (!fs.exists(path)) {
throw new InvalidInputException(path + " doesn't exist");
}
+ if (Path.getPathWithoutSchemeAndAuthority(path).toString().
+ startsWith(HDFS_RESERVED_RAW_DIRECTORY_NAME)) {
+ if (!targetIsReservedRaw) {
+ final String msg = "The source path '" + path + "' starts with " +
+ HDFS_RESERVED_RAW_DIRECTORY_NAME + " but the target path '" +
+ targetPath + "' does not. Either all or none of the paths must " +
+ "have this prefix.";
+ throw new InvalidInputException(msg);
+ }
+ } else if (targetIsReservedRaw) {
+ final String msg = "The target path '" + targetPath + "' starts with " +
+ HDFS_RESERVED_RAW_DIRECTORY_NAME + " but the source path '" +
+ path + "' does not. Either all or none of the paths must " +
+ "have this prefix.";
+ throw new InvalidInputException(msg);
+ }
+ }
+
+ if (targetIsReservedRaw) {
+ options.preserveRawXattrs();
+ getConf().setBoolean(DistCpConstants.CONF_LABEL_PRESERVE_RAWXATTRS, true);
}
/* This is requires to allow map tasks to access each of the source
@@ -135,6 +163,9 @@ public class SimpleCopyListing extends CopyListing {
try {
for (Path path: options.getSourcePaths()) {
FileSystem sourceFS = path.getFileSystem(getConf());
+ final boolean preserveAcls = options.shouldPreserve(FileAttribute.ACL);
+ final boolean preserveXAttrs = options.shouldPreserve(FileAttribute.XATTR);
+ final boolean preserveRawXAttrs = options.shouldPreserveRawXattrs();
path = makeQualified(path);
FileStatus rootStatus = sourceFS.getFileStatus(path);
@@ -145,8 +176,7 @@ public class SimpleCopyListing extends CopyListing {
if (!explore || rootStatus.isDirectory()) {
CopyListingFileStatus rootCopyListingStatus =
DistCpUtils.toCopyListingFileStatus(sourceFS, rootStatus,
- options.shouldPreserve(FileAttribute.ACL),
- options.shouldPreserve(FileAttribute.XATTR));
+ preserveAcls, preserveXAttrs, preserveRawXAttrs);
writeToFileListingRoot(fileListWriter, rootCopyListingStatus,
sourcePathRoot, options);
}
@@ -157,9 +187,9 @@ public class SimpleCopyListing extends CopyListing {
}
CopyListingFileStatus sourceCopyListingStatus =
DistCpUtils.toCopyListingFileStatus(sourceFS, sourceStatus,
- options.shouldPreserve(FileAttribute.ACL) &&
- sourceStatus.isDirectory(), options.shouldPreserve(
- FileAttribute.XATTR) && sourceStatus.isDirectory());
+ preserveAcls && sourceStatus.isDirectory(),
+ preserveXAttrs && sourceStatus.isDirectory(),
+ preserveRawXAttrs && sourceStatus.isDirectory());
writeToFileListing(fileListWriter, sourceCopyListingStatus,
sourcePathRoot, options);
@@ -261,6 +291,9 @@ public class SimpleCopyListing extends CopyListing {
DistCpOptions options)
throws IOException {
FileSystem sourceFS = sourcePathRoot.getFileSystem(getConf());
+ final boolean preserveAcls = options.shouldPreserve(FileAttribute.ACL);
+ final boolean preserveXAttrs = options.shouldPreserve(FileAttribute.XATTR);
+ final boolean preserveRawXattrs = options.shouldPreserveRawXattrs();
Stack<FileStatus> pathStack = new Stack<FileStatus>();
pathStack.push(sourceStatus);
@@ -271,8 +304,9 @@ public class SimpleCopyListing extends CopyListing {
+ sourceStatus.getPath() + " for copy.");
CopyListingFileStatus childCopyListingStatus =
DistCpUtils.toCopyListingFileStatus(sourceFS, child,
- options.shouldPreserve(FileAttribute.ACL) && child.isDirectory(),
- options.shouldPreserve(FileAttribute.XATTR) && child.isDirectory());
+ preserveAcls && child.isDirectory(),
+ preserveXAttrs && child.isDirectory(),
+ preserveRawXattrs && child.isDirectory());
writeToFileListing(fileListWriter, childCopyListingStatus,
sourcePathRoot, options);
if (isDirectoryAndNotEmpty(sourceFS, child)) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/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 4d16445..197edd9 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
@@ -83,7 +83,9 @@ public class CopyCommitter extends FileOutputCommitter {
cleanupTempFiles(jobContext);
String attributes = conf.get(DistCpConstants.CONF_LABEL_PRESERVE_STATUS);
- if (attributes != null && !attributes.isEmpty()) {
+ final boolean preserveRawXattrs =
+ conf.getBoolean(DistCpConstants.CONF_LABEL_PRESERVE_RAWXATTRS, false);
+ if ((attributes != null && !attributes.isEmpty()) || preserveRawXattrs) {
preserveFileAttributesForDirectories(conf);
}
@@ -167,6 +169,8 @@ public class CopyCommitter extends FileOutputCommitter {
LOG.info("About to preserve attributes: " + attrSymbols);
EnumSet<FileAttribute> attributes = DistCpUtils.unpackAttributes(attrSymbols);
+ final boolean preserveRawXattrs =
+ conf.getBoolean(DistCpConstants.CONF_LABEL_PRESERVE_RAWXATTRS, false);
Path sourceListing = new Path(conf.get(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH));
FileSystem clusterFS = sourceListing.getFileSystem(conf);
@@ -194,7 +198,8 @@ public class CopyCommitter extends FileOutputCommitter {
if (targetRoot.equals(targetFile) && syncOrOverwrite) continue;
FileSystem targetFS = targetFile.getFileSystem(conf);
- DistCpUtils.preserve(targetFS, targetFile, srcFileStatus, attributes);
+ DistCpUtils.preserve(targetFS, targetFile, srcFileStatus, attributes,
+ preserveRawXattrs);
taskAttemptContext.progress();
taskAttemptContext.setStatus("Preserving status on directory entries. [" +
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java
index 4ee003d..ab57127 100644
--- a/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java
+++ b/hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyMapper.java
@@ -200,6 +200,8 @@ public class CopyMapper extends Mapper<Text, CopyListingFileStatus, Text, Text>
EnumSet<DistCpOptions.FileAttribute> fileAttributes
= getFileAttributeSettings(context);
+ final boolean preserveRawXattrs = context.getConfiguration().getBoolean(
+ DistCpConstants.CONF_LABEL_PRESERVE_RAWXATTRS, false);
final String description = "Copying " + sourcePath + " to " + target;
context.setStatus(description);
@@ -211,10 +213,12 @@ public class CopyMapper extends Mapper<Text, CopyListingFileStatus, Text, Text>
FileSystem sourceFS;
try {
sourceFS = sourcePath.getFileSystem(conf);
+ final boolean preserveXAttrs =
+ fileAttributes.contains(FileAttribute.XATTR);
sourceCurrStatus = DistCpUtils.toCopyListingFileStatus(sourceFS,
sourceFS.getFileStatus(sourcePath),
fileAttributes.contains(FileAttribute.ACL),
- fileAttributes.contains(FileAttribute.XATTR));
+ preserveXAttrs, preserveRawXattrs);
} catch (FileNotFoundException e) {
throw new IOException(new RetriableFileCopyCommand.CopyReadException(e));
}
@@ -249,8 +253,8 @@ public class CopyMapper extends Mapper<Text, CopyListingFileStatus, Text, Text>
action, fileAttributes);
}
- DistCpUtils.preserve(target.getFileSystem(conf), target,
- sourceCurrStatus, fileAttributes);
+ DistCpUtils.preserve(target.getFileSystem(conf), target, sourceCurrStatus,
+ fileAttributes, preserveRawXattrs);
} catch (IOException exception) {
handleFailures(exception, sourceFileStatus, target, context);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/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 c7b29a1..abd30ee 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
@@ -18,6 +18,7 @@
package org.apache.hadoop.tools.util;
+import com.google.common.collect.Maps;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -25,6 +26,7 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileChecksum;
+import org.apache.hadoop.fs.XAttr;
import org.apache.hadoop.fs.permission.AclEntry;
import org.apache.hadoop.fs.permission.AclUtil;
import org.apache.hadoop.fs.permission.FsPermission;
@@ -151,7 +153,7 @@ public class DistCpUtils {
* @return - String containing first letters of each attribute to preserve
*/
public static String packAttributes(EnumSet<FileAttribute> attributes) {
- StringBuffer buffer = new StringBuffer(5);
+ StringBuffer buffer = new StringBuffer(FileAttribute.values().length);
int len = 0;
for (FileAttribute attribute : attributes) {
buffer.append(attribute.name().charAt(0));
@@ -186,13 +188,15 @@ public class DistCpUtils {
* @param targetFS - File system
* @param path - Path that needs to preserve original file status
* @param srcFileStatus - Original file status
- * @param attributes - Attribute set that need to be preserved
+ * @param attributes - Attribute set that needs to be preserved
+ * @param preserveRawXattrs if true, raw.* xattrs should be preserved
* @throws IOException - Exception if any (particularly relating to group/owner
* change or any transient error)
*/
public static void preserve(FileSystem targetFS, Path path,
CopyListingFileStatus srcFileStatus,
- EnumSet<FileAttribute> attributes) throws IOException {
+ EnumSet<FileAttribute> attributes,
+ boolean preserveRawXattrs) throws IOException {
FileStatus targetFileStatus = targetFS.getFileStatus(path);
String group = targetFileStatus.getGroup();
@@ -214,15 +218,20 @@ public class DistCpUtils {
!srcFileStatus.getPermission().equals(targetFileStatus.getPermission())) {
targetFS.setPermission(path, srcFileStatus.getPermission());
}
-
- if (attributes.contains(FileAttribute.XATTR)) {
+
+ final boolean preserveXAttrs = attributes.contains(FileAttribute.XATTR);
+ if (preserveXAttrs || preserveRawXattrs) {
+ final String rawNS = XAttr.NameSpace.RAW.name().toLowerCase();
Map<String, byte[]> srcXAttrs = srcFileStatus.getXAttrs();
Map<String, byte[]> targetXAttrs = getXAttrs(targetFS, path);
- if (!srcXAttrs.equals(targetXAttrs)) {
+ if (srcXAttrs != null && !srcXAttrs.equals(targetXAttrs)) {
Iterator<Entry<String, byte[]>> iter = srcXAttrs.entrySet().iterator();
while (iter.hasNext()) {
Entry<String, byte[]> entry = iter.next();
- targetFS.setXAttr(path, entry.getKey(), entry.getValue());
+ final String xattrName = entry.getKey();
+ if (xattrName.startsWith(rawNS) || preserveXAttrs) {
+ targetFS.setXAttr(path, entry.getKey(), entry.getValue());
+ }
}
}
}
@@ -286,11 +295,12 @@ public class DistCpUtils {
* @param fileStatus FileStatus of file
* @param preserveAcls boolean true if preserving ACLs
* @param preserveXAttrs boolean true if preserving XAttrs
+ * @param preserveRawXAttrs boolean true if preserving raw.* XAttrs
* @throws IOException if there is an I/O error
*/
public static CopyListingFileStatus toCopyListingFileStatus(
FileSystem fileSystem, FileStatus fileStatus, boolean preserveAcls,
- boolean preserveXAttrs) throws IOException {
+ boolean preserveXAttrs, boolean preserveRawXAttrs) throws IOException {
CopyListingFileStatus copyListingFileStatus =
new CopyListingFileStatus(fileStatus);
if (preserveAcls) {
@@ -301,9 +311,25 @@ public class DistCpUtils {
copyListingFileStatus.setAclEntries(aclEntries);
}
}
- if (preserveXAttrs) {
- Map<String, byte[]> xAttrs = fileSystem.getXAttrs(fileStatus.getPath());
- copyListingFileStatus.setXAttrs(xAttrs);
+ if (preserveXAttrs || preserveRawXAttrs) {
+ Map<String, byte[]> srcXAttrs = fileSystem.getXAttrs(fileStatus.getPath());
+ if (preserveXAttrs && preserveRawXAttrs) {
+ copyListingFileStatus.setXAttrs(srcXAttrs);
+ } else {
+ Map<String, byte[]> trgXAttrs = Maps.newHashMap();
+ final String rawNS = XAttr.NameSpace.RAW.name().toLowerCase();
+ for (Map.Entry<String, byte[]> ent : srcXAttrs.entrySet()) {
+ final String xattrName = ent.getKey();
+ if (xattrName.startsWith(rawNS)) {
+ if (preserveRawXAttrs) {
+ trgXAttrs.put(xattrName, ent.getValue());
+ }
+ } else if (preserveXAttrs) {
+ trgXAttrs.put(xattrName, ent.getValue());
+ }
+ }
+ copyListingFileStatus.setXAttrs(trgXAttrs);
+ }
}
return copyListingFileStatus;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java
new file mode 100644
index 0000000..5aef51a
--- /dev/null
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithRawXAttrs.java
@@ -0,0 +1,170 @@
+/**
+ * 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;
+
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.tools.util.DistCpTestUtils;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Tests distcp in combination with HDFS raw.* XAttrs.
+ */
+public class TestDistCpWithRawXAttrs {
+
+ private static MiniDFSCluster cluster;
+ private static Configuration conf;
+ private static FileSystem fs;
+
+ private static final String rawName1 = "raw.a1";
+ private static final byte[] rawValue1 = {0x37, 0x38, 0x39};
+ private static final String userName1 = "user.a1";
+ private static final byte[] userValue1 = {0x38, 0x38, 0x38};
+
+ private static final Path dir1 = new Path("/src/dir1");
+ private static final Path subDir1 = new Path(dir1, "subdir1");
+ private static final Path file1 = new Path("/src/file1");
+ private static final String rawRootName = "/.reserved/raw";
+ private static final String rootedDestName = "/dest";
+ private static final String rootedSrcName = "/src";
+ private static final String rawDestName = "/.reserved/raw/dest";
+ private static final String rawSrcName = "/.reserved/raw/src";
+
+ @BeforeClass
+ public static void init() throws Exception {
+ conf = new Configuration();
+ conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_XATTRS_ENABLED_KEY, true);
+ cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).format(true)
+ .build();
+ cluster.waitActive();
+ fs = cluster.getFileSystem();
+ }
+
+ @AfterClass
+ public static void shutdown() {
+ IOUtils.cleanup(null, fs);
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ /* Test that XAttrs and raw.* XAttrs are preserved when appropriate. */
+ @Test
+ public void testPreserveRawXAttrs1() throws Exception {
+ final String relSrc = "/./.reserved/../.reserved/raw/../raw/src/../src";
+ final String relDst = "/./.reserved/../.reserved/raw/../raw/dest/../dest";
+ doTestPreserveRawXAttrs(relSrc, relDst, "-px", true, true,
+ DistCpConstants.SUCCESS);
+ doTestPreserveRawXAttrs(rootedSrcName, rootedDestName, "-px",
+ false, true, DistCpConstants.SUCCESS);
+ doTestPreserveRawXAttrs(rootedSrcName, rawDestName, "-px",
+ false, true, DistCpConstants.INVALID_ARGUMENT);
+ doTestPreserveRawXAttrs(rawSrcName, rootedDestName, "-px",
+ false, true, DistCpConstants.INVALID_ARGUMENT);
+ doTestPreserveRawXAttrs(rawSrcName, rawDestName, "-px",
+ true, true, DistCpConstants.SUCCESS);
+ final Path savedWd = fs.getWorkingDirectory();
+ try {
+ fs.setWorkingDirectory(new Path("/.reserved/raw"));
+ doTestPreserveRawXAttrs("../.." + rawSrcName, "../.." + rawDestName,
+ "-px", true, true, DistCpConstants.SUCCESS);
+ } finally {
+ fs.setWorkingDirectory(savedWd);
+ }
+ }
+
+ /* Test that XAttrs are not preserved and raw.* are when appropriate. */
+ @Test
+ public void testPreserveRawXAttrs2() throws Exception {
+ doTestPreserveRawXAttrs(rootedSrcName, rootedDestName, "-p",
+ false, false, DistCpConstants.SUCCESS);
+ doTestPreserveRawXAttrs(rootedSrcName, rawDestName, "-p",
+ false, false, DistCpConstants.INVALID_ARGUMENT);
+ doTestPreserveRawXAttrs(rawSrcName, rootedDestName, "-p",
+ false, false, DistCpConstants.INVALID_ARGUMENT);
+ doTestPreserveRawXAttrs(rawSrcName, rawDestName, "-p",
+ true, false, DistCpConstants.SUCCESS);
+ }
+
+ /* Test that XAttrs are not preserved and raw.* are when appropriate. */
+ @Test
+ public void testPreserveRawXAttrs3() throws Exception {
+ doTestPreserveRawXAttrs(rootedSrcName, rootedDestName, null,
+ false, false, DistCpConstants.SUCCESS);
+ doTestPreserveRawXAttrs(rootedSrcName, rawDestName, null,
+ false, false, DistCpConstants.INVALID_ARGUMENT);
+ doTestPreserveRawXAttrs(rawSrcName, rootedDestName, null,
+ false, false, DistCpConstants.INVALID_ARGUMENT);
+ doTestPreserveRawXAttrs(rawSrcName, rawDestName, null,
+ true, false, DistCpConstants.SUCCESS);
+ }
+
+ private static Path[] pathnames = { new Path("dir1"),
+ new Path("dir1/subdir1"),
+ new Path("file1") };
+
+ private static void makeFilesAndDirs(FileSystem fs) throws Exception {
+ fs.delete(new Path("/src"), true);
+ fs.delete(new Path("/dest"), true);
+ fs.mkdirs(subDir1);
+ fs.create(file1).close();
+ }
+
+ private void initXAttrs() throws Exception {
+ makeFilesAndDirs(fs);
+ for (Path p : pathnames) {
+ fs.setXAttr(new Path(rawRootName + "/src", p), rawName1, rawValue1);
+ fs.setXAttr(new Path(rawRootName + "/src", p), userName1, userValue1);
+ }
+ }
+
+ private void doTestPreserveRawXAttrs(String src, String dest,
+ String preserveOpts, boolean expectRaw, boolean expectUser,
+ int expectedExitCode) throws Exception {
+ initXAttrs();
+
+ DistCpTestUtils.assertRunDistCp(expectedExitCode, src, dest,
+ preserveOpts, conf);
+
+ if (expectedExitCode == DistCpConstants.SUCCESS) {
+ Map<String, byte[]> xAttrs = Maps.newHashMap();
+ for (Path p : pathnames) {
+ xAttrs.clear();
+ if (expectRaw) {
+ xAttrs.put(rawName1, rawValue1);
+ }
+ if (expectUser) {
+ xAttrs.put(userName1, userValue1);
+ }
+ DistCpTestUtils.assertXAttrs(new Path(dest, p), fs, xAttrs);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithXAttrs.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithXAttrs.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithXAttrs.java
index cc13b8f..6b0e2b22 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithXAttrs.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestDistCpWithXAttrs.java
@@ -18,13 +18,9 @@
package org.apache.hadoop.tools;
-import static org.junit.Assert.*;
-
import java.io.IOException;
import java.net.URI;
-import java.util.Iterator;
import java.util.Map;
-import java.util.Map.Entry;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
@@ -37,8 +33,8 @@ import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.tools.util.DistCpTestUtils;
import org.apache.hadoop.util.Progressable;
-import org.apache.hadoop.util.ToolRunner;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -79,6 +75,7 @@ public class TestDistCpWithXAttrs {
private static final Path dstFile2 = new Path(dstDir2, "file2");
private static final Path dstFile3 = new Path(dstDir2, "file3");
private static final Path dstFile4 = new Path(dstDir2, "file4");
+ private static final String rootedSrcName = "/src";
@BeforeClass
public static void init() throws Exception {
@@ -125,55 +122,56 @@ public class TestDistCpWithXAttrs {
@Test
public void testPreserveXAttrs() throws Exception {
- assertRunDistCp(DistCpConstants.SUCCESS, "/dstPreserveXAttrs");
+ DistCpTestUtils.assertRunDistCp(DistCpConstants.SUCCESS, rootedSrcName,
+ "/dstPreserveXAttrs", "-px", conf);
// dstDir1
Map<String, byte[]> xAttrs = Maps.newHashMap();
xAttrs.put(name1, value1);
xAttrs.put(name2, value2);
- assertXAttrs(dstDir1, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstDir1, fs, xAttrs);
// dstSubDir1
xAttrs.clear();
xAttrs.put(name1, value1);
xAttrs.put(name3, new byte[0]);
- assertXAttrs(dstSubDir1, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstSubDir1, fs, xAttrs);
// dstFile1
xAttrs.clear();
xAttrs.put(name1, value1);
xAttrs.put(name2, value2);
xAttrs.put(name3, new byte[0]);
- assertXAttrs(dstFile1, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstFile1, fs, xAttrs);
// dstDir2
xAttrs.clear();
xAttrs.put(name2, value2);
- assertXAttrs(dstDir2, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstDir2, fs, xAttrs);
// dstFile2
xAttrs.clear();
xAttrs.put(name1, value1);
xAttrs.put(name4, new byte[0]);
- assertXAttrs(dstFile2, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstFile2, fs, xAttrs);
// dstFile3
xAttrs.clear();
xAttrs.put(name3, new byte[0]);
xAttrs.put(name4, new byte[0]);
- assertXAttrs(dstFile3, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstFile3, fs, xAttrs);
// dstFile4
xAttrs.clear();
- assertXAttrs(dstFile4, xAttrs);
+ DistCpTestUtils.assertXAttrs(dstFile4, fs, xAttrs);
}
@Test
public void testXAttrsNotEnabled() throws Exception {
try {
restart(false);
- assertRunDistCp(DistCpConstants.XATTRS_NOT_SUPPORTED,
- "/dstXAttrsNotEnabled");
+ DistCpTestUtils.assertRunDistCp(DistCpConstants.XATTRS_NOT_SUPPORTED,
+ rootedSrcName, "/dstXAttrsNotEnabled", "-px", conf);
} finally {
restart(true);
}
@@ -181,8 +179,8 @@ public class TestDistCpWithXAttrs {
@Test
public void testXAttrsNotImplemented() throws Exception {
- assertRunDistCp(DistCpConstants.XATTRS_NOT_SUPPORTED,
- "stubfs://dstXAttrsNotImplemented");
+ DistCpTestUtils.assertRunDistCp(DistCpConstants.XATTRS_NOT_SUPPORTED,
+ rootedSrcName, "stubfs://dstXAttrsNotImplemented", "-px", conf);
}
/**
@@ -252,45 +250,6 @@ public class TestDistCpWithXAttrs {
}
/**
- * Asserts the XAttrs returned by getXAttrs for a specific path.
- *
- * @param path String path to check
- * @param xAttrs XAttr[] expected xAttrs
- * @throws Exception if there is any error
- */
- private static void assertXAttrs(Path path, Map<String, byte[]> expectedXAttrs)
- throws Exception {
- Map<String, byte[]> xAttrs = fs.getXAttrs(path);
- assertEquals(expectedXAttrs.size(), xAttrs.size());
- Iterator<Entry<String, byte[]>> i = expectedXAttrs.entrySet().iterator();
- while (i.hasNext()) {
- Entry<String, byte[]> e = i.next();
- String name = e.getKey();
- byte[] value = e.getValue();
- if (value == null) {
- assertTrue(xAttrs.containsKey(name) && xAttrs.get(name) == null);
- } else {
- assertArrayEquals(value, xAttrs.get(name));
- }
- }
- }
-
- /**
- * Runs distcp from /src to specified destination, preserving XAttrs. Asserts
- * expected exit code.
- *
- * @param int exitCode expected exit code
- * @param dst String distcp destination
- * @throws Exception if there is any error
- */
- private static void assertRunDistCp(int exitCode, String dst)
- throws Exception {
- DistCp distCp = new DistCp(conf, null);
- assertEquals(exitCode,
- ToolRunner.run(conf, distCp, new String[] { "-px", "/src", dst }));
- }
-
- /**
* Initialize the cluster, wait for it to become active, and get FileSystem.
*
* @param format if true, format the NameNode and DataNodes before starting up
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestOptionsParser.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestOptionsParser.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestOptionsParser.java
index d3da68e..2c42269 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestOptionsParser.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestOptionsParser.java
@@ -357,7 +357,8 @@ public class TestOptionsParser {
DistCpOptions option = new DistCpOptions(new Path("abc"), new Path("xyz"));
String val = "DistCpOptions{atomicCommit=false, syncFolder=false, deleteMissing=false, " +
"ignoreFailures=false, maxMaps=20, sslConfigurationFile='null', copyStrategy='uniformsize', " +
- "sourceFileListing=abc, sourcePaths=null, targetPath=xyz, targetPathExists=true}";
+ "sourceFileListing=abc, sourcePaths=null, targetPath=xyz, targetPathExists=true, " +
+ "preserveRawXattrs=false}";
Assert.assertEquals(val, option.toString());
Assert.assertNotSame(DistCpOptionSwitch.ATOMIC_COMMIT.toString(),
DistCpOptionSwitch.ATOMIC_COMMIT.name());
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java
new file mode 100644
index 0000000..2721638
--- /dev/null
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java
@@ -0,0 +1,89 @@
+/**
+ * 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.util;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.hadoop.tools.DistCp;
+import org.apache.hadoop.util.ToolRunner;
+
+/**
+ * Utility class for DistCpTests
+ */
+public class DistCpTestUtils {
+
+ /**
+ * Asserts the XAttrs returned by getXAttrs for a specific path match an
+ * expected set of XAttrs.
+ *
+ * @param path String path to check
+ * @param fs FileSystem to use for the path
+ * @param expectedXAttrs XAttr[] expected xAttrs
+ * @throws Exception if there is any error
+ */
+ public static void assertXAttrs(Path path, FileSystem fs,
+ Map<String, byte[]> expectedXAttrs)
+ throws Exception {
+ Map<String, byte[]> xAttrs = fs.getXAttrs(path);
+ assertEquals(path.toString(), expectedXAttrs.size(), xAttrs.size());
+ Iterator<Entry<String, byte[]>> i = expectedXAttrs.entrySet().iterator();
+ while (i.hasNext()) {
+ Entry<String, byte[]> e = i.next();
+ String name = e.getKey();
+ byte[] value = e.getValue();
+ if (value == null) {
+ assertTrue(xAttrs.containsKey(name) && xAttrs.get(name) == null);
+ } else {
+ assertArrayEquals(value, xAttrs.get(name));
+ }
+ }
+ }
+
+ /**
+ * Runs distcp from src to dst, preserving XAttrs. Asserts the
+ * expected exit code.
+ *
+ * @param exitCode expected exit code
+ * @param src distcp src path
+ * @param dst distcp destination
+ * @param options distcp command line options
+ * @param conf Configuration to use
+ * @throws Exception if there is any error
+ */
+ public static void assertRunDistCp(int exitCode, String src, String dst,
+ String options, Configuration conf)
+ throws Exception {
+ DistCp distCp = new DistCp(conf, null);
+ String[] optsArr = options == null ?
+ new String[] { src, dst } :
+ new String[] { options, src, dst };
+ assertEquals(exitCode,
+ ToolRunner.run(conf, distCp, optsArr));
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/c77bd85b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/TestDistCpUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/TestDistCpUtils.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/TestDistCpUtils.java
index 4825e15..c4b64b9 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/TestDistCpUtils.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/TestDistCpUtils.java
@@ -114,14 +114,14 @@ public class TestDistCpUtils {
fs.setPermission(path, noPerm);
fs.setOwner(path, "nobody", "nobody");
- DistCpUtils.preserve(fs, path, srcStatus, attributes);
+ DistCpUtils.preserve(fs, path, srcStatus, attributes, false);
FileStatus target = fs.getFileStatus(path);
Assert.assertEquals(target.getPermission(), noPerm);
Assert.assertEquals(target.getOwner(), "nobody");
Assert.assertEquals(target.getGroup(), "nobody");
attributes.add(FileAttribute.PERMISSION);
- DistCpUtils.preserve(fs, path, srcStatus, attributes);
+ DistCpUtils.preserve(fs, path, srcStatus, attributes, false);
target = fs.getFileStatus(path);
Assert.assertEquals(target.getPermission(), srcStatus.getPermission());
Assert.assertEquals(target.getOwner(), "nobody");
@@ -129,7 +129,7 @@ public class TestDistCpUtils {
attributes.add(FileAttribute.GROUP);
attributes.add(FileAttribute.USER);
- DistCpUtils.preserve(fs, path, srcStatus, attributes);
+ DistCpUtils.preserve(fs, path, srcStatus, attributes, false);
target = fs.getFileStatus(path);
Assert.assertEquals(target.getPermission(), srcStatus.getPermission());
Assert.assertEquals(target.getOwner(), srcStatus.getOwner());