|
|
@@ -18,7 +18,10 @@
|
|
|
|
|
|
package org.apache.hadoop.mapreduce.v2.hs;
|
|
|
|
|
|
-import static junit.framework.Assert.assertEquals;
|
|
|
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic
|
|
|
+ .NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY;
|
|
|
+import static org.junit.Assert.assertEquals;
|
|
|
+import static org.junit.Assert.assertNotNull;
|
|
|
import static org.junit.Assert.assertNull;
|
|
|
import static org.junit.Assert.assertTrue;
|
|
|
|
|
|
@@ -26,6 +29,7 @@ import java.io.ByteArrayOutputStream;
|
|
|
import java.io.IOException;
|
|
|
import java.io.PrintStream;
|
|
|
import java.util.Arrays;
|
|
|
+import java.util.Collections;
|
|
|
import java.util.HashMap;
|
|
|
import java.util.List;
|
|
|
import java.util.Map;
|
|
|
@@ -37,9 +41,9 @@ import junit.framework.Assert;
|
|
|
import org.apache.commons.logging.Log;
|
|
|
import org.apache.commons.logging.LogFactory;
|
|
|
import org.apache.hadoop.conf.Configuration;
|
|
|
-import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
|
|
|
import org.apache.hadoop.fs.FSDataInputStream;
|
|
|
import org.apache.hadoop.fs.FileContext;
|
|
|
+import org.apache.hadoop.fs.FileSystem;
|
|
|
import org.apache.hadoop.fs.Path;
|
|
|
import org.apache.hadoop.mapreduce.Counters;
|
|
|
import org.apache.hadoop.mapreduce.JobID;
|
|
|
@@ -54,6 +58,7 @@ import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.AMInfo;
|
|
|
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.JobInfo;
|
|
|
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskAttemptInfo;
|
|
|
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskInfo;
|
|
|
+import org.apache.hadoop.mapreduce.jobhistory.JobUnsuccessfulCompletionEvent;
|
|
|
import org.apache.hadoop.mapreduce.jobhistory.TaskFailedEvent;
|
|
|
import org.apache.hadoop.mapreduce.jobhistory.TaskFinishedEvent;
|
|
|
import org.apache.hadoop.mapreduce.jobhistory.TaskStartedEvent;
|
|
|
@@ -67,8 +72,11 @@ import org.apache.hadoop.mapreduce.v2.api.records.impl.pb.JobIdPBImpl;
|
|
|
import org.apache.hadoop.mapreduce.v2.api.records.impl.pb.TaskIdPBImpl;
|
|
|
import org.apache.hadoop.mapreduce.v2.app.MRApp;
|
|
|
import org.apache.hadoop.mapreduce.v2.app.job.Job;
|
|
|
+import org.apache.hadoop.mapreduce.v2.app.job.impl.JobImpl;
|
|
|
import org.apache.hadoop.mapreduce.v2.app.job.Task;
|
|
|
import org.apache.hadoop.mapreduce.v2.app.job.TaskAttempt;
|
|
|
+import org.apache.hadoop.mapreduce.v2.app.job.event.JobEvent;
|
|
|
+import org.apache.hadoop.mapreduce.v2.app.job.event.JobEventType;
|
|
|
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
|
|
|
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
|
|
|
import org.apache.hadoop.mapreduce.v2.hs.HistoryFileManager.HistoryFileInfo;
|
|
|
@@ -150,7 +158,7 @@ public class TestJobHistoryParsing {
|
|
|
conf.set(MRJobConfig.USER_NAME, System.getProperty("user.name"));
|
|
|
long amStartTimeEst = System.currentTimeMillis();
|
|
|
conf.setClass(
|
|
|
- CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
MyResolver.class, DNSToSwitchMapping.class);
|
|
|
RackResolver.init(conf);
|
|
|
MRApp app = new MRAppWithHistory(numMaps, numReduces, true, this.getClass()
|
|
|
@@ -391,7 +399,7 @@ public class TestJobHistoryParsing {
|
|
|
try {
|
|
|
Configuration conf = new Configuration();
|
|
|
conf.setClass(
|
|
|
- CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
MyResolver.class, DNSToSwitchMapping.class);
|
|
|
RackResolver.init(conf);
|
|
|
MRApp app = new MRAppWithHistoryWithFailedAttempt(2, 1, true, this
|
|
|
@@ -456,7 +464,7 @@ public class TestJobHistoryParsing {
|
|
|
try {
|
|
|
Configuration conf = new Configuration();
|
|
|
conf.setClass(
|
|
|
- CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
MyResolver.class, DNSToSwitchMapping.class);
|
|
|
RackResolver.init(conf);
|
|
|
MRApp app = new MRAppWithHistoryWithFailedTask(2, 1, true, this
|
|
|
@@ -500,18 +508,85 @@ public class TestJobHistoryParsing {
|
|
|
Assert.assertNotNull("completed task report has null counters", ct
|
|
|
.getReport().getCounters());
|
|
|
}
|
|
|
+ final List<String> originalDiagnostics = job.getDiagnostics();
|
|
|
+ final String historyError = jobInfo.getErrorInfo();
|
|
|
+ assertTrue("No original diagnostics for a failed job",
|
|
|
+ originalDiagnostics != null && !originalDiagnostics.isEmpty());
|
|
|
+ assertNotNull("No history error info for a failed job ", historyError);
|
|
|
+ for (String diagString : originalDiagnostics) {
|
|
|
+ assertTrue(historyError.contains(diagString));
|
|
|
+ }
|
|
|
} finally {
|
|
|
LOG.info("FINISHED testCountersForFailedTask");
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ @Test(timeout = 60000)
|
|
|
+ public void testDiagnosticsForKilledJob() throws Exception {
|
|
|
+ LOG.info("STARTING testDiagnosticsForKilledJob");
|
|
|
+ try {
|
|
|
+ final Configuration conf = new Configuration();
|
|
|
+ conf.setClass(
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ MyResolver.class, DNSToSwitchMapping.class);
|
|
|
+ RackResolver.init(conf);
|
|
|
+ MRApp app = new MRAppWithHistoryWithJobKilled(2, 1, true, this
|
|
|
+ .getClass().getName(), true);
|
|
|
+ app.submit(conf);
|
|
|
+ Job job = app.getContext().getAllJobs().values().iterator().next();
|
|
|
+ JobId jobId = job.getID();
|
|
|
+ app.waitForState(job, JobState.KILLED);
|
|
|
+
|
|
|
+ // make sure all events are flushed
|
|
|
+ app.waitForState(Service.STATE.STOPPED);
|
|
|
+
|
|
|
+ JobHistory jobHistory = new JobHistory();
|
|
|
+ jobHistory.init(conf);
|
|
|
+
|
|
|
+ HistoryFileInfo fileInfo = jobHistory.getJobFileInfo(jobId);
|
|
|
+
|
|
|
+ JobHistoryParser parser;
|
|
|
+ JobInfo jobInfo;
|
|
|
+ synchronized (fileInfo) {
|
|
|
+ Path historyFilePath = fileInfo.getHistoryFile();
|
|
|
+ FSDataInputStream in = null;
|
|
|
+ FileContext fc = null;
|
|
|
+ try {
|
|
|
+ fc = FileContext.getFileContext(conf);
|
|
|
+ in = fc.open(fc.makeQualified(historyFilePath));
|
|
|
+ } catch (IOException ioe) {
|
|
|
+ LOG.info("Can not open history file: " + historyFilePath, ioe);
|
|
|
+ throw (new Exception("Can not open History File"));
|
|
|
+ }
|
|
|
+
|
|
|
+ parser = new JobHistoryParser(in);
|
|
|
+ jobInfo = parser.parse();
|
|
|
+ }
|
|
|
+ Exception parseException = parser.getParseException();
|
|
|
+ assertNull("Caught an expected exception " + parseException,
|
|
|
+ parseException);
|
|
|
+ final List<String> originalDiagnostics = job.getDiagnostics();
|
|
|
+ final String historyError = jobInfo.getErrorInfo();
|
|
|
+ assertTrue("No original diagnostics for a failed job",
|
|
|
+ originalDiagnostics != null && !originalDiagnostics.isEmpty());
|
|
|
+ assertNotNull("No history error info for a failed job ", historyError);
|
|
|
+ for (String diagString : originalDiagnostics) {
|
|
|
+ assertTrue(historyError.contains(diagString));
|
|
|
+ }
|
|
|
+ assertTrue("No killed message in diagnostics",
|
|
|
+ historyError.contains(JobImpl.JOB_KILLED_DIAG));
|
|
|
+ } finally {
|
|
|
+ LOG.info("FINISHED testDiagnosticsForKilledJob");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
@Test(timeout = 50000)
|
|
|
public void testScanningOldDirs() throws Exception {
|
|
|
LOG.info("STARTING testScanningOldDirs");
|
|
|
try {
|
|
|
Configuration conf = new Configuration();
|
|
|
conf.setClass(
|
|
|
- CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
MyResolver.class, DNSToSwitchMapping.class);
|
|
|
RackResolver.init(conf);
|
|
|
MRApp app = new MRAppWithHistory(1, 1, true, this.getClass().getName(),
|
|
|
@@ -591,6 +666,27 @@ public class TestJobHistoryParsing {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ static class MRAppWithHistoryWithJobKilled extends MRAppWithHistory {
|
|
|
+
|
|
|
+ public MRAppWithHistoryWithJobKilled(int maps, int reduces,
|
|
|
+ boolean autoComplete, String testName, boolean cleanOnStart) {
|
|
|
+ super(maps, reduces, autoComplete, testName, cleanOnStart);
|
|
|
+ }
|
|
|
+
|
|
|
+ @SuppressWarnings("unchecked")
|
|
|
+ @Override
|
|
|
+ protected void attemptLaunched(TaskAttemptId attemptID) {
|
|
|
+ if (attemptID.getTaskId().getId() == 0) {
|
|
|
+ getContext().getEventHandler().handle(
|
|
|
+ new JobEvent(attemptID.getTaskId().getJobId(),
|
|
|
+ JobEventType.JOB_KILL));
|
|
|
+ } else {
|
|
|
+ getContext().getEventHandler().handle(
|
|
|
+ new TaskAttemptEvent(attemptID, TaskAttemptEventType.TA_DONE));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
static class HistoryFileManagerForTest extends HistoryFileManager {
|
|
|
void deleteJobFromJobListCache(HistoryFileInfo fileInfo) {
|
|
|
jobListCache.delete(fileInfo);
|
|
|
@@ -613,7 +709,7 @@ public class TestJobHistoryParsing {
|
|
|
try {
|
|
|
Configuration conf = new Configuration();
|
|
|
conf.setClass(
|
|
|
- CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
MyResolver.class, DNSToSwitchMapping.class);
|
|
|
|
|
|
RackResolver.init(conf);
|
|
|
@@ -667,7 +763,7 @@ public class TestJobHistoryParsing {
|
|
|
Configuration configuration = new Configuration();
|
|
|
configuration
|
|
|
.setClass(
|
|
|
- CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
+ NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
|
|
MyResolver.class, DNSToSwitchMapping.class);
|
|
|
|
|
|
RackResolver.init(configuration);
|
|
|
@@ -739,7 +835,7 @@ public class TestJobHistoryParsing {
|
|
|
final org.apache.hadoop.mapreduce.TaskType taskType =
|
|
|
org.apache.hadoop.mapreduce.TaskType.MAP;
|
|
|
final TaskID[] tids = new TaskID[2];
|
|
|
- JobID jid = new JobID("1", 1);
|
|
|
+ final JobID jid = new JobID("1", 1);
|
|
|
tids[0] = new TaskID(jid, taskType, 0);
|
|
|
tids[1] = new TaskID(jid, taskType, 1);
|
|
|
Mockito.when(reader.getNextEvent()).thenAnswer(
|
|
|
@@ -758,6 +854,13 @@ public class TestJobHistoryParsing {
|
|
|
tfe.setDatum(tfe.getDatum());
|
|
|
return tfe;
|
|
|
}
|
|
|
+ if (eventId < 5) {
|
|
|
+ JobUnsuccessfulCompletionEvent juce =
|
|
|
+ new JobUnsuccessfulCompletionEvent(jid, 100L, 2, 0,
|
|
|
+ "JOB_FAILED", Collections.singletonList(
|
|
|
+ "Task failed: " + tids[0].toString()));
|
|
|
+ return juce;
|
|
|
+ }
|
|
|
return null;
|
|
|
}
|
|
|
});
|
|
|
@@ -765,4 +868,22 @@ public class TestJobHistoryParsing {
|
|
|
assertTrue("Task 0 not implicated",
|
|
|
info.getErrorInfo().contains(tids[0].toString()));
|
|
|
}
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void testFailedJobHistoryWithoutDiagnostics() throws Exception {
|
|
|
+ final Path histPath = new Path(getClass().getClassLoader().getResource(
|
|
|
+ "job_1393307629410_0001-1393307687476-user-Sleep+job-1393307723835-0-0-FAILED-default-1393307693920.jhist")
|
|
|
+ .getFile());
|
|
|
+ final FileSystem lfs = FileSystem.getLocal(new Configuration());
|
|
|
+ final FSDataInputStream fsdis = lfs.open(histPath);
|
|
|
+ try {
|
|
|
+ JobHistoryParser parser = new JobHistoryParser(fsdis);
|
|
|
+ JobInfo info = parser.parse();
|
|
|
+ assertEquals("History parsed jobId incorrectly",
|
|
|
+ info.getJobId(), JobID.forName("job_1393307629410_0001") );
|
|
|
+ assertEquals("Default diagnostics incorrect ", "", info.getErrorInfo());
|
|
|
+ } finally {
|
|
|
+ fsdis.close();
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
Jason Darrell Lowe