Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

MAPREDUCE-3808. Fixed an NPE in FileOutputCommitter for jobs with maps but no reduces. Contributed by Robert Joseph Evans. 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1241217 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Vinod Kumar Vavilapalli 2012-02-06 22:16:26 +00:00
parent c6282df3e5
commit 9f2b77aee4
4 changed files with 79 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-mapreduce-project/CHANGES.txt
View File

@ -715,6 +715,9 @@ Release 0.23.1 - Unreleased
MAPREDUCE-3813. Added a cache for resolved racks. (vinodkv via acmurthy)
MAPREDUCE-3808. Fixed an NPE in FileOutputCommitter for jobs with maps
but no reduces. (Robert Joseph Evans via vinodkv)
Release 0.23.0 - 2011-11-01
INCOMPATIBLE CHANGES

21
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/FileOutputCommitter.java
View File

@ -85,18 +85,21 @@ public class FileOutputCommitter extends OutputCommitter {
*/
@Private
Path getJobAttemptPath(JobContext context) {
return org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
.getJobAttemptPath(context, getOutputPath(context));
Path out = getOutputPath(context);
return out == null ? null :
org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
.getJobAttemptPath(context, out);
}
@Private
Path getTaskAttemptPath(TaskAttemptContext context) throws IOException {
return getTaskAttemptPath(context, getOutputPath(context));
Path out = getOutputPath(context);
return out == null ? null : getTaskAttemptPath(context, out);
}
private Path getTaskAttemptPath(TaskAttemptContext context, Path out) throws IOException {
Path workPath = FileOutputFormat.getWorkOutputPath(context.getJobConf());
if(workPath == null) {
if(workPath == null && out != null) {
return org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
.getTaskAttemptPath(context, out);
}
@ -110,14 +113,17 @@ public class FileOutputCommitter extends OutputCommitter {
* @return the path where the output of a committed task is stored until
* the entire job is committed.
*/
@Private
Path getCommittedTaskPath(TaskAttemptContext context) {
return org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
.getCommittedTaskPath(context, getOutputPath(context));
Path out = getOutputPath(context);
return out == null ? null :
org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
.getCommittedTaskPath(context, out);
}
public Path getWorkPath(TaskAttemptContext context, Path outputPath)
throws IOException {
return getTaskAttemptPath(context, outputPath);
return outputPath == null ? null : getTaskAttemptPath(context, outputPath);
}
@Override
@ -156,6 +162,7 @@ public class FileOutputCommitter extends OutputCommitter {
getWrapped(context).abortJob(context, state);
}
@Override
public void setupTask(TaskAttemptContext context) throws IOException {
getWrapped(context).setupTask(context);
}

4
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java
View File

@ -495,6 +495,7 @@ public class FileOutputCommitter extends OutputCommitter {
@Override
public void recoverTask(TaskAttemptContext context)
throws IOException {
if(hasOutputPath()) {
context.progress();
TaskAttemptID attemptId = context.getTaskAttemptID();
int previousAttempt = getAppAttemptId(context) - 1;
@ -526,5 +527,8 @@ public class FileOutputCommitter extends OutputCommitter {
} else {
LOG.warn(attemptId+" had no output to recover.");
}
} else {
LOG.warn("Output Path is null in recoverTask()");
}
}
}

28
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapred/TestFileOutputCommitter.java
View File

@ -104,7 +104,9 @@ public class TestFileOutputCommitter extends TestCase {
writeOutput(theRecordWriter, tContext);
// do commit
if(committer.needsTaskCommit(tContext)) {
committer.commitTask(tContext);
}
Path jobTempDir1 = committer.getCommittedTaskPath(tContext);
File jtd1 = new File(jobTempDir1.toUri().getPath());
assertTrue(jtd1.exists());
@ -188,7 +190,9 @@ public class TestFileOutputCommitter extends TestCase {
writeOutput(theRecordWriter, tContext);
// do commit
if(committer.needsTaskCommit(tContext)) {
committer.commitTask(tContext);
}
committer.commitJob(jContext);
// validate output
@ -214,7 +218,9 @@ public class TestFileOutputCommitter extends TestCase {
writeMapFileOutput(theRecordWriter, tContext);
// do commit
if(committer.needsTaskCommit(tContext)) {
committer.commitTask(tContext);
}
committer.commitJob(jContext);
// validate output
@ -222,6 +228,28 @@ public class TestFileOutputCommitter extends TestCase {
FileUtil.fullyDelete(new File(outDir.toString()));
}
public void testMapOnlyNoOutput() throws Exception {
JobConf conf = new JobConf();
//This is not set on purpose. FileOutputFormat.setOutputPath(conf, outDir);
conf.set(JobContext.TASK_ATTEMPT_ID, attempt);
JobContext jContext = new JobContextImpl(conf, taskID.getJobID());
TaskAttemptContext tContext = new TaskAttemptContextImpl(conf, taskID);
FileOutputCommitter committer = new FileOutputCommitter();
// setup
committer.setupJob(jContext);
committer.setupTask(tContext);
if(committer.needsTaskCommit(tContext)) {
// do commit
committer.commitTask(tContext);
}
committer.commitJob(jContext);
// validate output
FileUtil.fullyDelete(new File(outDir.toString()));
}
public void testAbort() throws IOException, InterruptedException {
JobConf conf = new JobConf();
FileOutputFormat.setOutputPath(conf, outDir);