You are viewing a plain text version of this content. The canonical link for it is here.
Posted to mapreduce-commits@hadoop.apache.org by sz...@apache.org on 2012/10/19 04:28:42 UTC
svn commit: r1399950 [4/11] - in
/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project: ./ bin/ conf/
dev-support/ hadoop-mapreduce-client/
hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/
hadoop-mapreduc...
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/AbstractCounters.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/AbstractCounters.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/AbstractCounters.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/AbstractCounters.java Fri Oct 19 02:25:55 2012
@@ -24,6 +24,7 @@ import static org.apache.hadoop.mapreduc
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
@@ -81,7 +82,7 @@ public abstract class AbstractCounters<C
TaskCounter.class.getName());
legacyMap.put("org.apache.hadoop.mapred.JobInProgress$Counter",
JobCounter.class.getName());
- legacyMap.put("FileSystemCounter", FileSystemCounter.class.getName());
+ legacyMap.put("FileSystemCounters", FileSystemCounter.class.getName());
}
private final Limits limits = new Limits();
@@ -185,7 +186,15 @@ public abstract class AbstractCounters<C
* @return Set of counter names.
*/
public synchronized Iterable<String> getGroupNames() {
- return Iterables.concat(fgroups.keySet(), groups.keySet());
+ HashSet<String> deprecated = new HashSet<String>();
+ for(Map.Entry<String, String> entry : legacyMap.entrySet()) {
+ String newGroup = entry.getValue();
+ boolean isFGroup = isFrameworkGroup(newGroup);
+ if(isFGroup ? fgroups.containsKey(newGroup) : groups.containsKey(newGroup)) {
+ deprecated.add(entry.getKey());
+ }
+ }
+ return Iterables.concat(fgroups.keySet(), groups.keySet(), deprecated);
}
@Override
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/Limits.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/Limits.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/Limits.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/counters/Limits.java Fri Oct 19 02:25:55 2012
@@ -20,12 +20,14 @@ package org.apache.hadoop.mapreduce.coun
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
+
import static org.apache.hadoop.mapreduce.MRJobConfig.*;
@InterfaceAudience.Private
public class Limits {
- static final Configuration conf = new Configuration();
+ static final Configuration conf = new JobConf();
public static final int GROUP_NAME_MAX =
conf.getInt(COUNTER_GROUP_NAME_MAX_KEY, COUNTER_GROUP_NAME_MAX_DEFAULT);
public static final int COUNTER_NAME_MAX =
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/filecache/DistributedCache.java Fri Oct 19 02:25:55 2012
@@ -55,8 +55,12 @@ import java.net.URI;
* Archives (zip, tar and tgz/tar.gz files) are un-archived at the slave nodes.
* Jars may be optionally added to the classpath of the tasks, a rudimentary
* software distribution mechanism. Files have execution permissions.
- * Optionally users can also direct it to symlink the distributed cache file(s)
- * into the working directory of the task.</p>
+ * In older version of Hadoop Map/Reduce users could optionally ask for symlinks
+ * to be created in the working directory of the child task. In the current
+ * version symlinks are always created. If the URL does not have a fragment
+ * the name of the file or directory will be used. If multiple files or
+ * directories map to the same link name, the last one added, will be used. All
+ * others will not even be downloaded.</p>
*
* <p><code>DistributedCache</code> tracks modification timestamps of the cache
* files. Clearly the cache files should not be modified by the application
@@ -98,8 +102,7 @@ import java.net.URI;
*
* public void configure(JobConf job) {
* // Get the cached archives/files
- * localArchives = DistributedCache.getLocalCacheArchives(job);
- * localFiles = DistributedCache.getLocalCacheFiles(job);
+ * File f = new File("./map.zip/some/file/in/zip.txt");
* }
*
* public void map(K key, V value,
@@ -375,32 +378,26 @@ public class DistributedCache {
}
/**
- * This method allows you to create symlinks in the current working directory
- * of the task to all the cache files/archives.
- * Intended to be used by user code.
+ * Originally intended to enable symlinks, but currently symlinks cannot be
+ * disabled. This is a NO-OP.
* @param conf the jobconf
- * @deprecated Use {@link Job#createSymlink()} instead
+ * @deprecated This is a NO-OP.
*/
@Deprecated
public static void createSymlink(Configuration conf){
- conf.set(MRJobConfig.CACHE_SYMLINK, "yes");
+ //NOOP
}
/**
- * This method checks to see if symlinks are to be create for the
- * localized cache files in the current working directory
- * Used by internal DistributedCache code.
+ * Originally intended to check if symlinks should be used, but currently
+ * symlinks cannot be disabled.
* @param conf the jobconf
- * @return true if symlinks are to be created- else return false
- * @deprecated Use {@link JobContext#getSymlink()} instead
+ * @return true
+ * @deprecated symlinks are always created.
*/
@Deprecated
public static boolean getSymlink(Configuration conf){
- String result = conf.get(MRJobConfig.CACHE_SYMLINK);
- if ("yes".equals(result)){
- return true;
- }
- return false;
+ return true;
}
private static boolean[] parseBooleans(String[] strs) {
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryParser.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryParser.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryParser.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryParser.java Fri Oct 19 02:25:55 2012
@@ -441,6 +441,7 @@ public class JobHistoryParser {
username = jobname = jobConfPath = jobQueueName = "";
tasksMap = new HashMap<TaskID, TaskInfo>();
jobACLs = new HashMap<JobACL, AccessControlList>();
+ priority = JobPriority.NORMAL;
}
/** Print all the job information */
@@ -454,12 +455,20 @@ public class JobHistoryParser {
System.out.println("PRIORITY: " + priority);
System.out.println("TOTAL_MAPS: " + totalMaps);
System.out.println("TOTAL_REDUCES: " + totalReduces);
- System.out.println("MAP_COUNTERS:" + mapCounters.toString());
- System.out.println("REDUCE_COUNTERS:" + reduceCounters.toString());
- System.out.println("TOTAL_COUNTERS: " + totalCounters.toString());
+ if (mapCounters != null) {
+ System.out.println("MAP_COUNTERS:" + mapCounters.toString());
+ }
+ if (reduceCounters != null) {
+ System.out.println("REDUCE_COUNTERS:" + reduceCounters.toString());
+ }
+ if (totalCounters != null) {
+ System.out.println("TOTAL_COUNTERS: " + totalCounters.toString());
+ }
System.out.println("UBERIZED: " + uberized);
- for (AMInfo amInfo : amInfos) {
- amInfo.printAll();
+ if (amInfos != null) {
+ for (AMInfo amInfo : amInfos) {
+ amInfo.printAll();
+ }
}
for (TaskInfo ti: tasksMap.values()) {
ti.printAll();
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java Fri Oct 19 02:25:55 2012
@@ -497,6 +497,11 @@ public abstract class CombineFileInputFo
if (locations == null) {
blocks = new OneBlockInfo[0];
} else {
+
+ if(locations.length == 0) {
+ locations = new BlockLocation[] { new BlockLocation() };
+ }
+
if (!isSplitable) {
// if the file is not splitable, just create the one block with
// full file length
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/LazyOutputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/LazyOutputFormat.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/LazyOutputFormat.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/LazyOutputFormat.java Fri Oct 19 02:25:55 2012
@@ -32,7 +32,10 @@ import org.apache.hadoop.mapreduce.TaskA
import org.apache.hadoop.util.ReflectionUtils;
/**
- * A Convenience class that creates output lazily.
+ * A Convenience class that creates output lazily.
+ * Use in conjuction with org.apache.hadoop.mapreduce.lib.output.MultipleOutputs to recreate the
+ * behaviour of org.apache.hadoop.mapred.lib.MultipleTextOutputFormat (etc) of the old Hadoop API.
+ * See {@link MultipleOutputs} documentation for more information.
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.java Fri Oct 19 02:25:55 2012
@@ -20,7 +20,10 @@ package org.apache.hadoop.mapreduce.lib.
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.Reducer.Context;
+import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.apache.hadoop.util.ReflectionUtils;
@@ -37,6 +40,7 @@ import java.util.*;
* Each additional output, or named output, may be configured with its own
* <code>OutputFormat</code>, with its own key class and with its own value
* class.
+ * </p>
*
* <p>
* Case two: to write data to different files provided by user
@@ -107,6 +111,64 @@ import java.util.*;
*
* }
* </pre>
+ *
+ * <p>
+ * When used in conjuction with org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat,
+ * MultipleOutputs can mimic the behaviour of MultipleTextOutputFormat and MultipleSequenceFileOutputFormat
+ * from the old Hadoop API - ie, output can be written from the Reducer to more than one location.
+ * </p>
+ *
+ * <p>
+ * Use <code>MultipleOutputs.write(KEYOUT key, VALUEOUT value, String baseOutputPath)</code> to write key and
+ * value to a path specified by <code>baseOutputPath</code>, with no need to specify a named output:
+ * </p>
+ *
+ * <pre>
+ * private MultipleOutputs<Text, Text> out;
+ *
+ * public void setup(Context context) {
+ * out = new MultipleOutputs<Text, Text>(context);
+ * ...
+ * }
+ *
+ * public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
+ * for (Text t : values) {
+ * out.write(key, t, generateFileName(<<i>parameter list...</i>>));
+ * }
+ * }
+ *
+ * protected void cleanup(Context context) throws IOException, InterruptedException {
+ * out.close();
+ * }
+ * </pre>
+ *
+ * <p>
+ * Use your own code in <code>generateFileName()</code> to create a custom path to your results.
+ * '/' characters in <code>baseOutputPath</code> will be translated into directory levels in your file system.
+ * Also, append your custom-generated path with "part" or similar, otherwise your output will be -00000, -00001 etc.
+ * No call to <code>context.write()</code> is necessary. See example <code>generateFileName()</code> code below.
+ * </p>
+ *
+ * <pre>
+ * private String generateFileName(Text k) {
+ * // expect Text k in format "Surname|Forename"
+ * String[] kStr = k.toString().split("\\|");
+ *
+ * String sName = kStr[0];
+ * String fName = kStr[1];
+ *
+ * // example for k = Smith|John
+ * // output written to /user/hadoop/path/to/output/Smith/John-r-00000 (etc)
+ * return sName + "/" + fName;
+ * }
+ * </pre>
+ *
+ * <p>
+ * Using MultipleOutputs in this way will still create zero-sized default output, eg part-00000.
+ * To prevent this use <code>LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);</code>
+ * instead of <code>job.setOutputFormatClass(TextOutputFormat.class);</code> in your Hadoop job configuration.
+ * </p>
+ *
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/partition/TotalOrderPartitioner.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/partition/TotalOrderPartitioner.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/partition/TotalOrderPartitioner.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/partition/TotalOrderPartitioner.java Fri Oct 19 02:25:55 2012
@@ -47,7 +47,7 @@ import org.apache.hadoop.util.Reflection
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
-public class TotalOrderPartitioner<K extends WritableComparable<?>,V>
+public class TotalOrderPartitioner<K,V>
extends Partitioner<K,V> implements Configurable {
private Node partitions;
@@ -298,12 +298,13 @@ public class TotalOrderPartitioner<K ext
@SuppressWarnings("unchecked") // map output key class
private K[] readPartitions(FileSystem fs, Path p, Class<K> keyClass,
Configuration conf) throws IOException {
- SequenceFile.Reader reader = new SequenceFile.Reader(fs, p, conf);
+ SequenceFile.Reader reader = new SequenceFile.Reader(
+ conf,
+ SequenceFile.Reader.file(p));
ArrayList<K> parts = new ArrayList<K>();
K key = ReflectionUtils.newInstance(keyClass, conf);
- NullWritable value = NullWritable.get();
try {
- while (reader.next(key, value)) {
+ while ((key = (K) reader.next(key)) != null) {
parts.add(key);
key = ReflectionUtils.newInstance(keyClass, conf);
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/security/TokenCache.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/security/TokenCache.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/security/TokenCache.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/security/TokenCache.java Fri Oct 19 02:25:55 2012
@@ -20,7 +20,6 @@ package org.apache.hadoop.mapreduce.secu
import java.io.IOException;
import java.util.HashSet;
-import java.util.List;
import java.util.Set;
import org.apache.commons.logging.Log;
@@ -110,7 +109,6 @@ public class TokenCache {
* @param conf
* @throws IOException
*/
- @SuppressWarnings("deprecation")
static void obtainTokensForNamenodesInternal(FileSystem fs,
Credentials credentials, Configuration conf) throws IOException {
String delegTokenRenewer = Master.getMasterPrincipal(conf);
@@ -120,26 +118,11 @@ public class TokenCache {
}
mergeBinaryTokens(credentials, conf);
- String fsName = fs.getCanonicalServiceName();
- if (TokenCache.getDelegationToken(credentials, fsName) == null) {
- List<Token<?>> tokens =
- fs.getDelegationTokens(delegTokenRenewer, credentials);
- if (tokens != null) {
- for (Token<?> token : tokens) {
- credentials.addToken(token.getService(), token);
- LOG.info("Got dt for " + fs.getUri() + ";uri="+ fsName +
- ";t.service="+token.getService());
- }
- }
- //Call getDelegationToken as well for now - for FS implementations
- // which may not have implmented getDelegationTokens (hftp)
- if (tokens == null || tokens.size() == 0) {
- Token<?> token = fs.getDelegationToken(delegTokenRenewer);
- if (token != null) {
- credentials.addToken(token.getService(), token);
- LOG.info("Got dt for " + fs.getUri() + ";uri=" + fsName
- + ";t.service=" + token.getService());
- }
+ final Token<?> tokens[] = fs.addDelegationTokens(delegTokenRenewer,
+ credentials);
+ if (tokens != null) {
+ for (Token<?> token : tokens) {
+ LOG.info("Got dt for " + fs.getUri() + "; "+token);
}
}
}
@@ -174,21 +157,6 @@ public class TokenCache {
private static final Text JOB_TOKEN = new Text("ShuffleAndJobToken");
/**
- *
- * @param namenode
- * @return delegation token
- */
- @InterfaceAudience.Private
- public static Token<?> getDelegationToken(
- Credentials credentials, String namenode) {
- //No fs specific tokens issues by this fs. It may however issue tokens
- // for other filesystems - which would be keyed by that filesystems name.
- if (namenode == null)
- return null;
- return (Token<?>) credentials.getToken(new Text(namenode));
- }
-
- /**
* load job token from a file
* @param conf
* @throws IOException
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java Fri Oct 19 02:25:55 2012
@@ -21,15 +21,18 @@ import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
-import java.net.HttpURLConnection;
import java.net.URLConnection;
+import java.security.GeneralSecurityException;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import javax.crypto.SecretKey;
+import javax.net.ssl.HttpsURLConnection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -42,14 +45,17 @@ import org.apache.hadoop.mapred.Counters
import org.apache.hadoop.mapred.IFileInputStream;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
+import org.apache.hadoop.security.ssl.SSLFactory;
import org.apache.hadoop.mapreduce.task.reduce.MapOutput.Type;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;
-@SuppressWarnings({"deprecation"})
+import com.google.common.annotations.VisibleForTesting;
+
class Fetcher<K,V> extends Thread {
private static final Log LOG = LogFactory.getLog(Fetcher.class);
@@ -92,10 +98,16 @@ class Fetcher<K,V> extends Thread {
private volatile boolean stopped = false;
+ private JobConf job;
+
+ private static boolean sslShuffle;
+ private static SSLFactory sslFactory;
+
public Fetcher(JobConf job, TaskAttemptID reduceId,
ShuffleScheduler<K,V> scheduler, MergeManager<K,V> merger,
Reporter reporter, ShuffleClientMetrics metrics,
ExceptionReporter exceptionReporter, SecretKey jobTokenSecret) {
+ this.job = job;
this.reporter = reporter;
this.scheduler = scheduler;
this.merger = merger;
@@ -135,6 +147,20 @@ class Fetcher<K,V> extends Thread {
setName("fetcher#" + id);
setDaemon(true);
+
+ synchronized (Fetcher.class) {
+ sslShuffle = job.getBoolean(MRConfig.SHUFFLE_SSL_ENABLED_KEY,
+ MRConfig.SHUFFLE_SSL_ENABLED_DEFAULT);
+ if (sslShuffle && sslFactory == null) {
+ sslFactory = new SSLFactory(SSLFactory.Mode.CLIENT, job);
+ try {
+ sslFactory.init();
+ } catch (Exception ex) {
+ sslFactory.destroy();
+ throw new RuntimeException(ex);
+ }
+ }
+ }
}
public void run() {
@@ -173,15 +199,34 @@ class Fetcher<K,V> extends Thread {
} catch (InterruptedException ie) {
LOG.warn("Got interrupt while joining " + getName(), ie);
}
+ if (sslFactory != null) {
+ sslFactory.destroy();
+ }
}
+ @VisibleForTesting
+ protected HttpURLConnection openConnection(URL url) throws IOException {
+ HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+ if (sslShuffle) {
+ HttpsURLConnection httpsConn = (HttpsURLConnection) conn;
+ try {
+ httpsConn.setSSLSocketFactory(sslFactory.createSSLSocketFactory());
+ } catch (GeneralSecurityException ex) {
+ throw new IOException(ex);
+ }
+ httpsConn.setHostnameVerifier(sslFactory.getHostnameVerifier());
+ }
+ return conn;
+ }
+
/**
* The crux of the matter...
*
* @param host {@link MapHost} from which we need to
* shuffle available map-outputs.
*/
- private void copyFromHost(MapHost host) throws IOException {
+ @VisibleForTesting
+ protected void copyFromHost(MapHost host) throws IOException {
// Get completed maps on 'host'
List<TaskAttemptID> maps = scheduler.getMapsForHost(host);
@@ -191,9 +236,9 @@ class Fetcher<K,V> extends Thread {
return;
}
- LOG.debug("Fetcher " + id + " going to fetch from " + host);
- for (TaskAttemptID tmp: maps) {
- LOG.debug(tmp);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Fetcher " + id + " going to fetch from " + host + " for: "
+ + maps);
}
// List of maps to be fetched yet
@@ -205,7 +250,7 @@ class Fetcher<K,V> extends Thread {
try {
URL url = getMapOutputURL(host, maps);
- HttpURLConnection connection = (HttpURLConnection)url.openConnection();
+ HttpURLConnection connection = openConnection(url);
// generate hash of the url
String msgToEncode = SecureShuffleUtils.buildMsgFrom(url);
@@ -266,17 +311,25 @@ class Fetcher<K,V> extends Thread {
try {
// Loop through available map-outputs and fetch them
- // On any error, good becomes false and we exit after putting back
- // the remaining maps to the yet_to_be_fetched list
- boolean good = true;
- while (!remaining.isEmpty() && good) {
- good = copyMapOutput(host, input, remaining);
+ // On any error, faildTasks is not null and we exit
+ // after putting back the remaining maps to the
+ // yet_to_be_fetched list and marking the failed tasks.
+ TaskAttemptID[] failedTasks = null;
+ while (!remaining.isEmpty() && failedTasks == null) {
+ failedTasks = copyMapOutput(host, input, remaining);
+ }
+
+ if(failedTasks != null && failedTasks.length > 0) {
+ LOG.warn("copyMapOutput failed for tasks "+Arrays.toString(failedTasks));
+ for(TaskAttemptID left: failedTasks) {
+ scheduler.copyFailed(left, host, true);
+ }
}
IOUtils.cleanup(LOG, input);
// Sanity check
- if (good && !remaining.isEmpty()) {
+ if (failedTasks == null && !remaining.isEmpty()) {
throw new IOException("server didn't return all expected map outputs: "
+ remaining.size() + " left.");
}
@@ -285,10 +338,11 @@ class Fetcher<K,V> extends Thread {
scheduler.putBackKnownMapOutput(host, left);
}
}
-
- }
+ }
+
+ private static TaskAttemptID[] EMPTY_ATTEMPT_ID_ARRAY = new TaskAttemptID[0];
- private boolean copyMapOutput(MapHost host,
+ private TaskAttemptID[] copyMapOutput(MapHost host,
DataInputStream input,
Set<TaskAttemptID> remaining) {
MapOutput<K,V> mapOutput = null;
@@ -310,18 +364,21 @@ class Fetcher<K,V> extends Thread {
} catch (IllegalArgumentException e) {
badIdErrs.increment(1);
LOG.warn("Invalid map id ", e);
- return false;
+ //Don't know which one was bad, so consider all of them as bad
+ return remaining.toArray(new TaskAttemptID[remaining.size()]);
}
// Do some basic sanity verification
if (!verifySanity(compressedLength, decompressedLength, forReduce,
remaining, mapId)) {
- return false;
+ return new TaskAttemptID[] {mapId};
}
- LOG.debug("header: " + mapId + ", len: " + compressedLength +
- ", decomp len: " + decompressedLength);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("header: " + mapId + ", len: " + compressedLength +
+ ", decomp len: " + decompressedLength);
+ }
// Get the location for the map output - either in-memory or on-disk
mapOutput = merger.reserve(mapId, decompressedLength, id);
@@ -329,7 +386,8 @@ class Fetcher<K,V> extends Thread {
// Check if we can shuffle *now* ...
if (mapOutput.getType() == Type.WAIT) {
LOG.info("fetcher#" + id + " - MergerManager returned Status.WAIT ...");
- return false;
+ //Not an error but wait to process data.
+ return EMPTY_ATTEMPT_ID_ARRAY;
}
// Go!
@@ -351,24 +409,27 @@ class Fetcher<K,V> extends Thread {
// Note successful shuffle
remaining.remove(mapId);
metrics.successFetch();
- return true;
+ return null;
} catch (IOException ioe) {
ioErrs.increment(1);
if (mapId == null || mapOutput == null) {
LOG.info("fetcher#" + id + " failed to read map header" +
mapId + " decomp: " +
decompressedLength + ", " + compressedLength, ioe);
- return false;
+ if(mapId == null) {
+ return remaining.toArray(new TaskAttemptID[remaining.size()]);
+ } else {
+ return new TaskAttemptID[] {mapId};
+ }
}
- LOG.info("Failed to shuffle output of " + mapId +
+ LOG.warn("Failed to shuffle output of " + mapId +
" from " + host.getHostName(), ioe);
// Inform the shuffle-scheduler
mapOutput.abort();
- scheduler.copyFailed(mapId, host, true);
metrics.failedFetch();
- return false;
+ return new TaskAttemptID[] {mapId};
}
}
@@ -481,7 +542,7 @@ class Fetcher<K,V> extends Thread {
int decompressedLength,
int compressedLength) throws IOException {
IFileInputStream checksumIn =
- new IFileInputStream(input, compressedLength);
+ new IFileInputStream(input, compressedLength, job);
input = checksumIn;
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ConfigUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ConfigUtil.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ConfigUtil.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ConfigUtil.java Fri Oct 19 02:25:55 2012
@@ -246,8 +246,6 @@ public class ConfigUtil {
new String[] {MRJobConfig.CACHE_FILE_TIMESTAMPS});
Configuration.addDeprecation("mapred.cache.archives.timestamps",
new String[] {MRJobConfig.CACHE_ARCHIVES_TIMESTAMPS});
- Configuration.addDeprecation("mapred.create.symlink",
- new String[] {MRJobConfig.CACHE_SYMLINK});
Configuration.addDeprecation("mapred.working.dir",
new String[] {MRJobConfig.WORKING_DIR});
Configuration.addDeprecation("user.name",
@@ -378,6 +376,8 @@ public class ConfigUtil {
new String[] {MRJobConfig.REDUCE_SKIP_MAXGROUPS});
Configuration.addDeprecation("mapred.reduce.child.log.level",
new String[] {MRJobConfig.REDUCE_LOG_LEVEL});
+ Configuration.addDeprecation("mapreduce.job.counters.limit",
+ new String[] {MRJobConfig.COUNTERS_MAX_KEY});
Configuration.addDeprecation("jobclient.completion.poll.interval",
new String[] {Job.COMPLETION_POLL_INTERVAL_KEY});
Configuration.addDeprecation("jobclient.progress.monitor.poll.interval",
@@ -519,6 +519,13 @@ public class ConfigUtil {
new String[] {
MRJobConfig.MR_AM_SECURITY_SERVICE_AUTHORIZATION_CLIENT
});
+ Configuration.addDeprecation("mapreduce.user.classpath.first",
+ MRJobConfig.MAPREDUCE_JOB_USER_CLASSPATH_FIRST);
+ }
+
+ public static void main(String[] args) {
+ loadResources();
+ Configuration.dumpDeprecatedKeys();
}
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/HostUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/HostUtil.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/HostUtil.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/HostUtil.java Fri Oct 19 02:25:55 2012
@@ -20,6 +20,7 @@ package org.apache.hadoop.mapreduce.util
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.http.HttpConfig;
@Private
@Unstable
@@ -33,9 +34,9 @@ public class HostUtil {
* @return the taskLogUrl
*/
public static String getTaskLogUrl(String taskTrackerHostName,
- String httpPort, String taskAttemptID) {
- return ("http://" + taskTrackerHostName + ":" + httpPort
- + "/tasklog?attemptid=" + taskAttemptID);
+ String httpPort, String taskAttemptID) {
+ return (HttpConfig.getSchemePrefix() + taskTrackerHostName + ":" +
+ httpPort + "/tasklog?attemptid=" + taskAttemptID);
}
public static String convertTrackerNameToHostName(String trackerName) {
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ProcfsBasedProcessTree.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ProcfsBasedProcessTree.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ProcfsBasedProcessTree.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ProcfsBasedProcessTree.java Fri Oct 19 02:25:55 2012
@@ -512,7 +512,7 @@ public class ProcfsBasedProcessTree exte
in = new BufferedReader(fReader);
} catch (FileNotFoundException f) {
// The process vanished in the interim!
- LOG.warn("The process " + pinfo.getPid()
+ LOG.info("The process " + pinfo.getPid()
+ " may have finished in the interim.");
return ret;
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml Fri Oct 19 02:25:55 2012
@@ -24,47 +24,6 @@
<configuration>
<property>
- <name>mapreduce.jobtracker.jobhistory.location</name>
- <value></value>
- <description> If job tracker is static the history files are stored
- in this single well known place. If No value is set here, by default,
- it is in the local file system at ${hadoop.log.dir}/history.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.jobhistory.task.numberprogresssplits</name>
- <value>12</value>
- <description> Every task attempt progresses from 0.0 to 1.0 [unless
- it fails or is killed]. We record, for each task attempt, certain
- statistics over each twelfth of the progress range. You can change
- the number of intervals we divide the entire range of progress into
- by setting this property. Higher values give more precision to the
- recorded data, but costs more memory in the job tracker at runtime.
- Each increment in this attribute costs 16 bytes per running task.
- </description>
-</property>
-
-<property>
- <name>mapreduce.job.userhistorylocation</name>
- <value></value>
- <description> User can specify a location to store the history files of
- a particular job. If nothing is specified, the logs are stored in
- output directory. The files are stored in "_logs/history/" in the directory.
- User can stop logging by giving the value "none".
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.jobhistory.completed.location</name>
- <value></value>
- <description> The completed job history files are stored at this single well
- known location. If nothing is specified, the files are stored at
- ${mapreduce.jobtracker.jobhistory.location}/done.
- </description>
-</property>
-
-<property>
<name>mapreduce.job.committer.setup.cleanup.needed</name>
<value>true</value>
<description> true, if job needs job-setup and job-cleanup.
@@ -99,15 +58,6 @@
</property>
<property>
- <name>mapreduce.jobtracker.address</name>
- <value>local</value>
- <description>The host and port that the MapReduce job tracker runs
- at. If "local", then jobs are run in-process as a single map
- and reduce task.
- </description>
-</property>
-
-<property>
<name>mapreduce.local.clientfactory.class.name</name>
<value>org.apache.hadoop.mapred.LocalClientFactory</value>
<description>This the client factory that is responsible for
@@ -115,138 +65,10 @@
</property>
<property>
- <name>mapreduce.jobtracker.http.address</name>
- <value>0.0.0.0:50030</value>
- <description>
- The job tracker http server address and port the server will listen on.
- If the port is 0 then the server will start on a free port.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.handler.count</name>
- <value>10</value>
- <description>
- The number of server threads for the JobTracker. This should be roughly
- 4% of the number of tasktracker nodes.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.report.address</name>
- <value>127.0.0.1:0</value>
- <description>The interface and port that task tracker server listens on.
- Since it is only connected to by the tasks, it uses the local interface.
- EXPERT ONLY. Should only be changed if your host does not have the loopback
- interface.</description>
-</property>
-
-<property>
- <name>mapreduce.cluster.local.dir</name>
- <value>${hadoop.tmp.dir}/mapred/local</value>
- <description>The local directory where MapReduce stores intermediate
- data files. May be a comma-separated list of
- directories on different devices in order to spread disk i/o.
- Directories that do not exist are ignored.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.system.dir</name>
- <value>${hadoop.tmp.dir}/mapred/system</value>
- <description>The directory where MapReduce stores control files.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.staging.root.dir</name>
- <value>${hadoop.tmp.dir}/mapred/staging</value>
- <description>The root of the staging area for users' job files
- In practice, this should be the directory where users' home
- directories are located (usually /user)
- </description>
-</property>
-
-<property>
- <name>mapreduce.cluster.temp.dir</name>
- <value>${hadoop.tmp.dir}/mapred/temp</value>
- <description>A shared directory for temporary files.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.local.dir.minspacestart</name>
- <value>0</value>
- <description>If the space in mapreduce.cluster.local.dir drops under this,
- do not ask for more tasks.
- Value in bytes.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.local.dir.minspacekill</name>
- <value>0</value>
- <description>If the space in mapreduce.cluster.local.dir drops under this,
- do not ask more tasks until all the current ones have finished and
- cleaned up. Also, to save the rest of the tasks we have running,
- kill one of them, to clean up some space. Start with the reduce tasks,
- then go with the ones that have finished the least.
- Value in bytes.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.expire.trackers.interval</name>
- <value>600000</value>
- <description>Expert: The time-interval, in miliseconds, after which
- a tasktracker is declared 'lost' if it doesn't send heartbeats.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.instrumentation</name>
- <value>org.apache.hadoop.mapred.TaskTrackerMetricsInst</value>
- <description>Expert: The instrumentation class to associate with each TaskTracker.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.resourcecalculatorplugin</name>
- <value></value>
- <description>
- Name of the class whose instance will be used to query resource information
- on the tasktracker.
-
- The class must be an instance of
- org.apache.hadoop.util.ResourceCalculatorPlugin. If the value is null, the
- tasktracker attempts to use a class appropriate to the platform.
- Currently, the only platform supported is Linux.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.taskmemorymanager.monitoringinterval</name>
- <value>5000</value>
- <description>The interval, in milliseconds, for which the tasktracker waits
- between two cycles of monitoring its tasks' memory usage. Used only if
- tasks' memory management is enabled via mapred.tasktracker.tasks.maxmemory.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.tasks.sleeptimebeforesigkill</name>
- <value>5000</value>
- <description>The time, in milliseconds, the tasktracker waits for sending a
- SIGKILL to a task, after it has been sent a SIGTERM. This is currently
- not used on WINDOWS where tasks are just sent a SIGTERM.
- </description>
-</property>
-
-<property>
<name>mapreduce.job.maps</name>
<value>2</value>
<description>The default number of map tasks per job.
- Ignored when mapreduce.jobtracker.address is "local".
+ Ignored when mapreduce.framework.name is "local".
</description>
</property>
@@ -256,54 +78,21 @@
<description>The default number of reduce tasks per job. Typically set to 99%
of the cluster's reduce capacity, so that if a node fails the reduces can
still be executed in a single wave.
- Ignored when mapreduce.jobtracker.address is "local".
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.restart.recover</name>
- <value>false</value>
- <description>"true" to enable (job) recovery upon restart,
- "false" to start afresh
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.jobhistory.block.size</name>
- <value>3145728</value>
- <description>The block size of the job history file. Since the job recovery
- uses job history, its important to dump job history to disk as
- soon as possible. Note that this is an expert level parameter.
- The default value is set to 3 MB.
+ Ignored when mapreduce.framework.name is "local".
</description>
</property>
<property>
- <name>mapreduce.jobtracker.taskscheduler</name>
- <value>org.apache.hadoop.mapred.JobQueueTaskScheduler</value>
- <description>The class responsible for scheduling the tasks.</description>
-</property>
-
-
-<property>
<name>mapreduce.job.split.metainfo.maxsize</name>
<value>10000000</value>
<description>The maximum permissible size of the split metainfo file.
- The JobTracker won't attempt to read split metainfo files bigger than
- the configured value.
+ The MapReduce ApplicationMaster won't attempt to read submitted split metainfo
+ files bigger than this configured value.
No limits if set to -1.
</description>
</property>
<property>
- <name>mapreduce.jobtracker.taskscheduler.maxrunningtasks.perjob</name>
- <value></value>
- <description>The maximum number of running tasks for a job before
- it gets preempted. No limits if undefined.
- </description>
-</property>
-
-<property>
<name>mapreduce.map.maxattempts</name>
<value>4</value>
<description>Expert: The maximum number of attempts per map task.
@@ -333,7 +122,7 @@
<name>mapreduce.reduce.shuffle.connect.timeout</name>
<value>180000</value>
<description>Expert: The maximum amount of time (in milli seconds) reduce
- task spends in trying to connect to a tasktracker for getting map output.
+ task spends in trying to connect to a remote node for getting map output.
</description>
</property>
@@ -355,51 +144,6 @@
</description>
</property>
-<property>
- <name>mapreduce.tasktracker.map.tasks.maximum</name>
- <value>2</value>
- <description>The maximum number of map tasks that will be run
- simultaneously by a task tracker.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.reduce.tasks.maximum</name>
- <value>2</value>
- <description>The maximum number of reduce tasks that will be run
- simultaneously by a task tracker.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.retiredjobs.cache.size</name>
- <value>1000</value>
- <description>The number of retired job status to keep in the cache.
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.outofband.heartbeat</name>
- <value>false</value>
- <description>Expert: Set this to true to let the tasktracker send an
- out-of-band heartbeat on task-completion for better latency.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.jobhistory.lru.cache.size</name>
- <value>5</value>
- <description>The number of job history files loaded in memory. The jobs are
- loaded when they are first accessed. The cache is cleared based on LRU.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.instrumentation</name>
- <value>org.apache.hadoop.mapred.JobTrackerMetricsInst</value>
- <description>Expert: The instrumentation class to associate with each JobTracker.
- </description>
-</property>
<property>
<name>mapred.child.java.opts</name>
@@ -424,7 +168,7 @@
<description>User added environment variables for the task tracker child
processes. Example :
1) A=foo This will set the env variable A to foo
- 2) B=$B:c This is inherit tasktracker's B env variable.
+ 2) B=$B:c This is inherit nodemanager's B env variable.
</description>
</property>
@@ -513,6 +257,21 @@
</property>
<property>
+ <name>mapreduce.shuffle.ssl.enabled</name>
+ <value>false</value>
+ <description>
+ Whether to use SSL for for the Shuffle HTTP endpoints.
+ </description>
+</property>
+
+<property>
+ <name>mapreduce.shuffle.ssl.file.buffer.size</name>
+ <value>65536</value>
+ <description>Buffer size for reading spills from file when using SSL.
+ </description>
+</property>
+
+<property>
<name>mapreduce.reduce.markreset.buffer.percent</name>
<value>0.0</value>
<description>The percentage of memory -relative to the maximum heap size- to
@@ -553,17 +312,9 @@
<name>mapreduce.job.speculative.slownodethreshold</name>
<value>1.0</value>
<description>The number of standard deviations by which a Task
- Tracker's ave map and reduce progress-rates (finishTime-dispatchTime)
+ Tracker's average map and reduce progress-rates (finishTime-dispatchTime)
must be lower than the average of all successful map/reduce task's for
- the TT to be considered too slow to give a speculative task to.
- </description>
-</property>
-
-<property>
- <name>mapreduce.job.jvm.numtasks</name>
- <value>1</value>
- <description>How many tasks to run per jvm. If set to -1, there is
- no limit.
+ the NodeManager to be considered too slow to give a speculative task to.
</description>
</property>
@@ -615,12 +366,6 @@
take priority over this setting.</description>
</property>
-<property>
- <name>mapreduce.jobtracker.maxtasks.perjob</name>
- <value>-1</value>
- <description>The maximum number of tasks for a single job.
- A value of -1 indicates that there is no maximum. </description>
-</property>
<property>
<name>mapreduce.client.submit.file.replication</name>
@@ -630,41 +375,6 @@
</description>
</property>
-
-<property>
- <name>mapreduce.tasktracker.dns.interface</name>
- <value>default</value>
- <description>The name of the Network Interface from which a task
- tracker should report its IP address.
- </description>
- </property>
-
-<property>
- <name>mapreduce.tasktracker.dns.nameserver</name>
- <value>default</value>
- <description>The host name or IP address of the name server (DNS)
- which a TaskTracker should use to determine the host name used by
- the JobTracker for communication and display purposes.
- </description>
- </property>
-
-<property>
- <name>mapreduce.tasktracker.http.threads</name>
- <value>40</value>
- <description>The number of worker threads that for the http server. This is
- used for map output fetching
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.http.address</name>
- <value>0.0.0.0:50060</value>
- <description>
- The task tracker http server address and port.
- If the port is 0 then the server will start on a free port.
- </description>
-</property>
-
<property>
<name>mapreduce.task.files.preserve.failedtasks</name>
<value>false</value>
@@ -737,54 +447,13 @@
</property>
<property>
- <name>mapreduce.job.userlog.retain.hours</name>
- <value>24</value>
- <description>The maximum time, in hours, for which the user-logs are to be
- retained after the job completion.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.hosts.filename</name>
- <value></value>
- <description>Names a file that contains the list of nodes that may
- connect to the jobtracker. If the value is empty, all hosts are
- permitted.</description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.hosts.exclude.filename</name>
- <value></value>
- <description>Names a file that contains the list of hosts that
- should be excluded by the jobtracker. If the value is empty, no
- hosts are excluded.</description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.heartbeats.in.second</name>
- <value>100</value>
- <description>Expert: Approximate number of heart-beats that could arrive
- at JobTracker in a second. Assuming each RPC can be processed
- in 10msec, the default value is made 100 RPCs in a second.
- </description>
-</property>
-
-<property>
- <name>mapreduce.jobtracker.tasktracker.maxblacklists</name>
- <value>4</value>
- <description>The number of blacklists for a taskTracker by various jobs
- after which the task tracker could be blacklisted across
- all jobs. The tracker will be given a tasks later
- (after a day). The tracker will become a healthy
- tracker after a restart.
- </description>
-</property>
-
-<property>
<name>mapreduce.job.maxtaskfailures.per.tracker</name>
- <value>4</value>
- <description>The number of task-failures on a tasktracker of a given job
- after which new tasks of that job aren't assigned to it.
+ <value>3</value>
+ <description>The number of task-failures on a node manager of a given job
+ after which new tasks of that job aren't assigned to it. It
+ MUST be less than mapreduce.map.maxattempts and
+ mapreduce.reduce.maxattempts otherwise the failed task will
+ never be tried on a different node.
</description>
</property>
@@ -802,8 +471,8 @@
<name>mapreduce.client.completion.pollinterval</name>
<value>5000</value>
<description>The interval (in milliseconds) between which the JobClient
- polls the JobTracker for updates about job status. You may want to set this
- to a lower value to make tests run faster on a single node system. Adjusting
+ polls the MapReduce ApplicationMaster for updates about job status. You may want to
+ set this to a lower value to make tests run faster on a single node system. Adjusting
this value in production may lead to unwanted client-server traffic.
</description>
</property>
@@ -818,32 +487,6 @@
</description>
</property>
- <property>
- <name>mapreduce.jobtracker.persist.jobstatus.active</name>
- <value>true</value>
- <description>Indicates if persistency of job status information is
- active or not.
- </description>
- </property>
-
- <property>
- <name>mapreduce.jobtracker.persist.jobstatus.hours</name>
- <value>1</value>
- <description>The number of hours job status information is persisted in DFS.
- The job status information will be available after it drops of the memory
- queue and between jobtracker restarts. With a zero value the job status
- information is not persisted at all in DFS.
- </description>
-</property>
-
- <property>
- <name>mapreduce.jobtracker.persist.jobstatus.dir</name>
- <value>/jobtracker/jobsInfo</value>
- <description>The directory where the job status information is persisted
- in a file system to be available after it drops of the memory queue and
- between jobtracker restarts.
- </description>
- </property>
<property>
<name>mapreduce.task.profile</name>
@@ -876,8 +519,8 @@
<description> The number of Task attempts AFTER which skip mode
will be kicked off. When skip mode is kicked off, the
tasks reports the range of records which it will process
- next, to the TaskTracker. So that on failures, TT knows which
- ones are possibly the bad records. On further executions,
+ next, to the MR ApplicationMaster. So that on failures, the MR AM
+ knows which ones are possibly the bad records. On further executions,
those are skipped.
</description>
</property>
@@ -944,6 +587,20 @@
acceptable.
</description>
</property>
+
+ <property>
+ <name>mapreduce.ifile.readahead</name>
+ <value>true</value>
+ <description>Configuration key to enable/disable IFile readahead.
+ </description>
+ </property>
+
+ <property>
+ <name>mapreduce.ifile.readahead.bytes</name>
+ <value>4194304</value>
+ <description>Configuration key to set the IFile readahead length in bytes.
+ </description>
+ </property>
<!-- Job Notification Configuration -->
@@ -974,15 +631,6 @@
calls</description>
</property>
-<!-- Proxy Configuration -->
-<property>
- <name>mapreduce.jobtracker.taskcache.levels</name>
- <value>2</value>
- <description> This is the max level of the task cache. For example, if
- the level is 2, the tasks cached are at the host level and at the rack
- level.
- </description>
-</property>
<property>
<name>mapreduce.job.queuename</name>
@@ -996,18 +644,29 @@
</property>
<property>
+ <name>mapreduce.cluster.local.dir</name>
+ <value>${hadoop.tmp.dir}/mapred/local</value>
+ <description>
+ The local directory where MapReduce stores intermediate
+ data files. May be a comma-separated list of
+ directories on different devices in order to spread disk i/o.
+ Directories that do not exist are ignored.
+ </description>
+</property>
+
+<property>
<name>mapreduce.cluster.acls.enabled</name>
<value>false</value>
<description> Specifies whether ACLs should be checked
for authorization of users for doing various queue and job level operations.
ACLs are disabled by default. If enabled, access control checks are made by
- JobTracker and TaskTracker when requests are made by users for queue
+ MapReduce ApplicationMaster when requests are made by users for queue
operations like submit job to a queue and kill a job in the queue and job
operations like viewing the job-details (See mapreduce.job.acl-view-job)
or for modifying the job (See mapreduce.job.acl-modify-job) using
Map/Reduce APIs, RPCs or via the console and web user interfaces.
- For enabling this flag(mapreduce.cluster.acls.enabled), this is to be set
- to true in mapred-site.xml on JobTracker node and on all TaskTracker nodes.
+ For enabling this flag, set to true in mapred-site.xml file of all
+ MapReduce clients (MR job submitting nodes).
</description>
</property>
@@ -1060,8 +719,8 @@
o job-level counters
o task-level counters
o tasks' diagnostic information
- o task-logs displayed on the TaskTracker web-UI and
- o job.xml showed by the JobTracker's web-UI
+ o task-logs displayed on the HistoryServer's web-UI and
+ o job.xml showed by the HistoryServer's web-UI
Every other piece of information of jobs is still accessible by any other
user, for e.g., JobStatus, JobProfile, list of jobs in the queue, etc.
@@ -1079,18 +738,10 @@
</property>
<property>
- <name>mapreduce.tasktracker.indexcache.mb</name>
- <value>10</value>
- <description> The maximum memory that a task tracker allows for the
- index cache that is used when serving map outputs to reducers.
- </description>
-</property>
-
-<property>
<name>mapreduce.task.merge.progress.records</name>
<value>10000</value>
<description> The number of records to process during merge before
- sending a progress notification to the TaskTracker.
+ sending a progress notification to the MR ApplicationMaster.
</description>
</property>
@@ -1111,22 +762,6 @@
</property>
<property>
- <name>mapreduce.tasktracker.taskcontroller</name>
- <value>org.apache.hadoop.mapred.DefaultTaskController</value>
- <description>TaskController which is used to launch and manage task execution
- </description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.group</name>
- <value></value>
- <description>Expert: Group to which TaskTracker belongs. If
- LinuxTaskController is configured via mapreduce.tasktracker.taskcontroller,
- the group owner of the task-controller binary should be same as this group.
- </description>
-</property>
-
-<property>
<name>mapreduce.shuffle.port</name>
<value>8080</value>
<description>Default port that the ShuffleHandler will run on. ShuffleHandler
@@ -1135,41 +770,7 @@
</description>
</property>
-<!-- Node health script variables -->
-
-<property>
- <name>mapreduce.tasktracker.healthchecker.script.path</name>
- <value></value>
- <description>Absolute path to the script which is
- periodicallyrun by the node health monitoring service to determine if
- the node is healthy or not. If the value of this key is empty or the
- file does not exist in the location configured here, the node health
- monitoring service is not started.</description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.healthchecker.interval</name>
- <value>60000</value>
- <description>Frequency of the node health script to be run,
- in milliseconds</description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.healthchecker.script.timeout</name>
- <value>600000</value>
- <description>Time after node health script should be killed if
- unresponsive and considered that the script has failed.</description>
-</property>
-
-<property>
- <name>mapreduce.tasktracker.healthchecker.script.args</name>
- <value></value>
- <description>List of arguments which are to be passed to
- node health script when it is being launched comma seperated.
- </description>
-</property>
-
-<!-- end of node health script variables -->
+<!-- MR YARN Application properties -->
<property>
<name>mapreduce.job.counters.limit</name>
@@ -1280,6 +881,13 @@
<description>The amount of memory the MR AppMaster needs.</description>
</property>
+<property>
+ <description>CLASSPATH for MR applications. A comma-separated list
+ of CLASSPATH entries</description>
+ <name>mapreduce.application.classpath</name>
+ <value>$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/*,$HADOOP_MAPRED_HOME/share/hadoop/mapreduce/lib/*</value>
+</property>
+
<!-- jobhistory properties -->
<property>
Propchange: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
------------------------------------------------------------------------------
Merged /hadoop/common/branches/HDFS-3077/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml:r1363593-1396941
Merged /hadoop/common/trunk/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml:r1360400-1399945
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestCounters.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestCounters.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestCounters.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestCounters.java Fri Oct 19 02:25:55 2012
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertTru
import java.io.IOException;
import java.text.ParseException;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.Random;
@@ -151,7 +152,7 @@ public class TestCounters {
Counters counters = new Counters();
counters.incrCounter(Task.Counter.MAP_INPUT_RECORDS, 1);
counters.incrCounter(JobInProgress.Counter.DATA_LOCAL_MAPS, 1);
- counters.findCounter("FileSystemCounter", "FILE_BYTES_READ").increment(1);
+ counters.findCounter("FileSystemCounters", "FILE_BYTES_READ").increment(1);
checkLegacyNames(counters);
}
@@ -179,7 +180,7 @@ public class TestCounters {
assertEquals("New name and method", 1, counters.findCounter("file",
FileSystemCounter.BYTES_READ).getValue());
assertEquals("Legacy name", 1, counters.findCounter(
- "FileSystemCounter",
+ "FileSystemCounters",
"FILE_BYTES_READ").getValue());
}
@@ -225,6 +226,23 @@ public class TestCounters {
}
@Test
+ public void testLegacyGetGroupNames() {
+ Counters counters = new Counters();
+ // create 2 filesystem counter groups
+ counters.findCounter("fs1", FileSystemCounter.BYTES_READ).increment(1);
+ counters.findCounter("fs2", FileSystemCounter.BYTES_READ).increment(1);
+ counters.incrCounter("group1", "counter1", 1);
+
+ HashSet<String> groups = new HashSet<String>(counters.getGroupNames());
+ HashSet<String> expectedGroups = new HashSet<String>();
+ expectedGroups.add("group1");
+ expectedGroups.add("FileSystemCounters"); //Legacy Name
+ expectedGroups.add("org.apache.hadoop.mapreduce.FileSystemCounter");
+
+ assertEquals(expectedGroups, groups);
+ }
+
+ @Test
public void testMakeCompactString() {
final String GC1 = "group1.counter1:1";
final String GC2 = "group2.counter2:3";
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestIndexCache.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestIndexCache.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestIndexCache.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestIndexCache.java Fri Oct 19 02:25:55 2012
@@ -35,16 +35,23 @@ import org.apache.hadoop.mapreduce.serve
import junit.framework.TestCase;
public class TestIndexCache extends TestCase {
+ private JobConf conf;
+ private FileSystem fs;
+ private Path p;
+
+ @Override
+ public void setUp() throws IOException {
+ conf = new JobConf();
+ fs = FileSystem.getLocal(conf).getRaw();
+ p = new Path(System.getProperty("test.build.data", "/tmp"),
+ "cache").makeQualified(fs.getUri(), fs.getWorkingDirectory());
+ }
public void testLRCPolicy() throws Exception {
Random r = new Random();
long seed = r.nextLong();
r.setSeed(seed);
System.out.println("seed: " + seed);
- JobConf conf = new JobConf();
- FileSystem fs = FileSystem.getLocal(conf).getRaw();
- Path p = new Path(System.getProperty("test.build.data", "/tmp"),
- "cache").makeQualified(fs);
fs.delete(p, true);
conf.setInt(TTConfig.TT_INDEX_CACHE, 1);
final int partsPerMap = 1000;
@@ -115,10 +122,6 @@ public class TestIndexCache extends Test
public void testBadIndex() throws Exception {
final int parts = 30;
- JobConf conf = new JobConf();
- FileSystem fs = FileSystem.getLocal(conf).getRaw();
- Path p = new Path(System.getProperty("test.build.data", "/tmp"),
- "cache").makeQualified(fs);
fs.delete(p, true);
conf.setInt(TTConfig.TT_INDEX_CACHE, 1);
IndexCache cache = new IndexCache(conf);
@@ -150,10 +153,6 @@ public class TestIndexCache extends Test
}
public void testInvalidReduceNumberOrLength() throws Exception {
- JobConf conf = new JobConf();
- FileSystem fs = FileSystem.getLocal(conf).getRaw();
- Path p = new Path(System.getProperty("test.build.data", "/tmp"),
- "cache").makeQualified(fs);
fs.delete(p, true);
conf.setInt(TTConfig.TT_INDEX_CACHE, 1);
final int partsPerMap = 1000;
@@ -199,10 +198,6 @@ public class TestIndexCache extends Test
// This test case may not repeatable. But on my macbook this test
// fails with probability of 100% on code before MAPREDUCE-2541,
// so it is repeatable in practice.
- JobConf conf = new JobConf();
- FileSystem fs = FileSystem.getLocal(conf).getRaw();
- Path p = new Path(System.getProperty("test.build.data", "/tmp"),
- "cache").makeQualified(fs);
fs.delete(p, true);
conf.setInt(TTConfig.TT_INDEX_CACHE, 10);
// Make a big file so removeMapThread almost surely runs faster than
@@ -247,6 +242,66 @@ public class TestIndexCache extends Test
}
}
+ public void testCreateRace() throws Exception {
+ fs.delete(p, true);
+ conf.setInt(TTConfig.TT_INDEX_CACHE, 1);
+ final int partsPerMap = 1000;
+ final int bytesPerFile = partsPerMap * 24;
+ final IndexCache cache = new IndexCache(conf);
+
+ final Path racy = new Path(p, "racyIndex");
+ final String user =
+ UserGroupInformation.getCurrentUser().getShortUserName();
+ writeFile(fs, racy, bytesPerFile, partsPerMap);
+
+ // run multiple instances
+ Thread[] getInfoThreads = new Thread[50];
+ for (int i = 0; i < 50; i++) {
+ getInfoThreads[i] = new Thread() {
+ @Override
+ public void run() {
+ try {
+ cache.getIndexInformation("racyIndex", partsPerMap, racy, user);
+ cache.removeMap("racyIndex");
+ } catch (Exception e) {
+ // should not be here
+ }
+ }
+ };
+ }
+
+ for (int i = 0; i < 50; i++) {
+ getInfoThreads[i].start();
+ }
+
+ final Thread mainTestThread = Thread.currentThread();
+
+ Thread timeoutThread = new Thread() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(15000);
+ mainTestThread.interrupt();
+ } catch (InterruptedException ie) {
+ // we are done;
+ }
+ }
+ };
+
+ for (int i = 0; i < 50; i++) {
+ try {
+ getInfoThreads[i].join();
+ } catch (InterruptedException ie) {
+ // we haven't finished in time. Potential deadlock/race.
+ fail("Unexpectedly long delay during concurrent cache entry creations");
+ }
+ }
+ // stop the timeoutThread. If we get interrupted before stopping, there
+ // must be something wrong, although it wasn't a deadlock. No need to
+ // catch and swallow.
+ timeoutThread.interrupt();
+ }
+
private static void checkRecord(IndexRecord rec, long fill) {
assertEquals(fill, rec.startOffset);
assertEquals(fill, rec.rawLength);
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/TestJob.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/TestJob.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/TestJob.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/TestJob.java Fri Oct 19 02:25:55 2012
@@ -18,14 +18,18 @@
package org.apache.hadoop.mapreduce;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
import java.io.IOException;
+import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.JobStatus.State;
import org.apache.hadoop.mapreduce.protocol.ClientProtocol;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
import org.junit.Assert;
import org.junit.Test;
@@ -50,4 +54,22 @@ public class TestJob {
Assert.assertNotNull(job.toString());
}
+ @Test
+ public void testUGICredentialsPropogation() throws Exception {
+ Credentials creds = new Credentials();
+ Token<?> token = mock(Token.class);
+ Text tokenService = new Text("service");
+ Text secretName = new Text("secret");
+ byte secret[] = new byte[]{};
+
+ creds.addToken(tokenService, token);
+ creds.addSecretKey(secretName, secret);
+ UserGroupInformation.getLoginUser().addCredentials(creds);
+
+ JobConf jobConf = new JobConf();
+ Job job = new Job(jobConf);
+
+ assertSame(token, job.getCredentials().getToken(tokenService));
+ assertSame(secret, job.getCredentials().getSecretKey(secretName));
+ }
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/security/TestTokenCache.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/security/TestTokenCache.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/security/TestTokenCache.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/security/TestTokenCache.java Fri Oct 19 02:25:55 2012
@@ -18,23 +18,16 @@
package org.apache.hadoop.mapreduce.security;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.junit.Assert.*;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.*;
import java.io.IOException;
import java.net.URI;
-import java.util.LinkedList;
-import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystemTestHelper.MockFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Master;
@@ -43,145 +36,42 @@ import org.apache.hadoop.security.Creden
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
public class TestTokenCache {
-
- @Test
- @SuppressWarnings("deprecation")
- public void testGetDelegationTokensNotImplemented() throws Exception {
- Credentials credentials = new Credentials();
- Configuration conf = new Configuration();
+ private static Configuration conf;
+ private static String renewer;
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ conf = new Configuration();
conf.set(YarnConfiguration.RM_PRINCIPAL, "mapred/host@REALM");
- String renewer = Master.getMasterPrincipal(conf);
-
- FileSystem fs = setupSingleFsWithoutGetDelegationTokens();
- TokenCache.obtainTokensForNamenodesInternal(fs, credentials, conf);
- assertEquals(1, credentials.getAllTokens().size());
-
- verify(fs).getDelegationTokens(renewer, credentials);
- verify(fs).getDelegationToken(renewer);
+ renewer = Master.getMasterPrincipal(conf);
}
@Test
- public void testManagedFileSystem() throws Exception {
+ public void testObtainTokens() throws Exception {
Credentials credentials = new Credentials();
- Configuration conf = new Configuration();
- conf.set(YarnConfiguration.RM_PRINCIPAL, "mapred/host@REALM");
- String renewer = Master.getMasterPrincipal(conf);
-
- FileSystem singleFs = setupSingleFs();
- FileSystem multiFs = setupMultiFs(singleFs, renewer, credentials);
-
- TokenCache.obtainTokensForNamenodesInternal(singleFs, credentials, conf);
- assertEquals(1, credentials.getAllTokens().size());
-
- TokenCache.obtainTokensForNamenodesInternal(singleFs, credentials, conf);
- assertEquals(1, credentials.getAllTokens().size());
-
- TokenCache.obtainTokensForNamenodesInternal(multiFs, credentials, conf);
- assertEquals(2, credentials.getAllTokens().size());
-
- TokenCache.obtainTokensForNamenodesInternal(multiFs, credentials, conf);
- assertEquals(2, credentials.getAllTokens().size());
-
- verify(singleFs, times(1)).getDelegationTokens(renewer, credentials);
- verify(multiFs, times(2)).getDelegationTokens(renewer, credentials);
- // A call to getDelegationToken would have generated an exception.
- }
-
- @SuppressWarnings("deprecation")
- private FileSystem setupSingleFsWithoutGetDelegationTokens() throws Exception {
- FileSystem mockFs = mock(FileSystem.class);
- when(mockFs.getCanonicalServiceName()).thenReturn("singlefs4");
- when(mockFs.getUri()).thenReturn(new URI("singlefs4:///"));
-
- final Token<?> mockToken = (Token<?>) mock(Token.class);
- when(mockToken.getService()).thenReturn(new Text("singlefs4"));
-
- when(mockFs.getDelegationToken(any(String.class))).thenAnswer(
- new Answer<Token<?>>() {
- @Override
- public Token<?> answer(InvocationOnMock invocation) throws Throwable {
- return mockToken;
- }
- });
-
- when(mockFs.getDelegationTokens(any(String.class), any(Credentials.class)))
- .thenReturn(new LinkedList<Token<?>>());
-
- return mockFs;
- }
-
- private FileSystem setupSingleFs() throws Exception {
- FileSystem mockFs = mock(FileSystem.class);
- when(mockFs.getCanonicalServiceName()).thenReturn("singlefs1");
- when(mockFs.getUri()).thenReturn(new URI("singlefs1:///"));
-
- List<Token<?>> tokens = new LinkedList<Token<?>>();
- Token<?> mockToken = mock(Token.class);
- when(mockToken.getService()).thenReturn(new Text("singlefs1"));
- tokens.add(mockToken);
-
- when(mockFs.getDelegationTokens(any(String.class))).thenThrow(
- new RuntimeException(
- "getDelegationTokens(renewer) should not be called"));
- when(mockFs.getDelegationTokens(any(String.class), any(Credentials.class)))
- .thenReturn(tokens);
-
- return mockFs;
- }
-
- private FileSystem setupMultiFs(final FileSystem singleFs,
- final String renewer, final Credentials credentials) throws Exception {
- FileSystem mockFs = mock(FileSystem.class);
- when(mockFs.getCanonicalServiceName()).thenReturn(null);
- when(mockFs.getUri()).thenReturn(new URI("multifs:///"));
-
- when(mockFs.getDelegationTokens(any(String.class))).thenThrow(
- new RuntimeException(
- "getDelegationTokens(renewer) should not be called"));
- when(mockFs.getDelegationTokens(renewer, credentials)).thenAnswer(
- new Answer<List<Token<?>>>() {
-
- @Override
- public List<Token<?>> answer(InvocationOnMock invocation)
- throws Throwable {
- List<Token<?>> newTokens = new LinkedList<Token<?>>();
- if (credentials.getToken(new Text("singlefs1")) == null) {
- newTokens.addAll(singleFs.getDelegationTokens(renewer,
- credentials));
- } else {
- newTokens.add(credentials.getToken(new Text("singlefs1")));
- }
- Token<?> mockToken2 = mock(Token.class);
- when(mockToken2.getService()).thenReturn(new Text("singlefs2"));
- newTokens.add(mockToken2);
- return newTokens;
- }
- });
-
- return mockFs;
+ FileSystem fs = mock(FileSystem.class);
+ TokenCache.obtainTokensForNamenodesInternal(fs, credentials, conf);
+ verify(fs).addDelegationTokens(eq(renewer), eq(credentials));
}
@Test
@SuppressWarnings("deprecation")
public void testBinaryCredentials() throws Exception {
- Configuration conf = new Configuration();
- conf.set(YarnConfiguration.RM_PRINCIPAL, "mapred/host@REALM");
- String renewer = Master.getMasterPrincipal(conf);
-
Path TEST_ROOT_DIR =
new Path(System.getProperty("test.build.data","test/build/data"));
// ick, but need fq path minus file:/
String binaryTokenFile = FileSystem.getLocal(conf).makeQualified(
new Path(TEST_ROOT_DIR, "tokenFile")).toUri().getPath();
- FileSystem fs1 = createFileSystemForService("service1");
- FileSystem fs2 = createFileSystemForService("service2");
- FileSystem fs3 = createFileSystemForService("service3");
+ MockFileSystem fs1 = createFileSystemForServiceName("service1");
+ MockFileSystem fs2 = createFileSystemForServiceName("service2");
+ MockFileSystem fs3 = createFileSystemForServiceName("service3");
// get the tokens for fs1 & fs2 and write out to binary creds file
Credentials creds = new Credentials();
@@ -196,7 +86,7 @@ public class TestTokenCache {
// re-init creds and add a newer token for fs1
creds = new Credentials();
Token<?> newerToken1 = fs1.getDelegationToken(renewer);
- assertFalse(newerToken1.equals(token1));
+ assertNotSame(newerToken1, token1);
creds.addToken(newerToken1.getService(), newerToken1);
checkToken(creds, newerToken1);
@@ -230,10 +120,9 @@ public class TestTokenCache {
}
}
- @SuppressWarnings("deprecation")
- private FileSystem createFileSystemForService(final String service)
+ private MockFileSystem createFileSystemForServiceName(final String service)
throws IOException {
- FileSystem mockFs = mock(FileSystem.class);
+ MockFileSystem mockFs = new MockFileSystem();
when(mockFs.getCanonicalServiceName()).thenReturn(service);
when(mockFs.getDelegationToken(any(String.class))).thenAnswer(
new Answer<Token<?>>() {
@@ -258,7 +147,8 @@ public class TestTokenCache {
String renewer = Master.getMasterPrincipal(conf);
Credentials credentials = new Credentials();
- FileSystem mockFs = mock(FileSystem.class);
+ final MockFileSystem fs = new MockFileSystem();
+ final MockFileSystem mockFs = (MockFileSystem) fs.getRawFileSystem();
when(mockFs.getCanonicalServiceName()).thenReturn("host:0");
when(mockFs.getUri()).thenReturn(new URI("mockfs://host:0"));
@@ -266,9 +156,9 @@ public class TestTokenCache {
when(mockPath.getFileSystem(conf)).thenReturn(mockFs);
Path[] paths = new Path[]{ mockPath, mockPath };
- when(mockFs.getDelegationTokens("me", credentials)).thenReturn(null);
+ when(mockFs.addDelegationTokens("me", credentials)).thenReturn(null);
TokenCache.obtainTokensForNamenodesInternal(credentials, paths, conf);
- verify(mockFs, times(1)).getDelegationTokens(renewer, credentials);
+ verify(mockFs, times(1)).addDelegationTokens(renewer, credentials);
}
@Test
@@ -278,5 +168,4 @@ public class TestTokenCache {
TokenCache.cleanUpTokenReferral(conf);
assertNull(conf.get(MRJobConfig.MAPREDUCE_JOB_CREDENTIALS_BINARY));
}
-
-}
+}
\ No newline at end of file
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedTask.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedTask.java?rev=1399950&r1=1399949&r2=1399950&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedTask.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/CompletedTask.java Fri Oct 19 02:25:55 2012
@@ -42,6 +42,8 @@ import org.apache.hadoop.yarn.util.Recor
public class CompletedTask implements Task {
+ private static final Counters EMPTY_COUNTERS = new Counters();
+
private final TaskId taskId;
private final TaskInfo taskInfo;
private TaskReport report;
@@ -120,11 +122,20 @@ public class CompletedTask implements Ta
loadAllTaskAttempts();
this.report = Records.newRecord(TaskReport.class);
report.setTaskId(taskId);
- report.setStartTime(taskInfo.getStartTime());
+ long minLaunchTime = Long.MAX_VALUE;
+ for(TaskAttempt attempt: attempts.values()) {
+ minLaunchTime = Math.min(minLaunchTime, attempt.getLaunchTime());
+ }
+ minLaunchTime = minLaunchTime == Long.MAX_VALUE ? -1 : minLaunchTime;
+ report.setStartTime(minLaunchTime);
report.setFinishTime(taskInfo.getFinishTime());
report.setTaskState(getState());
report.setProgress(getProgress());
- report.setCounters(TypeConverter.toYarn(getCounters()));
+ Counters counters = getCounters();
+ if (counters == null) {
+ counters = EMPTY_COUNTERS;
+ }
+ report.setCounters(TypeConverter.toYarn(counters));
if (successfulAttempt != null) {
report.setSuccessfulAttempt(successfulAttempt);
}