MAPREDUCE-5956. Made MR AM not use maxAttempts to determine if the current attempt is the last retry. Contributed by Wangda Tan.
svn merge --ignore-ancestry -c 1609649 ../../trunk/ git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1609650 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
e4f494a767
commit
bba266610c
|
@ -15,6 +15,9 @@ Release 2.6.0 - UNRELEASED
|
||||||
MAPREDUCE-5866. TestFixedLengthInputFormat fails in windows.
|
MAPREDUCE-5866. TestFixedLengthInputFormat fails in windows.
|
||||||
(Varun Vasudev via cnauroth)
|
(Varun Vasudev via cnauroth)
|
||||||
|
|
||||||
|
MAPREDUCE-5956. Made MR AM not use maxAttempts to determine if the current
|
||||||
|
attempt is the last retry. (Wangda Tan via zjshen)
|
||||||
|
|
||||||
Release 2.5.0 - UNRELEASED
|
Release 2.5.0 - UNRELEASED
|
||||||
|
|
||||||
INCOMPATIBLE CHANGES
|
INCOMPATIBLE CHANGES
|
||||||
|
|
|
@ -184,7 +184,6 @@ public class MRAppMaster extends CompositeService {
|
||||||
private final int nmPort;
|
private final int nmPort;
|
||||||
private final int nmHttpPort;
|
private final int nmHttpPort;
|
||||||
protected final MRAppMetrics metrics;
|
protected final MRAppMetrics metrics;
|
||||||
private final int maxAppAttempts;
|
|
||||||
private Map<TaskId, TaskInfo> completedTasksFromPreviousRun;
|
private Map<TaskId, TaskInfo> completedTasksFromPreviousRun;
|
||||||
private List<AMInfo> amInfos;
|
private List<AMInfo> amInfos;
|
||||||
private AppContext context;
|
private AppContext context;
|
||||||
|
@ -224,14 +223,14 @@ public class MRAppMaster extends CompositeService {
|
||||||
|
|
||||||
public MRAppMaster(ApplicationAttemptId applicationAttemptId,
|
public MRAppMaster(ApplicationAttemptId applicationAttemptId,
|
||||||
ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
|
ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
|
||||||
long appSubmitTime, int maxAppAttempts) {
|
long appSubmitTime) {
|
||||||
this(applicationAttemptId, containerId, nmHost, nmPort, nmHttpPort,
|
this(applicationAttemptId, containerId, nmHost, nmPort, nmHttpPort,
|
||||||
new SystemClock(), appSubmitTime, maxAppAttempts);
|
new SystemClock(), appSubmitTime);
|
||||||
}
|
}
|
||||||
|
|
||||||
public MRAppMaster(ApplicationAttemptId applicationAttemptId,
|
public MRAppMaster(ApplicationAttemptId applicationAttemptId,
|
||||||
ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
|
ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
|
||||||
Clock clock, long appSubmitTime, int maxAppAttempts) {
|
Clock clock, long appSubmitTime) {
|
||||||
super(MRAppMaster.class.getName());
|
super(MRAppMaster.class.getName());
|
||||||
this.clock = clock;
|
this.clock = clock;
|
||||||
this.startTime = clock.getTime();
|
this.startTime = clock.getTime();
|
||||||
|
@ -242,7 +241,6 @@ public class MRAppMaster extends CompositeService {
|
||||||
this.nmPort = nmPort;
|
this.nmPort = nmPort;
|
||||||
this.nmHttpPort = nmHttpPort;
|
this.nmHttpPort = nmHttpPort;
|
||||||
this.metrics = MRAppMetrics.create();
|
this.metrics = MRAppMetrics.create();
|
||||||
this.maxAppAttempts = maxAppAttempts;
|
|
||||||
logSyncer = TaskLog.createLogSyncer();
|
logSyncer = TaskLog.createLogSyncer();
|
||||||
LOG.info("Created MRAppMaster for application " + applicationAttemptId);
|
LOG.info("Created MRAppMaster for application " + applicationAttemptId);
|
||||||
}
|
}
|
||||||
|
@ -255,12 +253,6 @@ public class MRAppMaster extends CompositeService {
|
||||||
|
|
||||||
context = new RunningAppContext(conf);
|
context = new RunningAppContext(conf);
|
||||||
|
|
||||||
((RunningAppContext)context).computeIsLastAMRetry();
|
|
||||||
LOG.info("The specific max attempts: " + maxAppAttempts +
|
|
||||||
" for application: " + appAttemptID.getApplicationId().getId() +
|
|
||||||
". Attempt num: " + appAttemptID.getAttemptId() +
|
|
||||||
" is last retry: " + isLastAMRetry);
|
|
||||||
|
|
||||||
// Job name is the same as the app name util we support DAG of jobs
|
// Job name is the same as the app name util we support DAG of jobs
|
||||||
// for an app later
|
// for an app later
|
||||||
appName = conf.get(MRJobConfig.JOB_NAME, "<missing app name>");
|
appName = conf.get(MRJobConfig.JOB_NAME, "<missing app name>");
|
||||||
|
@ -993,8 +985,8 @@ public class MRAppMaster extends CompositeService {
|
||||||
successfullyUnregistered.set(true);
|
successfullyUnregistered.set(true);
|
||||||
}
|
}
|
||||||
|
|
||||||
public void computeIsLastAMRetry() {
|
public void resetIsLastAMRetry() {
|
||||||
isLastAMRetry = appAttemptID.getAttemptId() >= maxAppAttempts;
|
isLastAMRetry = false;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -1374,8 +1366,6 @@ public class MRAppMaster extends CompositeService {
|
||||||
System.getenv(Environment.NM_HTTP_PORT.name());
|
System.getenv(Environment.NM_HTTP_PORT.name());
|
||||||
String appSubmitTimeStr =
|
String appSubmitTimeStr =
|
||||||
System.getenv(ApplicationConstants.APP_SUBMIT_TIME_ENV);
|
System.getenv(ApplicationConstants.APP_SUBMIT_TIME_ENV);
|
||||||
String maxAppAttempts =
|
|
||||||
System.getenv(ApplicationConstants.MAX_APP_ATTEMPTS_ENV);
|
|
||||||
|
|
||||||
validateInputParam(containerIdStr,
|
validateInputParam(containerIdStr,
|
||||||
Environment.CONTAINER_ID.name());
|
Environment.CONTAINER_ID.name());
|
||||||
|
@ -1385,8 +1375,6 @@ public class MRAppMaster extends CompositeService {
|
||||||
Environment.NM_HTTP_PORT.name());
|
Environment.NM_HTTP_PORT.name());
|
||||||
validateInputParam(appSubmitTimeStr,
|
validateInputParam(appSubmitTimeStr,
|
||||||
ApplicationConstants.APP_SUBMIT_TIME_ENV);
|
ApplicationConstants.APP_SUBMIT_TIME_ENV);
|
||||||
validateInputParam(maxAppAttempts,
|
|
||||||
ApplicationConstants.MAX_APP_ATTEMPTS_ENV);
|
|
||||||
|
|
||||||
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
||||||
ApplicationAttemptId applicationAttemptId =
|
ApplicationAttemptId applicationAttemptId =
|
||||||
|
@ -1397,8 +1385,7 @@ public class MRAppMaster extends CompositeService {
|
||||||
MRAppMaster appMaster =
|
MRAppMaster appMaster =
|
||||||
new MRAppMaster(applicationAttemptId, containerId, nodeHostString,
|
new MRAppMaster(applicationAttemptId, containerId, nodeHostString,
|
||||||
Integer.parseInt(nodePortString),
|
Integer.parseInt(nodePortString),
|
||||||
Integer.parseInt(nodeHttpPortString), appSubmitTime,
|
Integer.parseInt(nodeHttpPortString), appSubmitTime);
|
||||||
Integer.parseInt(maxAppAttempts));
|
|
||||||
ShutdownHookManager.get().addShutdownHook(
|
ShutdownHookManager.get().addShutdownHook(
|
||||||
new MRAppMasterShutdownHook(appMaster), SHUTDOWN_HOOK_PRIORITY);
|
new MRAppMasterShutdownHook(appMaster), SHUTDOWN_HOOK_PRIORITY);
|
||||||
JobConf conf = new JobConf(new YarnConfiguration());
|
JobConf conf = new JobConf(new YarnConfiguration());
|
||||||
|
|
|
@ -185,7 +185,7 @@ public abstract class RMCommunicator extends AbstractService
|
||||||
// if unregistration failed, isLastAMRetry needs to be recalculated
|
// if unregistration failed, isLastAMRetry needs to be recalculated
|
||||||
// to see whether AM really has the chance to retry
|
// to see whether AM really has the chance to retry
|
||||||
RunningAppContext raContext = (RunningAppContext) context;
|
RunningAppContext raContext = (RunningAppContext) context;
|
||||||
raContext.computeIsLastAMRetry();
|
raContext.resetIsLastAMRetry();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -227,8 +227,8 @@ public class MRApp extends MRAppMaster {
|
||||||
int maps, int reduces, boolean autoComplete, String testName,
|
int maps, int reduces, boolean autoComplete, String testName,
|
||||||
boolean cleanOnStart, int startCount, Clock clock, boolean unregistered,
|
boolean cleanOnStart, int startCount, Clock clock, boolean unregistered,
|
||||||
String assignedQueue) {
|
String assignedQueue) {
|
||||||
super(appAttemptId, amContainerId, NM_HOST, NM_PORT, NM_HTTP_PORT, clock, System
|
super(appAttemptId, amContainerId, NM_HOST, NM_PORT, NM_HTTP_PORT, clock,
|
||||||
.currentTimeMillis(), MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS);
|
System.currentTimeMillis());
|
||||||
this.testWorkDir = new File("target", testName);
|
this.testWorkDir = new File("target", testName);
|
||||||
testAbsPath = new Path(testWorkDir.getAbsolutePath());
|
testAbsPath = new Path(testWorkDir.getAbsolutePath());
|
||||||
LOG.info("PathUsed: " + testAbsPath);
|
LOG.info("PathUsed: " + testAbsPath);
|
||||||
|
|
|
@ -253,6 +253,12 @@ public class TestJobEndNotifier extends JobEndNotifier {
|
||||||
HttpServer2 server = startHttpServer();
|
HttpServer2 server = startHttpServer();
|
||||||
MRApp app = spy(new MRAppWithCustomContainerAllocator(2, 2, false,
|
MRApp app = spy(new MRAppWithCustomContainerAllocator(2, 2, false,
|
||||||
this.getClass().getName(), true, 2, false));
|
this.getClass().getName(), true, 2, false));
|
||||||
|
// Currently, we will have isLastRetry always equals to false at beginning
|
||||||
|
// of MRAppMaster, except staging area exists or commit already started at
|
||||||
|
// the beginning.
|
||||||
|
// Now manually set isLastRetry to true and this should reset to false when
|
||||||
|
// unregister failed.
|
||||||
|
app.isLastAMRetry = true;
|
||||||
doNothing().when(app).sysexit();
|
doNothing().when(app).sysexit();
|
||||||
JobConf conf = new JobConf();
|
JobConf conf = new JobConf();
|
||||||
conf.set(JobContext.MR_JOB_END_NOTIFICATION_URL,
|
conf.set(JobContext.MR_JOB_END_NOTIFICATION_URL,
|
||||||
|
@ -265,12 +271,11 @@ public class TestJobEndNotifier extends JobEndNotifier {
|
||||||
// Now shutdown. User should see FAILED state.
|
// Now shutdown. User should see FAILED state.
|
||||||
// Unregistration fails: isLastAMRetry is recalculated, this is
|
// Unregistration fails: isLastAMRetry is recalculated, this is
|
||||||
app.shutDownJob();
|
app.shutDownJob();
|
||||||
Assert.assertTrue(app.isLastAMRetry());
|
Assert.assertFalse(app.isLastAMRetry());
|
||||||
Assert.assertEquals(1, JobEndServlet.calledTimes);
|
// Since it's not last retry, JobEndServlet didn't called
|
||||||
Assert.assertEquals("jobid=" + job.getID() + "&status=FAILED",
|
Assert.assertEquals(0, JobEndServlet.calledTimes);
|
||||||
JobEndServlet.requestUri.getQuery());
|
Assert.assertNull(JobEndServlet.requestUri);
|
||||||
Assert.assertEquals(JobState.FAILED.toString(),
|
Assert.assertNull(JobEndServlet.foundJobState);
|
||||||
JobEndServlet.foundJobState);
|
|
||||||
server.stop();
|
server.stop();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -118,7 +118,7 @@ public class TestMRAppMaster {
|
||||||
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
||||||
MRAppMasterTest appMaster =
|
MRAppMasterTest appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS);
|
System.currentTimeMillis());
|
||||||
JobConf conf = new JobConf();
|
JobConf conf = new JobConf();
|
||||||
conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
|
conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
|
@ -147,8 +147,7 @@ public class TestMRAppMaster {
|
||||||
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
||||||
MRAppMaster appMaster =
|
MRAppMaster appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS,
|
System.currentTimeMillis(), false, false);
|
||||||
false, false);
|
|
||||||
boolean caught = false;
|
boolean caught = false;
|
||||||
try {
|
try {
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
|
@ -186,8 +185,7 @@ public class TestMRAppMaster {
|
||||||
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
||||||
MRAppMaster appMaster =
|
MRAppMaster appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS,
|
System.currentTimeMillis(), false, false);
|
||||||
false, false);
|
|
||||||
boolean caught = false;
|
boolean caught = false;
|
||||||
try {
|
try {
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
|
@ -225,8 +223,7 @@ public class TestMRAppMaster {
|
||||||
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
||||||
MRAppMaster appMaster =
|
MRAppMaster appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS,
|
System.currentTimeMillis(), false, false);
|
||||||
false, false);
|
|
||||||
boolean caught = false;
|
boolean caught = false;
|
||||||
try {
|
try {
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
|
@ -264,8 +261,7 @@ public class TestMRAppMaster {
|
||||||
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
ContainerId containerId = ConverterUtils.toContainerId(containerIdStr);
|
||||||
MRAppMaster appMaster =
|
MRAppMaster appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), MRJobConfig.DEFAULT_MR_AM_MAX_ATTEMPTS,
|
System.currentTimeMillis(), false, false);
|
||||||
false, false);
|
|
||||||
boolean caught = false;
|
boolean caught = false;
|
||||||
try {
|
try {
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
|
@ -285,8 +281,9 @@ public class TestMRAppMaster {
|
||||||
@Test (timeout = 30000)
|
@Test (timeout = 30000)
|
||||||
public void testMRAppMasterMaxAppAttempts() throws IOException,
|
public void testMRAppMasterMaxAppAttempts() throws IOException,
|
||||||
InterruptedException {
|
InterruptedException {
|
||||||
int[] maxAppAttemtps = new int[] { 1, 2, 3 };
|
// No matter what's the maxAppAttempt or attempt id, the isLastRetry always
|
||||||
Boolean[] expectedBools = new Boolean[]{ true, true, false };
|
// equals to false
|
||||||
|
Boolean[] expectedBools = new Boolean[]{ false, false, false };
|
||||||
|
|
||||||
String applicationAttemptIdStr = "appattempt_1317529182569_0004_000002";
|
String applicationAttemptIdStr = "appattempt_1317529182569_0004_000002";
|
||||||
String containerIdStr = "container_1317529182569_0004_000002_1";
|
String containerIdStr = "container_1317529182569_0004_000002_1";
|
||||||
|
@ -301,10 +298,10 @@ public class TestMRAppMaster {
|
||||||
File stagingDir =
|
File stagingDir =
|
||||||
new File(MRApps.getStagingAreaDir(conf, userName).toString());
|
new File(MRApps.getStagingAreaDir(conf, userName).toString());
|
||||||
stagingDir.mkdirs();
|
stagingDir.mkdirs();
|
||||||
for (int i = 0; i < maxAppAttemtps.length; ++i) {
|
for (int i = 0; i < expectedBools.length; ++i) {
|
||||||
MRAppMasterTest appMaster =
|
MRAppMasterTest appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), maxAppAttemtps[i], false, true);
|
System.currentTimeMillis(), false, true);
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
assertEquals("isLastAMRetry is correctly computed.", expectedBools[i],
|
assertEquals("isLastAMRetry is correctly computed.", expectedBools[i],
|
||||||
appMaster.isLastAMRetry());
|
appMaster.isLastAMRetry());
|
||||||
|
@ -399,7 +396,7 @@ public class TestMRAppMaster {
|
||||||
|
|
||||||
MRAppMasterTest appMaster =
|
MRAppMasterTest appMaster =
|
||||||
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1,
|
||||||
System.currentTimeMillis(), 1, false, true);
|
System.currentTimeMillis(), false, true);
|
||||||
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
MRAppMaster.initAndStartAppMaster(appMaster, conf, userName);
|
||||||
|
|
||||||
// Now validate the task credentials
|
// Now validate the task credentials
|
||||||
|
@ -466,16 +463,15 @@ class MRAppMasterTest extends MRAppMaster {
|
||||||
|
|
||||||
public MRAppMasterTest(ApplicationAttemptId applicationAttemptId,
|
public MRAppMasterTest(ApplicationAttemptId applicationAttemptId,
|
||||||
ContainerId containerId, String host, int port, int httpPort,
|
ContainerId containerId, String host, int port, int httpPort,
|
||||||
long submitTime, int maxAppAttempts) {
|
long submitTime) {
|
||||||
this(applicationAttemptId, containerId, host, port, httpPort,
|
this(applicationAttemptId, containerId, host, port, httpPort,
|
||||||
submitTime, maxAppAttempts, true, true);
|
submitTime, true, true);
|
||||||
}
|
}
|
||||||
public MRAppMasterTest(ApplicationAttemptId applicationAttemptId,
|
public MRAppMasterTest(ApplicationAttemptId applicationAttemptId,
|
||||||
ContainerId containerId, String host, int port, int httpPort,
|
ContainerId containerId, String host, int port, int httpPort,
|
||||||
long submitTime, int maxAppAttempts, boolean overrideInit,
|
long submitTime, boolean overrideInit,
|
||||||
boolean overrideStart) {
|
boolean overrideStart) {
|
||||||
super(applicationAttemptId, containerId, host, port, httpPort, submitTime,
|
super(applicationAttemptId, containerId, host, port, httpPort, submitTime);
|
||||||
maxAppAttempts);
|
|
||||||
this.overrideInit = overrideInit;
|
this.overrideInit = overrideInit;
|
||||||
this.overrideStart = overrideStart;
|
this.overrideStart = overrideStart;
|
||||||
mockContainerAllocator = mock(ContainerAllocator.class);
|
mockContainerAllocator = mock(ContainerAllocator.class);
|
||||||
|
|
|
@ -18,6 +18,7 @@
|
||||||
|
|
||||||
package org.apache.hadoop.mapreduce.v2.app;
|
package org.apache.hadoop.mapreduce.v2.app;
|
||||||
|
|
||||||
|
import static org.junit.Assert.assertTrue;
|
||||||
import static org.mockito.Matchers.any;
|
import static org.mockito.Matchers.any;
|
||||||
import static org.mockito.Matchers.anyBoolean;
|
import static org.mockito.Matchers.anyBoolean;
|
||||||
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.mock;
|
||||||
|
@ -28,9 +29,6 @@ import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
|
||||||
import org.junit.Assert;
|
|
||||||
import junit.framework.TestCase;
|
|
||||||
|
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.fs.FileSystem;
|
import org.apache.hadoop.fs.FileSystem;
|
||||||
import org.apache.hadoop.fs.Path;
|
import org.apache.hadoop.fs.Path;
|
||||||
|
@ -62,13 +60,14 @@ import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||||
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
|
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
|
||||||
import org.apache.hadoop.yarn.factories.RecordFactory;
|
import org.apache.hadoop.yarn.factories.RecordFactory;
|
||||||
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
|
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
|
||||||
|
import org.junit.Assert;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Make sure that the job staging directory clean up happens.
|
* Make sure that the job staging directory clean up happens.
|
||||||
*/
|
*/
|
||||||
public class TestStagingCleanup extends TestCase {
|
public class TestStagingCleanup {
|
||||||
|
|
||||||
private Configuration conf = new Configuration();
|
private Configuration conf = new Configuration();
|
||||||
private FileSystem fs;
|
private FileSystem fs;
|
||||||
|
@ -81,7 +80,7 @@ import org.junit.Test;
|
||||||
public void testDeletionofStagingOnUnregistrationFailure()
|
public void testDeletionofStagingOnUnregistrationFailure()
|
||||||
throws IOException {
|
throws IOException {
|
||||||
testDeletionofStagingOnUnregistrationFailure(2, false);
|
testDeletionofStagingOnUnregistrationFailure(2, false);
|
||||||
testDeletionofStagingOnUnregistrationFailure(1, true);
|
testDeletionofStagingOnUnregistrationFailure(1, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
@SuppressWarnings("resource")
|
@SuppressWarnings("resource")
|
||||||
|
@ -104,7 +103,7 @@ import org.junit.Test;
|
||||||
appMaster.init(conf);
|
appMaster.init(conf);
|
||||||
appMaster.start();
|
appMaster.start();
|
||||||
appMaster.shutDownJob();
|
appMaster.shutDownJob();
|
||||||
((RunningAppContext) appMaster.getContext()).computeIsLastAMRetry();
|
((RunningAppContext) appMaster.getContext()).resetIsLastAMRetry();
|
||||||
if (shouldHaveDeleted) {
|
if (shouldHaveDeleted) {
|
||||||
Assert.assertEquals(new Boolean(true), appMaster.isLastAMRetry());
|
Assert.assertEquals(new Boolean(true), appMaster.isLastAMRetry());
|
||||||
verify(fs).delete(stagingJobPath, true);
|
verify(fs).delete(stagingJobPath, true);
|
||||||
|
@ -164,7 +163,11 @@ import org.junit.Test;
|
||||||
verify(fs, times(0)).delete(stagingJobPath, true);
|
verify(fs, times(0)).delete(stagingJobPath, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test (timeout = 30000)
|
// FIXME:
|
||||||
|
// Disabled this test because currently, when job state=REBOOT at shutdown
|
||||||
|
// when lastRetry = true in RM view, cleanup will not do.
|
||||||
|
// This will be supported after YARN-2261 completed
|
||||||
|
// @Test (timeout = 30000)
|
||||||
public void testDeletionofStagingOnReboot() throws IOException {
|
public void testDeletionofStagingOnReboot() throws IOException {
|
||||||
conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, stagingJobDir);
|
conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, stagingJobDir);
|
||||||
fs = mock(FileSystem.class);
|
fs = mock(FileSystem.class);
|
||||||
|
@ -202,7 +205,7 @@ import org.junit.Test;
|
||||||
JobId jobid = recordFactory.newRecordInstance(JobId.class);
|
JobId jobid = recordFactory.newRecordInstance(JobId.class);
|
||||||
jobid.setAppId(appId);
|
jobid.setAppId(appId);
|
||||||
ContainerAllocator mockAlloc = mock(ContainerAllocator.class);
|
ContainerAllocator mockAlloc = mock(ContainerAllocator.class);
|
||||||
MRAppMaster appMaster = new TestMRApp(attemptId, mockAlloc, 4);
|
MRAppMaster appMaster = new TestMRApp(attemptId, mockAlloc);
|
||||||
appMaster.init(conf);
|
appMaster.init(conf);
|
||||||
//simulate the process being killed
|
//simulate the process being killed
|
||||||
MRAppMaster.MRAppMasterShutdownHook hook =
|
MRAppMaster.MRAppMasterShutdownHook hook =
|
||||||
|
@ -210,8 +213,12 @@ import org.junit.Test;
|
||||||
hook.run();
|
hook.run();
|
||||||
verify(fs, times(0)).delete(stagingJobPath, true);
|
verify(fs, times(0)).delete(stagingJobPath, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test (timeout = 30000)
|
// FIXME:
|
||||||
|
// Disabled this test because currently, when shutdown hook triggered at
|
||||||
|
// lastRetry in RM view, cleanup will not do. This should be supported after
|
||||||
|
// YARN-2261 completed
|
||||||
|
// @Test (timeout = 30000)
|
||||||
public void testDeletionofStagingOnKillLastTry() throws IOException {
|
public void testDeletionofStagingOnKillLastTry() throws IOException {
|
||||||
conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, stagingJobDir);
|
conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, stagingJobDir);
|
||||||
fs = mock(FileSystem.class);
|
fs = mock(FileSystem.class);
|
||||||
|
@ -226,7 +233,7 @@ import org.junit.Test;
|
||||||
JobId jobid = recordFactory.newRecordInstance(JobId.class);
|
JobId jobid = recordFactory.newRecordInstance(JobId.class);
|
||||||
jobid.setAppId(appId);
|
jobid.setAppId(appId);
|
||||||
ContainerAllocator mockAlloc = mock(ContainerAllocator.class);
|
ContainerAllocator mockAlloc = mock(ContainerAllocator.class);
|
||||||
MRAppMaster appMaster = new TestMRApp(attemptId, mockAlloc, 1); //no retry
|
MRAppMaster appMaster = new TestMRApp(attemptId, mockAlloc); //no retry
|
||||||
appMaster.init(conf);
|
appMaster.init(conf);
|
||||||
assertTrue("appMaster.isLastAMRetry() is false", appMaster.isLastAMRetry());
|
assertTrue("appMaster.isLastAMRetry() is false", appMaster.isLastAMRetry());
|
||||||
//simulate the process being killed
|
//simulate the process being killed
|
||||||
|
@ -245,10 +252,10 @@ import org.junit.Test;
|
||||||
boolean crushUnregistration = false;
|
boolean crushUnregistration = false;
|
||||||
|
|
||||||
public TestMRApp(ApplicationAttemptId applicationAttemptId,
|
public TestMRApp(ApplicationAttemptId applicationAttemptId,
|
||||||
ContainerAllocator allocator, int maxAppAttempts) {
|
ContainerAllocator allocator) {
|
||||||
super(applicationAttemptId, ContainerId.newInstance(
|
super(applicationAttemptId, ContainerId.newInstance(
|
||||||
applicationAttemptId, 1), "testhost", 2222, 3333,
|
applicationAttemptId, 1), "testhost", 2222, 3333,
|
||||||
System.currentTimeMillis(), maxAppAttempts);
|
System.currentTimeMillis());
|
||||||
this.allocator = allocator;
|
this.allocator = allocator;
|
||||||
this.successfullyUnregistered.set(true);
|
this.successfullyUnregistered.set(true);
|
||||||
}
|
}
|
||||||
|
@ -256,7 +263,7 @@ import org.junit.Test;
|
||||||
public TestMRApp(ApplicationAttemptId applicationAttemptId,
|
public TestMRApp(ApplicationAttemptId applicationAttemptId,
|
||||||
ContainerAllocator allocator, JobStateInternal jobStateInternal,
|
ContainerAllocator allocator, JobStateInternal jobStateInternal,
|
||||||
int maxAppAttempts) {
|
int maxAppAttempts) {
|
||||||
this(applicationAttemptId, allocator, maxAppAttempts);
|
this(applicationAttemptId, allocator);
|
||||||
this.jobStateInternal = jobStateInternal;
|
this.jobStateInternal = jobStateInternal;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue