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 cu...@apache.org on 2014/08/20 03:34:56 UTC
svn commit: r1619019 [2/2] - in
/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project: ./ bin/ conf/
hadoop-mapreduce-client/hadoop-mapreduce-client-app/
hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/
h...
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRWebAppUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRWebAppUtil.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRWebAppUtil.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapreduce/v2/util/MRWebAppUtil.java Wed Aug 20 01:34:29 2014
@@ -29,6 +29,7 @@ import org.apache.hadoop.mapreduce.v2.jo
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
import java.net.InetAddress;
import java.net.InetSocketAddress;
@@ -105,11 +106,15 @@ public class MRWebAppUtil {
public static InetSocketAddress getJHSWebBindAddress(Configuration conf) {
if (httpPolicyInJHS == Policy.HTTPS_ONLY) {
- return conf.getSocketAddr(JHAdminConfig.MR_HISTORY_WEBAPP_HTTPS_ADDRESS,
+ return conf.getSocketAddr(
+ JHAdminConfig.MR_HISTORY_BIND_HOST,
+ JHAdminConfig.MR_HISTORY_WEBAPP_HTTPS_ADDRESS,
JHAdminConfig.DEFAULT_MR_HISTORY_WEBAPP_HTTPS_ADDRESS,
JHAdminConfig.DEFAULT_MR_HISTORY_WEBAPP_HTTPS_PORT);
} else {
- return conf.getSocketAddr(JHAdminConfig.MR_HISTORY_WEBAPP_ADDRESS,
+ return conf.getSocketAddr(
+ JHAdminConfig.MR_HISTORY_BIND_HOST,
+ JHAdminConfig.MR_HISTORY_WEBAPP_ADDRESS,
JHAdminConfig.DEFAULT_MR_HISTORY_WEBAPP_ADDRESS,
JHAdminConfig.DEFAULT_MR_HISTORY_WEBAPP_PORT);
}
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/InputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/InputFormat.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/InputFormat.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/InputFormat.java Wed Aug 20 01:34:29 2014
@@ -50,7 +50,7 @@ import org.apache.hadoop.fs.FileSystem;
* bytes, of the input files. However, the {@link FileSystem} blocksize of
* the input files is treated as an upper bound for input splits. A lower bound
* on the split size can be set via
- * <a href="{@docRoot}/../mapred-default.html#mapreduce.input.fileinputformat.split.minsize">
+ * <a href="{@docRoot}/../hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml#mapreduce.input.fileinputformat.split.minsize">
* mapreduce.input.fileinputformat.split.minsize</a>.</p>
*
* <p>Clearly, logical splits based on input-size is insufficient for many
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/Task.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/Task.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/Task.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/Task.java Wed Aug 20 01:34:29 2014
@@ -66,6 +66,7 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.util.Progress;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringInterner;
import org.apache.hadoop.util.StringUtils;
@@ -322,6 +323,11 @@ abstract public class Task implements Wr
protected void reportFatalError(TaskAttemptID id, Throwable throwable,
String logMsg) {
LOG.fatal(logMsg);
+
+ if (ShutdownHookManager.get().isShutdownInProgress()) {
+ return;
+ }
+
Throwable tCause = throwable.getCause();
String cause = tCause == null
? StringUtils.stringifyException(throwable)
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskCompletionEvent.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskCompletionEvent.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskCompletionEvent.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/TaskCompletionEvent.java Wed Aug 20 01:34:29 2014
@@ -90,8 +90,8 @@ public class TaskCompletionEvent
}
/**
- * Returns enum Status.SUCESS or Status.FAILURE.
- * @return task tracker status
+ * Returns {@link Status}
+ * @return task completion status
*/
public Status getTaskStatus() {
return Status.valueOf(super.getStatus().name());
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/join/CompositeInputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/join/CompositeInputFormat.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/join/CompositeInputFormat.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/join/CompositeInputFormat.java Wed Aug 20 01:34:29 2014
@@ -36,7 +36,6 @@ import org.apache.hadoop.mapred.Reporter
/**
* An InputFormat capable of performing joins over a set of data sources sorted
* and partitioned the same way.
- * @see #setFormat
*
* A user may define new join types by setting the property
* <tt>mapred.join.define.<ident></tt> to a classname. In the expression
@@ -44,6 +43,7 @@ import org.apache.hadoop.mapred.Reporter
* ComposableRecordReader.
* <tt>mapred.join.keycomparator</tt> can be a classname used to compare keys
* in the join.
+ * @see #setFormat
* @see JoinRecordReader
* @see MultiFilterRecordReader
*/
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/InputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/InputFormat.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/InputFormat.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/InputFormat.java Wed Aug 20 01:34:29 2014
@@ -52,7 +52,7 @@ import org.apache.hadoop.mapreduce.lib.i
* bytes, of the input files. However, the {@link FileSystem} blocksize of
* the input files is treated as an upper bound for input splits. A lower bound
* on the split size can be set via
- * <a href="{@docRoot}/../mapred-default.html#mapreduce.input.fileinputformat.split.minsize">
+ * <a href="{@docRoot}/../hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml#mapreduce.input.fileinputformat.split.minsize">
* mapreduce.input.fileinputformat.split.minsize</a>.</p>
*
* <p>Clearly, logical splits based on input-size is insufficient for many
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java Wed Aug 20 01:34:29 2014
@@ -54,7 +54,7 @@ import org.apache.hadoop.util.StringUtil
* <p>Here is an example on how to submit a job:</p>
* <p><blockquote><pre>
* // Create a new Job
- * Job job = new Job(new Configuration());
+ * Job job = Job.getInstance();
* job.setJarByClass(MyJob.class);
*
* // Specify various job-specific parameters
@@ -113,16 +113,25 @@ public class Job extends JobContextImpl
private long statustime;
private Cluster cluster;
+ /**
+ * @deprecated Use {@link #getInstance()}
+ */
@Deprecated
public Job() throws IOException {
this(new Configuration());
}
+ /**
+ * @deprecated Use {@link #getInstance(Configuration)}
+ */
@Deprecated
public Job(Configuration conf) throws IOException {
this(new JobConf(conf));
}
+ /**
+ * @deprecated Use {@link #getInstance(Configuration, String)}
+ */
@Deprecated
public Job(Configuration conf, String jobName) throws IOException {
this(conf);
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java Wed Aug 20 01:34:29 2014
@@ -265,6 +265,7 @@ public interface MRJobConfig {
public static final String REDUCE_MEMORY_TOTAL_BYTES = "mapreduce.reduce.memory.totalbytes";
public static final String SHUFFLE_INPUT_BUFFER_PERCENT = "mapreduce.reduce.shuffle.input.buffer.percent";
+ public static final float DEFAULT_SHUFFLE_INPUT_BUFFER_PERCENT = 0.70f;
public static final String SHUFFLE_MEMORY_LIMIT_PERCENT
= "mapreduce.reduce.shuffle.memory.limit.percent";
@@ -292,11 +293,19 @@ public interface MRJobConfig {
public static final String SHUFFLE_READ_TIMEOUT = "mapreduce.reduce.shuffle.read.timeout";
public static final String SHUFFLE_FETCH_FAILURES = "mapreduce.reduce.shuffle.maxfetchfailures";
+ public static final String MAX_ALLOWED_FETCH_FAILURES_FRACTION = "mapreduce.reduce.shuffle.max-fetch-failures-fraction";
+ public static final float DEFAULT_MAX_ALLOWED_FETCH_FAILURES_FRACTION = 0.5f;
+
+ public static final String MAX_FETCH_FAILURES_NOTIFICATIONS = "mapreduce.reduce.shuffle.max-fetch-failures-notifications";
+ public static final int DEFAULT_MAX_FETCH_FAILURES_NOTIFICATIONS = 3;
public static final String SHUFFLE_NOTIFY_READERROR = "mapreduce.reduce.shuffle.notify.readerror";
public static final String MAX_SHUFFLE_FETCH_RETRY_DELAY = "mapreduce.reduce.shuffle.retry-delay.max.ms";
public static final long DEFAULT_MAX_SHUFFLE_FETCH_RETRY_DELAY = 60000;
+
+ public static final String MAX_SHUFFLE_FETCH_HOST_FAILURES = "mapreduce.reduce.shuffle.max-host-failures";
+ public static final int DEFAULT_MAX_SHUFFLE_FETCH_HOST_FAILURES = 5;
public static final String REDUCE_SKIP_INCR_PROC_COUNT = "mapreduce.reduce.skip.proc-count.auto-incr";
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/TaskCompletionEvent.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/TaskCompletionEvent.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/TaskCompletionEvent.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/TaskCompletionEvent.java Wed Aug 20 01:34:29 2014
@@ -95,8 +95,8 @@ public class TaskCompletionEvent impleme
}
/**
- * Returns enum Status.SUCESS or Status.FAILURE.
- * @return task tracker status
+ * Returns {@link Status}
+ * @return task completion status
*/
public Status getStatus() {
return status;
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/db/OracleDBRecordReader.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/db/OracleDBRecordReader.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/db/OracleDBRecordReader.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/db/OracleDBRecordReader.java Wed Aug 20 01:34:29 2014
@@ -81,15 +81,14 @@ public class OracleDBRecordReader<T exte
try {
DBInputFormat.DBInputSplit split = getSplit();
- if (split.getLength() > 0 && split.getStart() > 0){
+ if (split.getLength() > 0){
String querystring = query.toString();
query = new StringBuilder();
query.append("SELECT * FROM (SELECT a.*,ROWNUM dbif_rno FROM ( ");
query.append(querystring);
- query.append(" ) a WHERE rownum <= ").append(split.getStart());
- query.append(" + ").append(split.getLength());
- query.append(" ) WHERE dbif_rno >= ").append(split.getStart());
+ query.append(" ) a WHERE rownum <= ").append(split.getEnd());
+ query.append(" ) WHERE dbif_rno > ").append(split.getStart());
}
} catch (IOException ex) {
// ignore, will not throw.
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/KeyValueTextInputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/KeyValueTextInputFormat.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/KeyValueTextInputFormat.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/KeyValueTextInputFormat.java Wed Aug 20 01:34:29 2014
@@ -38,6 +38,9 @@ import org.apache.hadoop.mapreduce.TaskA
* Either line feed or carriage-return are used to signal end of line.
* Each line is divided into key and value parts by a separator byte. If no
* such a byte exists, the key will be the entire line and value will be empty.
+ * The separator byte can be specified in config file under the attribute name
+ * mapreduce.input.keyvaluelinerecordreader.key.value.separator. The default
+ * is the tab character ('\t').
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/join/CompositeInputFormat.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/join/CompositeInputFormat.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/join/CompositeInputFormat.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/join/CompositeInputFormat.java Wed Aug 20 01:34:29 2014
@@ -39,7 +39,6 @@ import org.apache.hadoop.mapreduce.TaskA
/**
* An InputFormat capable of performing joins over a set of data sources sorted
* and partitioned the same way.
- * @see #setFormat
*
* A user may define new join types by setting the property
* <tt>mapreduce.join.define.<ident></tt> to a classname.
@@ -47,6 +46,7 @@ import org.apache.hadoop.mapreduce.TaskA
* assumed to be a ComposableRecordReader.
* <tt>mapreduce.join.keycomparator</tt> can be a classname used to compare
* keys in the join.
+ * @see #setFormat
* @see JoinRecordReader
* @see MultiFilterRecordReader
*/
Modified: hadoop/common/branches/YARN-1051/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/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/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/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java Wed Aug 20 01:34:29 2014
@@ -319,6 +319,7 @@ class Fetcher<K,V> extends Thread {
// If connect did not succeed, just mark all the maps as failed,
// indirectly penalizing the host
+ scheduler.hostFailed(host.getHostName());
for(TaskAttemptID left: remaining) {
scheduler.copyFailed(left, host, false, connectExcpt);
}
@@ -343,6 +344,7 @@ class Fetcher<K,V> extends Thread {
if(failedTasks != null && failedTasks.length > 0) {
LOG.warn("copyMapOutput failed for tasks "+Arrays.toString(failedTasks));
+ scheduler.hostFailed(host.getHostName());
for(TaskAttemptID left: failedTasks) {
scheduler.copyFailed(left, host, true, false);
}
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/MergeManagerImpl.java Wed Aug 20 01:34:29 2014
@@ -156,7 +156,8 @@ public class MergeManagerImpl<K, V> impl
this.rfs = ((LocalFileSystem)localFS).getRaw();
final float maxInMemCopyUse =
- jobConf.getFloat(MRJobConfig.SHUFFLE_INPUT_BUFFER_PERCENT, 0.90f);
+ jobConf.getFloat(MRJobConfig.SHUFFLE_INPUT_BUFFER_PERCENT,
+ MRJobConfig.DEFAULT_SHUFFLE_INPUT_BUFFER_PERCENT);
if (maxInMemCopyUse > 1.0 || maxInMemCopyUse < 0.0) {
throw new IllegalArgumentException("Invalid value for " +
MRJobConfig.SHUFFLE_INPUT_BUFFER_PERCENT + ": " +
@@ -197,7 +198,7 @@ public class MergeManagerImpl<K, V> impl
"memToMemMergeOutputsThreshold=" + memToMemMergeOutputsThreshold);
if (this.maxSingleShuffleLimit >= this.mergeThreshold) {
- throw new RuntimeException("Invlaid configuration: "
+ throw new RuntimeException("Invalid configuration: "
+ "maxSingleShuffleLimit should be less than mergeThreshold"
+ "maxSingleShuffleLimit: " + this.maxSingleShuffleLimit
+ "mergeThreshold: " + this.mergeThreshold);
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java Wed Aug 20 01:34:29 2014
@@ -18,7 +18,6 @@
package org.apache.hadoop.mapreduce.task.reduce;
import java.io.IOException;
-
import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
@@ -101,6 +100,7 @@ public class ShuffleSchedulerImpl<K,V> i
private final boolean reportReadErrorImmediately;
private long maxDelay = MRJobConfig.DEFAULT_MAX_SHUFFLE_FETCH_RETRY_DELAY;
+ private int maxHostFailures;
public ShuffleSchedulerImpl(JobConf job, TaskStatus status,
TaskAttemptID reduceId,
@@ -132,6 +132,9 @@ public class ShuffleSchedulerImpl<K,V> i
this.maxDelay = job.getLong(MRJobConfig.MAX_SHUFFLE_FETCH_RETRY_DELAY,
MRJobConfig.DEFAULT_MAX_SHUFFLE_FETCH_RETRY_DELAY);
+ this.maxHostFailures = job.getInt(
+ MRJobConfig.MAX_SHUFFLE_FETCH_HOST_FAILURES,
+ MRJobConfig.DEFAULT_MAX_SHUFFLE_FETCH_HOST_FAILURES);
}
@Override
@@ -213,9 +216,18 @@ public class ShuffleSchedulerImpl<K,V> i
progress.setStatus("copy(" + mapsDone + " of " + totalMaps + " at "
+ mbpsFormat.format(transferRate) + " MB/s)");
}
+
+ public synchronized void hostFailed(String hostname) {
+ if (hostFailures.containsKey(hostname)) {
+ IntWritable x = hostFailures.get(hostname);
+ x.set(x.get() + 1);
+ } else {
+ hostFailures.put(hostname, new IntWritable(1));
+ }
+ }
public synchronized void copyFailed(TaskAttemptID mapId, MapHost host,
- boolean readError, boolean connectExcpt) {
+ boolean readError, boolean connectExcpt) {
host.penalize();
int failures = 1;
if (failureCounts.containsKey(mapId)) {
@@ -226,12 +238,9 @@ public class ShuffleSchedulerImpl<K,V> i
failureCounts.put(mapId, new IntWritable(1));
}
String hostname = host.getHostName();
- if (hostFailures.containsKey(hostname)) {
- IntWritable x = hostFailures.get(hostname);
- x.set(x.get() + 1);
- } else {
- hostFailures.put(hostname, new IntWritable(1));
- }
+ //report failure if already retried maxHostFailures times
+ boolean hostFail = hostFailures.get(hostname).get() > getMaxHostFailures() ? true : false;
+
if (failures >= abortFailureLimit) {
try {
throw new IOException(failures + " failures downloading " + mapId);
@@ -240,7 +249,7 @@ public class ShuffleSchedulerImpl<K,V> i
}
}
- checkAndInformJobTracker(failures, mapId, readError, connectExcpt);
+ checkAndInformJobTracker(failures, mapId, readError, connectExcpt, hostFail);
checkReducerHealth();
@@ -270,9 +279,9 @@ public class ShuffleSchedulerImpl<K,V> i
// after every 'maxFetchFailuresBeforeReporting' failures
private void checkAndInformJobTracker(
int failures, TaskAttemptID mapId, boolean readError,
- boolean connectExcpt) {
+ boolean connectExcpt, boolean hostFailed) {
if (connectExcpt || (reportReadErrorImmediately && readError)
- || ((failures % maxFetchFailuresBeforeReporting) == 0)) {
+ || ((failures % maxFetchFailuresBeforeReporting) == 0) || hostFailed) {
LOG.info("Reporting fetch failure for " + mapId + " to jobtracker.");
status.addFetchFailedMap((org.apache.hadoop.mapred.TaskAttemptID) mapId);
}
@@ -507,4 +516,7 @@ public class ShuffleSchedulerImpl<K,V> i
referee.join();
}
+ public int getMaxHostFailures() {
+ return maxHostFailures;
+ }
}
Modified: hadoop/common/branches/YARN-1051/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/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml Wed Aug 20 01:34:29 2014
@@ -1225,9 +1225,9 @@
<property>
<name>mapreduce.job.classloader.system.classes</name>
- <value>java.,javax.,org.apache.commons.logging.,org.apache.log4j.,
- org.apache.hadoop.,core-default.xml,hdfs-default.xml,
- mapred-default.xml,yarn-default.xml</value>
+ <value>java.,javax.,org.w3c.dom.,org.xml.sax.,org.apache.commons.logging.,
+ org.apache.log4j.,org.apache.hadoop.,core-default.xml,
+ hdfs-default.xml,mapred-default.xml,yarn-default.xml</value>
<description>A comma-separated list of classes that should be loaded from the
system classpath, not the user-supplied JARs, when mapreduce.job.classloader
is enabled. Names ending in '.' (period) are treated as package names,
Propchange: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
------------------------------------------------------------------------------
Merged /hadoop/common/trunk/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml:r1613508-1619017
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/DistCp.md.vm Wed Aug 20 01:34:29 2014
@@ -118,9 +118,9 @@ $H3 Basic Usage
$H3 Update and Overwrite
- `-update` is used to copy files from source that don't exist at the target,
- or have different contents. `-overwrite` overwrites target-files even if they
- exist at the source, or have the same contents.
+ `-update` is used to copy files from source that don't exist at the target
+ or differ than the target version. `-overwrite` overwrites target-files that
+ exist at the target.
Update and Overwrite options warrant special attention, since their handling
of source-paths varies from the defaults in a very subtle manner. Consider a
@@ -201,7 +201,7 @@ Flag | Description
`-log <logdir>` | Write logs to \<logdir\> | DistCp keeps logs of each file it attempts to copy as map output. If a map fails, the log output will not be retained if it is re-executed.
`-m <num_maps>` | Maximum number of simultaneous copies | Specify the number of maps to copy data. Note that more maps may not necessarily improve throughput.
`-overwrite` | Overwrite destination | If a map fails and `-i` is not specified, all the files in the split, not only those that failed, will be recopied. As discussed in the Usage documentation, it also changes the semantics for generating destination paths, so users should use this carefully.
-`-update` | Overwrite if src size different from dst size | As noted in the preceding, this is not a "sync" operation. The only criterion examined is the source and destination file sizes; if they differ, the source file replaces the destination file. As discussed in the Usage documentation, it also changes the semantics for generating destination paths, so users should use this carefully.
+`-update` | Overwrite if source and destination differ in size, blocksize, or checksum | As noted in the preceding, this is not a "sync" operation. The criteria examined are the source and destination file sizes, blocksizes, and checksums; if they differ, the source file replaces the destination file. As discussed in the Usage documentation, it also changes the semantics for generating destination paths, so users should use this carefully.
`-f <urilist_uri>` | Use list at \<urilist_uri\> as src list | This is equivalent to listing each source on the command line. The `urilist_uri` list should be a fully qualified URI.
`-filelimit <n>` | Limit the total number of files to be <= n | **Deprecated!** Ignored in the new DistCp.
`-sizelimit <n>` | Limit the total size to be <= n bytes | **Deprecated!** Ignored in the new DistCp.
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/HadoopArchives.md.vm
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/HadoopArchives.md.vm?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/HadoopArchives.md.vm (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/HadoopArchives.md.vm Wed Aug 20 01:34:29 2014
@@ -20,6 +20,7 @@ Hadoop Archives Guide
- [Overview](#Overview)
- [How to Create an Archive](#How_to_Create_an_Archive)
- [How to Look Up Files in Archives](#How_to_Look_Up_Files_in_Archives)
+ - [How to Unarchive an Archive](#How_to_Unarchive_an_Archive)
- [Archives Examples](#Archives_Examples)
- [Creating an Archive](#Creating_an_Archive)
- [Looking Up Files](#Looking_Up_Files)
@@ -70,6 +71,20 @@ How to Look Up Files in Archives
`har:///archivepath/fileinarchive`
+How to Unarchive an Archive
+---------------------------
+
+ Since all the fs shell commands in the archives work transparently,
+ unarchiving is just a matter of copying.
+
+ To unarchive sequentially:
+
+ `hdfs dfs -cp har:///user/zoo/foo.har/dir1 hdfs:/user/zoo/newdir`
+
+ To unarchive in parallel, use DistCp:
+
+ `hadoop distcp har:///user/zoo/foo.har/dir1 hdfs:/user/zoo/newdir`
+
Archives Examples
-----------------
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/db/TestDbClasses.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/db/TestDbClasses.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/db/TestDbClasses.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/db/TestDbClasses.java Wed Aug 20 01:34:29 2014
@@ -110,7 +110,7 @@ public class TestDbClasses {
splitter, NullDBWritable.class, configuration, connect,
dbConfiguration, "condition", fields, "table");
assertEquals(
- "SELECT * FROM (SELECT a.*,ROWNUM dbif_rno FROM ( SELECT f1, f2 FROM table WHERE condition ORDER BY Order ) a WHERE rownum <= 1 + 9 ) WHERE dbif_rno >= 1",
+ "SELECT * FROM (SELECT a.*,ROWNUM dbif_rno FROM ( SELECT f1, f2 FROM table WHERE condition ORDER BY Order ) a WHERE rownum <= 10 ) WHERE dbif_rno > 1",
recorder.getSelectQuery());
}
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryClientService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryClientService.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryClientService.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryClientService.java Wed Aug 20 01:34:29 2014
@@ -83,6 +83,7 @@ import org.apache.hadoop.security.token.
import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.util.Records;
import org.apache.hadoop.yarn.webapp.WebApp;
@@ -119,6 +120,7 @@ public class HistoryClientService extend
YarnRPC rpc = YarnRPC.create(conf);
initializeWebApp(conf);
InetSocketAddress address = conf.getSocketAddr(
+ JHAdminConfig.MR_HISTORY_BIND_HOST,
JHAdminConfig.MR_HISTORY_ADDRESS,
JHAdminConfig.DEFAULT_MR_HISTORY_ADDRESS,
JHAdminConfig.DEFAULT_MR_HISTORY_PORT);
@@ -137,9 +139,11 @@ public class HistoryClientService extend
}
server.start();
- this.bindAddress = conf.updateConnectAddr(JHAdminConfig.MR_HISTORY_ADDRESS,
+ this.bindAddress = conf.updateConnectAddr(JHAdminConfig.MR_HISTORY_BIND_HOST,
+ JHAdminConfig.MR_HISTORY_ADDRESS,
+ JHAdminConfig.DEFAULT_MR_HISTORY_ADDRESS,
server.getListenerAddress());
- LOG.info("Instantiated MRClientService at " + this.bindAddress);
+ LOG.info("Instantiated HistoryClientService at " + this.bindAddress);
super.serviceStart();
}
@@ -158,8 +162,9 @@ public class HistoryClientService extend
JHAdminConfig.MR_WEBAPP_SPNEGO_USER_NAME_KEY)
.at(NetUtils.getHostPortString(bindAddress)).start(webApp);
+ String connectHost = MRWebAppUtil.getJHSWebappURLWithoutScheme(conf).split(":")[0];
MRWebAppUtil.setJHSWebappURLWithoutScheme(conf,
- NetUtils.getHostPortString(webApp.getListenerAddress()));
+ connectHost + ":" + webApp.getListenerAddress().getPort());
}
@Override
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryServerFileSystemStateStoreService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryServerFileSystemStateStoreService.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryServerFileSystemStateStoreService.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryServerFileSystemStateStoreService.java Wed Aug 20 01:34:29 2014
@@ -24,6 +24,8 @@ import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -64,6 +66,7 @@ public class HistoryServerFileSystemStat
private static final String TOKEN_MASTER_KEY_FILE_PREFIX = "key_";
private static final String TOKEN_FILE_PREFIX = "token_";
private static final String TMP_FILE_PREFIX = "tmp-";
+ private static final String UPDATE_TMP_FILE_PREFIX = "update-";
private static final FsPermission DIR_PERMISSIONS =
new FsPermission((short)0700);
private static final FsPermission FILE_PERMISSIONS = Shell.WINDOWS
@@ -90,7 +93,7 @@ public class HistoryServerFileSystemStat
@Override
protected void startStorage() throws IOException {
- fs = rootStatePath.getFileSystem(getConfig());
+ fs = createFileSystem();
createDir(rootStatePath);
tokenStatePath = new Path(rootStatePath, TOKEN_STATE_DIR_NAME);
createDir(tokenStatePath);
@@ -101,6 +104,10 @@ public class HistoryServerFileSystemStat
}
}
+ FileSystem createFileSystem() throws IOException {
+ return rootStatePath.getFileSystem(getConfig());
+ }
+
@Override
protected void closeStorage() throws IOException {
// don't close the filesystem as it's part of the filesystem cache
@@ -127,7 +134,7 @@ public class HistoryServerFileSystemStat
throw new IOException(tokenPath + " already exists");
}
- createFile(tokenPath, buildTokenData(tokenId, renewDate));
+ createNewFile(tokenPath, buildTokenData(tokenId, renewDate));
}
@Override
@@ -136,7 +143,25 @@ public class HistoryServerFileSystemStat
if (LOG.isDebugEnabled()) {
LOG.debug("Updating token " + tokenId.getSequenceNumber());
}
- createFile(getTokenPath(tokenId), buildTokenData(tokenId, renewDate));
+
+ // Files cannot be atomically replaced, therefore we write a temporary
+ // update file, remove the original token file, then rename the update
+ // file to the token file. During recovery either the token file will be
+ // used or if that is missing and an update file is present then the
+ // update file is used.
+ Path tokenPath = getTokenPath(tokenId);
+ Path tmp = new Path(tokenPath.getParent(),
+ UPDATE_TMP_FILE_PREFIX + tokenPath.getName());
+ writeFile(tmp, buildTokenData(tokenId, renewDate));
+ try {
+ deleteFile(tokenPath);
+ } catch (IOException e) {
+ fs.delete(tmp, false);
+ throw e;
+ }
+ if (!fs.rename(tmp, tokenPath)) {
+ throw new IOException("Could not rename " + tmp + " to " + tokenPath);
+ }
}
@Override
@@ -168,7 +193,7 @@ public class HistoryServerFileSystemStat
IOUtils.cleanup(LOG, dataStream);
}
- createFile(keyPath, memStream.toByteArray());
+ createNewFile(keyPath, memStream.toByteArray());
}
@Override
@@ -213,23 +238,33 @@ public class HistoryServerFileSystemStat
}
}
- private void createFile(Path file, byte[] data) throws IOException {
- final int WRITE_BUFFER_SIZE = 4096;
+ private void createNewFile(Path file, byte[] data)
+ throws IOException {
Path tmp = new Path(file.getParent(), TMP_FILE_PREFIX + file.getName());
- FSDataOutputStream out = fs.create(tmp, FILE_PERMISSIONS, true,
- WRITE_BUFFER_SIZE, fs.getDefaultReplication(tmp),
- fs.getDefaultBlockSize(tmp), null);
+ writeFile(tmp, data);
+ try {
+ if (!fs.rename(tmp, file)) {
+ throw new IOException("Could not rename " + tmp + " to " + file);
+ }
+ } catch (IOException e) {
+ fs.delete(tmp, false);
+ throw e;
+ }
+ }
+
+ private void writeFile(Path file, byte[] data) throws IOException {
+ final int WRITE_BUFFER_SIZE = 4096;
+ FSDataOutputStream out = fs.create(file, FILE_PERMISSIONS, true,
+ WRITE_BUFFER_SIZE, fs.getDefaultReplication(file),
+ fs.getDefaultBlockSize(file), null);
try {
try {
out.write(data);
} finally {
IOUtils.cleanup(LOG, out);
}
- if (!fs.rename(tmp, file)) {
- throw new IOException("Could not rename " + tmp + " to " + file);
- }
} catch (IOException e) {
- fs.delete(tmp, false);
+ fs.delete(file, false);
throw e;
}
}
@@ -284,6 +319,19 @@ public class HistoryServerFileSystemStat
state.tokenMasterKeyState.add(key);
}
+ private void loadTokenFromBucket(int bucketId,
+ HistoryServerState state, Path tokenFile, long numTokenFileBytes)
+ throws IOException {
+ MRDelegationTokenIdentifier token =
+ loadToken(state, tokenFile, numTokenFileBytes);
+ int tokenBucketId = getBucketId(token);
+ if (tokenBucketId != bucketId) {
+ throw new IOException("Token " + tokenFile
+ + " should be in bucket " + tokenBucketId + ", found in bucket "
+ + bucketId);
+ }
+ }
+
private MRDelegationTokenIdentifier loadToken(HistoryServerState state,
Path tokenFile, long numTokenFileBytes) throws IOException {
MRDelegationTokenIdentifier tokenId = new MRDelegationTokenIdentifier();
@@ -308,18 +356,29 @@ public class HistoryServerFileSystemStat
final int bucketId = Integer.parseInt(numStr);
int numTokens = 0;
FileStatus[] tokenStats = fs.listStatus(bucket);
+ Set<String> loadedTokens = new HashSet<String>(tokenStats.length);
for (FileStatus stat : tokenStats) {
String name = stat.getPath().getName();
if (name.startsWith(TOKEN_FILE_PREFIX)) {
- MRDelegationTokenIdentifier token =
- loadToken(state, stat.getPath(), stat.getLen());
- int tokenBucketId = getBucketId(token);
- if (tokenBucketId != bucketId) {
- throw new IOException("Token " + stat.getPath()
- + " should be in bucket " + tokenBucketId + ", found in bucket "
- + bucketId);
- }
+ loadTokenFromBucket(bucketId, state, stat.getPath(), stat.getLen());
+ loadedTokens.add(name);
++numTokens;
+ } else if (name.startsWith(UPDATE_TMP_FILE_PREFIX)) {
+ String tokenName = name.substring(UPDATE_TMP_FILE_PREFIX.length());
+ if (loadedTokens.contains(tokenName)) {
+ // already have the token, update may be partial so ignore it
+ fs.delete(stat.getPath(), false);
+ } else {
+ // token is missing, so try to parse the update temp file
+ loadTokenFromBucket(bucketId, state, stat.getPath(), stat.getLen());
+ fs.rename(stat.getPath(),
+ new Path(stat.getPath().getParent(), tokenName));
+ loadedTokens.add(tokenName);
+ ++numTokens;
+ }
+ } else if (name.startsWith(TMP_FILE_PREFIX)) {
+ // cleanup incomplete temp files
+ fs.delete(stat.getPath(), false);
} else {
LOG.warn("Skipping unexpected file in history server token bucket: "
+ stat.getPath());
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistory.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistory.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistory.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/JobHistory.java Wed Aug 20 01:34:29 2014
@@ -394,4 +394,9 @@ public class JobHistory extends Abstract
return true;
}
+ @Override
+ public String getNMHostname() {
+ // bogus - Not Required
+ return null;
+ }
}
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/server/HSAdminServer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/server/HSAdminServer.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/server/HSAdminServer.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/server/HSAdminServer.java Wed Aug 20 01:34:29 2014
@@ -34,6 +34,7 @@ import org.apache.hadoop.security.Groups
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.security.authorize.ProxyUsers;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.logaggregation.AggregatedLogDeletionService;
import org.apache.hadoop.security.proto.RefreshUserMappingsProtocolProtos.RefreshUserMappingsProtocolService;
import org.apache.hadoop.security.protocolPB.RefreshUserMappingsProtocolPB;
@@ -94,7 +95,9 @@ public class HSAdminServer extends Abstr
WritableRpcEngine.ensureInitialized();
- clientRpcAddress = conf.getSocketAddr(JHAdminConfig.JHS_ADMIN_ADDRESS,
+ clientRpcAddress = conf.getSocketAddr(
+ JHAdminConfig.MR_HISTORY_BIND_HOST,
+ JHAdminConfig.JHS_ADMIN_ADDRESS,
JHAdminConfig.DEFAULT_JHS_ADMIN_ADDRESS,
JHAdminConfig.DEFAULT_JHS_ADMIN_PORT);
clientRpcServer = new RPC.Builder(conf)
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/webapp/HsTaskPage.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/webapp/HsTaskPage.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/webapp/HsTaskPage.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/webapp/HsTaskPage.java Wed Aug 20 01:34:29 2014
@@ -149,7 +149,8 @@ public class HsTaskPage extends HsView {
attemptsTableData.append("[\"")
.append(sortId + " ").append(taid).append("\",\"")
.append(ta.getState()).append("\",\"")
- .append(ta.getStatus()).append("\",\"")
+ .append(StringEscapeUtils.escapeJavaScript(
+ StringEscapeUtils.escapeHtml(ta.getStatus()))).append("\",\"")
.append("<a class='nodelink' href='" + MRWebAppUtil.getYARNWebappScheme() + nodeHttpAddr + "'>")
.append(nodeRackName + "/" + nodeHttpAddr + "</a>\",\"")
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryFileManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryFileManager.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryFileManager.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryFileManager.java Wed Aug 20 01:34:29 2014
@@ -19,42 +19,74 @@
package org.apache.hadoop.mapreduce.v2.hs;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.util.UUID;
import org.junit.Assert;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.mapreduce.v2.app.ControlledClock;
import org.apache.hadoop.mapreduce.v2.jobhistory.JHAdminConfig;
+import org.apache.hadoop.test.CoreTestDriver;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.util.SystemClock;
+import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.Rule;
import org.junit.Test;
-
-import java.util.UUID;
+import org.junit.rules.TestName;
public class TestHistoryFileManager {
private static MiniDFSCluster dfsCluster = null;
+ private static MiniDFSCluster dfsCluster2 = null;
+ private static String coreSitePath;
+
+ @Rule
+ public TestName name = new TestName();
@BeforeClass
public static void setUpClass() throws Exception {
+ coreSitePath = "." + File.separator + "target" + File.separator +
+ "test-classes" + File.separator + "core-site.xml";
Configuration conf = new HdfsConfiguration();
+ Configuration conf2 = new HdfsConfiguration();
dfsCluster = new MiniDFSCluster.Builder(conf).build();
+ conf2.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR,
+ conf.get(MiniDFSCluster.HDFS_MINIDFS_BASEDIR) + "_2");
+ dfsCluster2 = new MiniDFSCluster.Builder(conf2).build();
}
@AfterClass
public static void cleanUpClass() throws Exception {
dfsCluster.shutdown();
+ dfsCluster2.shutdown();
+ }
+
+ @After
+ public void cleanTest() throws Exception {
+ new File(coreSitePath).delete();
+ }
+
+ private String getDoneDirNameForTest() {
+ return "/" + name.getMethodName();
+ }
+
+ private String getIntermediateDoneDirNameForTest() {
+ return "/intermediate_" + name.getMethodName();
}
private void testTryCreateHistoryDirs(Configuration conf, boolean expected)
throws Exception {
- conf.set(JHAdminConfig.MR_HISTORY_DONE_DIR, "/" + UUID.randomUUID());
- conf.set(JHAdminConfig.MR_HISTORY_INTERMEDIATE_DONE_DIR, "/" + UUID.randomUUID());
+ conf.set(JHAdminConfig.MR_HISTORY_DONE_DIR, getDoneDirNameForTest());
+ conf.set(JHAdminConfig.MR_HISTORY_INTERMEDIATE_DONE_DIR, getIntermediateDoneDirNameForTest());
HistoryFileManager hfm = new HistoryFileManager();
hfm.conf = conf;
Assert.assertEquals(expected, hfm.tryCreatingHistoryDirs(false));
@@ -76,6 +108,36 @@ public class TestHistoryFileManager {
}
@Test
+ public void testCreateDirsWithAdditionalFileSystem() throws Exception {
+ dfsCluster.getFileSystem().setSafeMode(
+ HdfsConstants.SafeModeAction.SAFEMODE_LEAVE);
+ dfsCluster2.getFileSystem().setSafeMode(
+ HdfsConstants.SafeModeAction.SAFEMODE_LEAVE);
+ Assert.assertFalse(dfsCluster.getFileSystem().isInSafeMode());
+ Assert.assertFalse(dfsCluster2.getFileSystem().isInSafeMode());
+
+ // Set default configuration to the first cluster
+ Configuration conf = new Configuration(false);
+ conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY,
+ dfsCluster.getURI().toString());
+ FileOutputStream os = new FileOutputStream(coreSitePath);
+ conf.writeXml(os);
+ os.close();
+
+ testTryCreateHistoryDirs(dfsCluster2.getConfiguration(0), true);
+
+ // Directories should be created only in the default file system (dfsCluster)
+ Assert.assertTrue(dfsCluster.getFileSystem()
+ .exists(new Path(getDoneDirNameForTest())));
+ Assert.assertTrue(dfsCluster.getFileSystem()
+ .exists(new Path(getIntermediateDoneDirNameForTest())));
+ Assert.assertFalse(dfsCluster2.getFileSystem()
+ .exists(new Path(getDoneDirNameForTest())));
+ Assert.assertFalse(dfsCluster2.getFileSystem()
+ .exists(new Path(getIntermediateDoneDirNameForTest())));
+ }
+
+ @Test
public void testCreateDirsWithFileSystemInSafeMode() throws Exception {
dfsCluster.getFileSystem().setSafeMode(
HdfsConstants.SafeModeAction.SAFEMODE_ENTER);
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryServerFileSystemStateStoreService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryServerFileSystemStateStoreService.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryServerFileSystemStateStoreService.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryServerFileSystemStateStoreService.java Wed Aug 20 01:34:29 2014
@@ -21,12 +21,19 @@ import static org.junit.Assert.assertEqu
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.Mockito.argThat;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.isA;
+import static org.mockito.Mockito.spy;
import java.io.File;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
+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.Text;
import org.apache.hadoop.mapreduce.v2.api.MRDelegationTokenIdentifier;
import org.apache.hadoop.mapreduce.v2.hs.HistoryServerStateStoreService.HistoryServerState;
@@ -35,6 +42,7 @@ import org.apache.hadoop.security.token.
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.ArgumentMatcher;
public class TestHistoryServerFileSystemStateStoreService {
@@ -74,8 +82,8 @@ public class TestHistoryServerFileSystem
return store;
}
- @Test
- public void testTokenStore() throws IOException {
+ private void testTokenStore(String stateStoreUri) throws IOException {
+ conf.set(JHAdminConfig.MR_HS_FS_STATE_STORE_URI, stateStoreUri);
HistoryServerStateStoreService store = createAndStartStore();
HistoryServerState state = store.loadState();
@@ -161,4 +169,77 @@ public class TestHistoryServerFileSystem
assertTrue("missing master key 3",
state.tokenMasterKeyState.contains(key3));
}
+
+ @Test
+ public void testTokenStore() throws IOException {
+ testTokenStore(testDir.getAbsoluteFile().toURI().toString());
+ }
+
+ @Test
+ public void testTokenStoreHdfs() throws IOException {
+ MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build();
+ conf = cluster.getConfiguration(0);
+ try {
+ testTokenStore("/tmp/historystore");
+ } finally {
+ cluster.shutdown();
+ }
+ }
+
+ @Test
+ public void testUpdatedTokenRecovery() throws IOException {
+ IOException intentionalErr = new IOException("intentional error");
+ FileSystem fs = FileSystem.getLocal(conf);
+ final FileSystem spyfs = spy(fs);
+ // make the update token process fail halfway through where we're left
+ // with just the temporary update file and no token file
+ ArgumentMatcher<Path> updateTmpMatcher = new ArgumentMatcher<Path>() {
+ @Override
+ public boolean matches(Object argument) {
+ if (argument instanceof Path) {
+ return ((Path) argument).getName().startsWith("update");
+ }
+ return false;
+ }
+ };
+ doThrow(intentionalErr)
+ .when(spyfs).rename(argThat(updateTmpMatcher), isA(Path.class));
+
+ conf.set(JHAdminConfig.MR_HS_FS_STATE_STORE_URI,
+ testDir.getAbsoluteFile().toURI().toString());
+ HistoryServerStateStoreService store =
+ new HistoryServerFileSystemStateStoreService() {
+ @Override
+ FileSystem createFileSystem() throws IOException {
+ return spyfs;
+ }
+ };
+ store.init(conf);
+ store.start();
+
+ final MRDelegationTokenIdentifier token1 =
+ new MRDelegationTokenIdentifier(new Text("tokenOwner1"),
+ new Text("tokenRenewer1"), new Text("tokenUser1"));
+ token1.setSequenceNumber(1);
+ final Long tokenDate1 = 1L;
+ store.storeToken(token1, tokenDate1);
+ final Long newTokenDate1 = 975318642L;
+ try {
+ store.updateToken(token1, newTokenDate1);
+ fail("intentional error not thrown");
+ } catch (IOException e) {
+ assertEquals(intentionalErr, e);
+ }
+ store.close();
+
+ // verify the update file is seen and parsed upon recovery when
+ // original token file is missing
+ store = createAndStartStore();
+ HistoryServerState state = store.loadState();
+ assertEquals("incorrect loaded token count", 1, state.tokenState.size());
+ assertTrue("missing token 1", state.tokenState.containsKey(token1));
+ assertEquals("incorrect token 1 date", newTokenDate1,
+ state.tokenState.get(token1));
+ store.close();
+ }
}
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestBlocks.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestBlocks.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestBlocks.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/webapp/TestBlocks.java Wed Aug 20 01:34:29 2014
@@ -159,7 +159,7 @@ public class TestBlocks {
when(taReport.getSortFinishTime()).thenReturn(taSortFinishTime);
when(taReport.getContainerId()).thenReturn(containerId);
when(taReport.getProgress()).thenReturn(1.0f);
- when(taReport.getStateString()).thenReturn("Processed 128/128 records");
+ when(taReport.getStateString()).thenReturn("Processed 128/128 records <p> \n");
when(taReport.getTaskAttemptState()).thenReturn(taState);
when(taReport.getDiagnosticInfo()).thenReturn("");
@@ -184,6 +184,8 @@ public class TestBlocks {
// should be printed information about attempts
assertTrue(data.toString().contains("0 attempt_0_0001_r_000000_0"));
assertTrue(data.toString().contains("SUCCEEDED"));
+ assertFalse(data.toString().contains("Processed 128/128 records <p> \n"));
+ assertTrue(data.toString().contains("Processed 128\\/128 records <p> \\n"));
assertTrue(data.toString().contains(
"_0005_01_000001:attempt_0_0001_r_000000_0:User:"));
assertTrue(data.toString().contains("100002"));
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/YARNRunner.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/YARNRunner.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/YARNRunner.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/main/java/org/apache/hadoop/mapred/YARNRunner.java Wed Aug 20 01:34:29 2014
@@ -447,6 +447,15 @@ public class YARNRunner implements Clien
Map<String, String> environment = new HashMap<String, String>();
MRApps.setClasspath(environment, conf);
+ // Shell
+ environment.put(Environment.SHELL.name(),
+ conf.get(MRJobConfig.MAPRED_ADMIN_USER_SHELL,
+ MRJobConfig.DEFAULT_SHELL));
+
+ // Add the container working directory at the front of LD_LIBRARY_PATH
+ MRApps.addToEnvironment(environment, Environment.LD_LIBRARY_PATH.name(),
+ MRApps.crossPlatformifyMREnv(conf, Environment.PWD), conf);
+
// Setup the environment variables for Admin first
MRApps.setEnvFromInputString(environment,
conf.get(MRJobConfig.MR_AM_ADMIN_USER_ENV), conf);
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestYARNRunner.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestYARNRunner.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestYARNRunner.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestYARNRunner.java Wed Aug 20 01:34:29 2014
@@ -29,6 +29,7 @@ import static org.mockito.Mockito.when;
import java.io.ByteArrayOutputStream;
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
@@ -36,6 +37,7 @@ import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.security.PrivilegedExceptionAction;
import java.util.List;
+import java.util.Map;
import junit.framework.TestCase;
@@ -44,22 +46,27 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.JobPriority;
import org.apache.hadoop.mapreduce.JobStatus.State;
+import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
import org.apache.hadoop.mapreduce.v2.api.MRDelegationTokenIdentifier;
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetDelegationTokenRequest;
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetDelegationTokenResponse;
+import org.apache.hadoop.mapreduce.v2.util.MRApps;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
@@ -94,6 +101,7 @@ import org.apache.log4j.Layout;
import org.apache.log4j.Logger;
import org.apache.log4j.SimpleLayout;
import org.apache.log4j.WriterAppender;
+import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
@@ -146,8 +154,12 @@ public class TestYARNRunner extends Test
FileContext.getLocalFSFileContext().delete(new Path(testWorkDir.toString()), true);
}
testWorkDir.mkdirs();
- }
+ }
+ @After
+ public void cleanup() {
+ FileUtil.fullyDelete(testWorkDir);
+ }
@Test(timeout=20000)
public void testJobKill() throws Exception {
@@ -397,21 +409,8 @@ public class TestYARNRunner extends Test
YARNRunner yarnRunner = new YARNRunner(jobConf);
- File jobxml = new File(testWorkDir, MRJobConfig.JOB_CONF_FILE);
- OutputStream out = new FileOutputStream(jobxml);
- conf.writeXml(out);
- out.close();
-
- File jobsplit = new File(testWorkDir, MRJobConfig.JOB_SPLIT);
- out = new FileOutputStream(jobsplit);
- out.close();
-
- File jobsplitmetainfo = new File(testWorkDir, MRJobConfig.JOB_SPLIT_METAINFO);
- out = new FileOutputStream(jobsplitmetainfo);
- out.close();
-
- ApplicationSubmissionContext submissionContext =
- yarnRunner.createApplicationSubmissionContext(jobConf, testWorkDir.toString(), new Credentials());
+ ApplicationSubmissionContext submissionContext =
+ buildSubmitContext(yarnRunner, jobConf);
ContainerLaunchContext containerSpec = submissionContext.getAMContainerSpec();
List<String> commands = containerSpec.getCommands();
@@ -463,22 +462,9 @@ public class TestYARNRunner extends Test
YARNRunner yarnRunner = new YARNRunner(jobConf);
- File jobxml = new File(testWorkDir, MRJobConfig.JOB_CONF_FILE);
- OutputStream out = new FileOutputStream(jobxml);
- conf.writeXml(out);
- out.close();
-
- File jobsplit = new File(testWorkDir, MRJobConfig.JOB_SPLIT);
- out = new FileOutputStream(jobsplit);
- out.close();
-
- File jobsplitmetainfo = new File(testWorkDir, MRJobConfig.JOB_SPLIT_METAINFO);
- out = new FileOutputStream(jobsplitmetainfo);
- out.close();
-
@SuppressWarnings("unused")
- ApplicationSubmissionContext submissionContext =
- yarnRunner.createApplicationSubmissionContext(jobConf, testWorkDir.toString(), new Credentials());
+ ApplicationSubmissionContext submissionContext =
+ buildSubmitContext(yarnRunner, jobConf);
String logMsg = bout.toString();
assertTrue(logMsg.contains("WARN - Usage of -Djava.library.path in " +
@@ -492,4 +478,60 @@ public class TestYARNRunner extends Test
"be set as part of the LD_LIBRARY_PATH in the app master JVM env " +
"using yarn.app.mapreduce.am.env config settings."));
}
+
+ @Test
+ public void testAMStandardEnv() throws Exception {
+ final String ADMIN_LIB_PATH = "foo";
+ final String USER_LIB_PATH = "bar";
+ final String USER_SHELL = "shell";
+ JobConf jobConf = new JobConf();
+
+ jobConf.set(MRJobConfig.MR_AM_ADMIN_USER_ENV, "LD_LIBRARY_PATH=" +
+ ADMIN_LIB_PATH);
+ jobConf.set(MRJobConfig.MR_AM_ENV, "LD_LIBRARY_PATH="
+ + USER_LIB_PATH);
+ jobConf.set(MRJobConfig.MAPRED_ADMIN_USER_SHELL, USER_SHELL);
+
+ YARNRunner yarnRunner = new YARNRunner(jobConf);
+ ApplicationSubmissionContext appSubCtx =
+ buildSubmitContext(yarnRunner, jobConf);
+
+ // make sure PWD is first in the lib path
+ ContainerLaunchContext clc = appSubCtx.getAMContainerSpec();
+ Map<String, String> env = clc.getEnvironment();
+ String libPath = env.get(Environment.LD_LIBRARY_PATH.name());
+ assertNotNull("LD_LIBRARY_PATH not set", libPath);
+ String cps = jobConf.getBoolean(
+ MRConfig.MAPREDUCE_APP_SUBMISSION_CROSS_PLATFORM,
+ MRConfig.DEFAULT_MAPREDUCE_APP_SUBMISSION_CROSS_PLATFORM)
+ ? ApplicationConstants.CLASS_PATH_SEPARATOR : File.pathSeparator;
+ assertEquals("Bad AM LD_LIBRARY_PATH setting",
+ MRApps.crossPlatformifyMREnv(conf, Environment.PWD)
+ + cps + ADMIN_LIB_PATH + cps + USER_LIB_PATH, libPath);
+
+ // make sure SHELL is set
+ String shell = env.get(Environment.SHELL.name());
+ assertNotNull("SHELL not set", shell);
+ assertEquals("Bad SHELL setting", USER_SHELL, shell);
+ }
+
+ private ApplicationSubmissionContext buildSubmitContext(
+ YARNRunner yarnRunner, JobConf jobConf) throws IOException {
+ File jobxml = new File(testWorkDir, MRJobConfig.JOB_CONF_FILE);
+ OutputStream out = new FileOutputStream(jobxml);
+ conf.writeXml(out);
+ out.close();
+
+ File jobsplit = new File(testWorkDir, MRJobConfig.JOB_SPLIT);
+ out = new FileOutputStream(jobsplit);
+ out.close();
+
+ File jobsplitmetainfo = new File(testWorkDir,
+ MRJobConfig.JOB_SPLIT_METAINFO);
+ out = new FileOutputStream(jobsplitmetainfo);
+ out.close();
+
+ return yarnRunner.createApplicationSubmissionContext(jobConf,
+ testWorkDir.toString(), new Credentials());
+ }
}
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleHandler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleHandler.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleHandler.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/main/java/org/apache/hadoop/mapred/ShuffleHandler.java Wed Aug 20 01:34:29 2014
@@ -82,13 +82,13 @@ import org.apache.hadoop.security.token.
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.NMDBSchemaVersionProto;
+import org.apache.hadoop.yarn.proto.YarnServerCommonProtos.VersionProto;
import org.apache.hadoop.yarn.server.api.ApplicationInitializationContext;
import org.apache.hadoop.yarn.server.api.ApplicationTerminationContext;
import org.apache.hadoop.yarn.server.api.AuxiliaryService;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
-import org.apache.hadoop.yarn.server.nodemanager.recovery.records.NMDBSchemaVersion;
-import org.apache.hadoop.yarn.server.nodemanager.recovery.records.impl.pb.NMDBSchemaVersionPBImpl;
+import org.apache.hadoop.yarn.server.records.Version;
+import org.apache.hadoop.yarn.server.records.impl.pb.VersionPBImpl;
import org.apache.hadoop.yarn.server.utils.LeveldbIterator;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.fusesource.leveldbjni.JniDBFactory;
@@ -151,8 +151,8 @@ public class ShuffleHandler extends Auxi
private static final String STATE_DB_NAME = "mapreduce_shuffle_state";
private static final String STATE_DB_SCHEMA_VERSION_KEY = "shuffle-schema-version";
- protected static final NMDBSchemaVersion CURRENT_VERSION_INFO =
- NMDBSchemaVersion.newInstance(1, 0);
+ protected static final Version CURRENT_VERSION_INFO =
+ Version.newInstance(1, 0);
private int port;
private ChannelFactory selector;
@@ -491,21 +491,21 @@ public class ShuffleHandler extends Auxi
}
@VisibleForTesting
- NMDBSchemaVersion loadVersion() throws IOException {
+ Version loadVersion() throws IOException {
byte[] data = stateDb.get(bytes(STATE_DB_SCHEMA_VERSION_KEY));
// if version is not stored previously, treat it as 1.0.
if (data == null || data.length == 0) {
- return NMDBSchemaVersion.newInstance(1, 0);
+ return Version.newInstance(1, 0);
}
- NMDBSchemaVersion version =
- new NMDBSchemaVersionPBImpl(NMDBSchemaVersionProto.parseFrom(data));
+ Version version =
+ new VersionPBImpl(VersionProto.parseFrom(data));
return version;
}
- private void storeSchemaVersion(NMDBSchemaVersion version) throws IOException {
+ private void storeSchemaVersion(Version version) throws IOException {
String key = STATE_DB_SCHEMA_VERSION_KEY;
byte[] data =
- ((NMDBSchemaVersionPBImpl) version).getProto().toByteArray();
+ ((VersionPBImpl) version).getProto().toByteArray();
try {
stateDb.put(bytes(key), data);
} catch (DBException e) {
@@ -519,11 +519,11 @@ public class ShuffleHandler extends Auxi
// Only used for test
@VisibleForTesting
- void storeVersion(NMDBSchemaVersion version) throws IOException {
+ void storeVersion(Version version) throws IOException {
storeSchemaVersion(version);
}
- protected NMDBSchemaVersion getCurrentVersion() {
+ protected Version getCurrentVersion() {
return CURRENT_VERSION_INFO;
}
@@ -538,7 +538,7 @@ public class ShuffleHandler extends Auxi
* upgrade shuffle info or remove incompatible old state.
*/
private void checkVersion() throws IOException {
- NMDBSchemaVersion loadedVersion = loadVersion();
+ Version loadedVersion = loadVersion();
LOG.info("Loaded state DB schema version info " + loadedVersion);
if (loadedVersion.equals(getCurrentVersion())) {
return;
Modified: hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/java/org/apache/hadoop/mapred/TestShuffleHandler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/java/org/apache/hadoop/mapred/TestShuffleHandler.java?rev=1619019&r1=1619018&r2=1619019&view=diff
==============================================================================
--- hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/java/org/apache/hadoop/mapred/TestShuffleHandler.java (original)
+++ hadoop/common/branches/YARN-1051/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-shuffle/src/test/java/org/apache/hadoop/mapred/TestShuffleHandler.java Wed Aug 20 01:34:29 2014
@@ -75,7 +75,7 @@ import org.apache.hadoop.yarn.conf.YarnC
import org.apache.hadoop.yarn.server.api.ApplicationInitializationContext;
import org.apache.hadoop.yarn.server.api.ApplicationTerminationContext;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
-import org.apache.hadoop.yarn.server.nodemanager.recovery.records.NMDBSchemaVersion;
+import org.apache.hadoop.yarn.server.records.Version;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelHandlerContext;
@@ -764,11 +764,11 @@ public class TestShuffleHandler {
// verify we are still authorized to shuffle to the old application
rc = getShuffleResponseCode(shuffle, jt);
Assert.assertEquals(HttpURLConnection.HTTP_OK, rc);
- NMDBSchemaVersion version = NMDBSchemaVersion.newInstance(1, 0);
+ Version version = Version.newInstance(1, 0);
Assert.assertEquals(version, shuffle.getCurrentVersion());
// emulate shuffle handler restart with compatible version
- NMDBSchemaVersion version11 = NMDBSchemaVersion.newInstance(1, 1);
+ Version version11 = Version.newInstance(1, 1);
// update version info before close shuffle
shuffle.storeVersion(version11);
Assert.assertEquals(version11, shuffle.loadVersion());
@@ -785,7 +785,7 @@ public class TestShuffleHandler {
Assert.assertEquals(HttpURLConnection.HTTP_OK, rc);
// emulate shuffle handler restart with incompatible version
- NMDBSchemaVersion version21 = NMDBSchemaVersion.newInstance(2, 1);
+ Version version21 = Version.newInstance(2, 1);
shuffle.storeVersion(version21);
Assert.assertEquals(version21, shuffle.loadVersion());
shuffle.close();