big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

YARN-5617. AMs only intended to run one attempt can be run more than once? Contributed by Jason Lowe.

Browse Source
This commit is contained in:
Eric Payne 2017-04-24 12:03:28 -05:00
parent 561718e05d
commit 52adf71914
2 changed files with 123 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
View File

@ -1517,6 +1517,13 @@ public AttemptFailedTransition(RMAppState initialState) {
@Override
public RMAppState transition(RMAppImpl app, RMAppEvent event) {
int numberOfFailure = app.getNumFailedAppAttempts();
if (app.maxAppAttempts == 1) {
// If the user explicitly set the attempts to 1 then there are likely
// correctness issues if the AM restarts for any reason.
LOG.info("Max app attempts is 1 for " + app.applicationId
+ ", preventing further attempts.");
numberOfFailure = app.maxAppAttempts;
} else {
LOG.info("The number of failed attempts"
+ (app.attemptFailuresValidityInterval > 0 ? " in previous "
+ app.attemptFailuresValidityInterval + " milliseconds " : " ")
@ -1526,6 +1533,7 @@ public RMAppState transition(RMAppImpl app, RMAppEvent event) {
if (app.attemptFailuresValidityInterval > 0) {
removeExcessAttempts(app);
}
}
if (!app.submissionContext.getUnmanagedAM()
&& numberOfFailure < app.maxAppAttempts) {

130
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/TestAMRestart.java
View File

@ -378,8 +378,7 @@ public void testShouldNotCountFailureToMaxAttemptRetry() throws Exception {
YarnConfiguration conf = new YarnConfiguration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
ResourceScheduler.class);
// explicitly set max-am-retry count as 1.
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 1);
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 2);
conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName());
MemoryRMStateStore memStore = new MemoryRMStateStore();
@ -396,7 +395,7 @@ public void testShouldNotCountFailureToMaxAttemptRetry() throws Exception {
(CapacityScheduler) rm1.getResourceScheduler();
ContainerId amContainer =
ContainerId.newContainerId(am1.getApplicationAttemptId(), 1);
// Preempt the first attempt;
// Preempt the next attempt;
scheduler.markContainerForKillable(scheduler.getRMContainer(amContainer));
rm1.waitForState(am1.getApplicationAttemptId(), RMAppAttemptState.FAILED);
@ -464,7 +463,7 @@ public void testShouldNotCountFailureToMaxAttemptRetry() throws Exception {
TestSchedulerUtils.waitSchedulerApplicationAttemptStopped(scheduler,
am4.getApplicationAttemptId());
Assert.assertTrue(! attempt4.shouldCountTowardsMaxAttemptRetry());
Assert.assertFalse(attempt4.shouldCountTowardsMaxAttemptRetry());
Assert.assertEquals(ContainerExitStatus.ABORTED,
appState.getAttempt(am4.getApplicationAttemptId())
.getAMContainerExitStatus());
@ -481,9 +480,58 @@ public void testShouldNotCountFailureToMaxAttemptRetry() throws Exception {
Assert.assertTrue(attempt5.shouldCountTowardsMaxAttemptRetry());
// launch next AM in nm2
MockAM am6 =
rm1.waitForNewAMToLaunchAndRegister(app1.getApplicationId(), 6, nm2);
RMAppAttempt attempt6 = app1.getCurrentAppAttempt();
// fail the AM normally
nm2
.nodeHeartbeat(am6.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
rm1.waitForState(am6.getApplicationAttemptId(), RMAppAttemptState.FAILED);
TestSchedulerUtils.waitSchedulerApplicationAttemptStopped(scheduler,
am6.getApplicationAttemptId());
Assert.assertTrue(attempt6.shouldCountTowardsMaxAttemptRetry());
// AM should not be restarted.
rm1.waitForState(app1.getApplicationId(), RMAppState.FAILED);
Assert.assertEquals(5, app1.getAppAttempts().size());
Assert.assertEquals(6, app1.getAppAttempts().size());
rm1.stop();
}
@Test(timeout = 100000)
public void testMaxAttemptOneMeansOne() throws Exception {
YarnConfiguration conf = new YarnConfiguration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
ResourceScheduler.class);
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 1);
conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName());
MemoryRMStateStore memStore = new MemoryRMStateStore();
memStore.init(conf);
MockRM rm1 = new MockRM(conf, memStore);
rm1.start();
MockNM nm1 =
new MockNM("127.0.0.1:1234", 8000, rm1.getResourceTrackerService());
nm1.registerNode();
RMApp app1 = rm1.submitApp(200);
RMAppAttempt attempt1 = app1.getCurrentAppAttempt();
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
CapacityScheduler scheduler =
(CapacityScheduler) rm1.getResourceScheduler();
ContainerId amContainer =
ContainerId.newContainerId(am1.getApplicationAttemptId(), 1);
// Preempt the attempt;
scheduler.markContainerForKillable(scheduler.getRMContainer(amContainer));
rm1.waitForState(am1.getApplicationAttemptId(), RMAppAttemptState.FAILED);
TestSchedulerUtils.waitSchedulerApplicationAttemptStopped(scheduler,
am1.getApplicationAttemptId());
// AM should not be restarted.
rm1.waitForState(app1.getApplicationId(), RMAppState.FAILED);
Assert.assertEquals(1, app1.getAppAttempts().size());
rm1.stop();
}
@ -499,8 +547,7 @@ public void testPreemptedAMRestartOnRMRestart() throws Exception {
conf.setBoolean(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED, false);
conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName());
// explicitly set max-am-retry count as 1.
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 1);
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 2);
MemoryRMStateStore memStore = new MemoryRMStateStore();
memStore.init(conf);
@ -517,20 +564,35 @@ public void testPreemptedAMRestartOnRMRestart() throws Exception {
ContainerId amContainer =
ContainerId.newContainerId(am1.getApplicationAttemptId(), 1);
// fail the AM normally
nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 1,
ContainerState.COMPLETE);
rm1.waitForState(am1.getApplicationAttemptId(), RMAppAttemptState.FAILED);
TestSchedulerUtils.waitSchedulerApplicationAttemptStopped(scheduler,
am1.getApplicationAttemptId());
Assert.assertTrue(attempt1.shouldCountTowardsMaxAttemptRetry());
// wait for the next AM to start
rm1.waitForState(app1.getApplicationId(), RMAppState.ACCEPTED);
MockAM am2 =
rm1.waitForNewAMToLaunchAndRegister(app1.getApplicationId(), 2, nm1);
RMAppAttempt attempt2 = app1.getCurrentAppAttempt();
// Forcibly preempt the am container;
amContainer = ContainerId.newContainerId(am2.getApplicationAttemptId(), 1);
scheduler.markContainerForKillable(scheduler.getRMContainer(amContainer));
rm1.waitForState(am1.getApplicationAttemptId(), RMAppAttemptState.FAILED);
Assert.assertTrue(! attempt1.shouldCountTowardsMaxAttemptRetry());
rm1.waitForState(am2.getApplicationAttemptId(), RMAppAttemptState.FAILED);
Assert.assertFalse(attempt2.shouldCountTowardsMaxAttemptRetry());
rm1.waitForState(app1.getApplicationId(), RMAppState.ACCEPTED);
// state store has 1 attempt stored.
// state store has 2 attempts stored.
ApplicationStateData appState =
memStore.getState().getApplicationState().get(app1.getApplicationId());
Assert.assertEquals(1, appState.getAttemptCount());
Assert.assertEquals(2, appState.getAttemptCount());
// attempt stored has the preempted container exit status.
Assert.assertEquals(ContainerExitStatus.PREEMPTED,
appState.getAttempt(am1.getApplicationAttemptId())
appState.getAttempt(am2.getApplicationAttemptId())
.getAMContainerExitStatus());
// Restart rm.
MockRM rm2 = new MockRM(conf, memStore);
@ -539,15 +601,15 @@ public void testPreemptedAMRestartOnRMRestart() throws Exception {
rm2.start();
// Restarted RM should re-launch the am.
MockAM am2 =
MockAM am3 =
rm2.waitForNewAMToLaunchAndRegister(app1.getApplicationId(), 2, nm1);
MockRM.finishAMAndVerifyAppState(app1, rm2, nm1, am2);
RMAppAttempt attempt2 =
MockRM.finishAMAndVerifyAppState(app1, rm2, nm1, am3);
RMAppAttempt attempt3 =
rm2.getRMContext().getRMApps().get(app1.getApplicationId())
.getCurrentAppAttempt();
Assert.assertTrue(attempt2.shouldCountTowardsMaxAttemptRetry());
Assert.assertTrue(attempt3.shouldCountTowardsMaxAttemptRetry());
Assert.assertEquals(ContainerExitStatus.INVALID,
appState.getAttempt(am2.getApplicationAttemptId())
appState.getAttempt(am3.getApplicationAttemptId())
.getAMContainerExitStatus());
rm1.stop();
rm2.stop();
@ -566,13 +628,15 @@ public void testRMRestartOrFailoverNotCountedForAMFailures()
conf.setBoolean(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED, false);
conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName());
// explicitly set max-am-retry count as 1.
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 1);
// explicitly set max-am-retry count as 2.
conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 2);
MemoryRMStateStore memStore = new MemoryRMStateStore();
memStore.init(conf);
MockRM rm1 = new MockRM(conf, memStore);
rm1.start();
CapacityScheduler scheduler =
(CapacityScheduler) rm1.getResourceScheduler();
MockNM nm1 =
new MockNM("127.0.0.1:1234", 8000, rm1.getResourceTrackerService());
nm1.registerNode();
@ -581,6 +645,20 @@ public void testRMRestartOrFailoverNotCountedForAMFailures()
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
RMAppAttempt attempt1 = app1.getCurrentAppAttempt();
// fail the AM normally
nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 1,
ContainerState.COMPLETE);
rm1.waitForState(am1.getApplicationAttemptId(), RMAppAttemptState.FAILED);
TestSchedulerUtils.waitSchedulerApplicationAttemptStopped(scheduler,
am1.getApplicationAttemptId());
Assert.assertTrue(attempt1.shouldCountTowardsMaxAttemptRetry());
// wait for the next AM to start
rm1.waitForState(app1.getApplicationId(), RMAppState.ACCEPTED);
MockAM am2 =
rm1.waitForNewAMToLaunchAndRegister(app1.getApplicationId(), 2, nm1);
RMAppAttempt attempt2 = app1.getCurrentAppAttempt();
// Restart rm.
MockRM rm2 = new MockRM(conf, memStore);
rm2.start();
@ -591,26 +669,26 @@ public void testRMRestartOrFailoverNotCountedForAMFailures()
NMContainerStatus status = Records.newRecord(NMContainerStatus.class);
status
.setContainerExitStatus(ContainerExitStatus.KILLED_BY_RESOURCEMANAGER);
status.setContainerId(attempt1.getMasterContainer().getId());
status.setContainerId(attempt2.getMasterContainer().getId());
status.setContainerState(ContainerState.COMPLETE);
status.setDiagnostics("");
nm1.registerNode(Collections.singletonList(status), null);
rm2.waitForState(attempt1.getAppAttemptId(), RMAppAttemptState.FAILED);
rm2.waitForState(attempt2.getAppAttemptId(), RMAppAttemptState.FAILED);
Assert.assertEquals(ContainerExitStatus.KILLED_BY_RESOURCEMANAGER,
appState.getAttempt(am1.getApplicationAttemptId())
appState.getAttempt(am2.getApplicationAttemptId())
.getAMContainerExitStatus());
// Will automatically start a new AppAttempt in rm2
rm2.waitForState(app1.getApplicationId(), RMAppState.ACCEPTED);
MockAM am2 =
rm2.waitForNewAMToLaunchAndRegister(app1.getApplicationId(), 2, nm1);
MockRM.finishAMAndVerifyAppState(app1, rm2, nm1, am2);
MockAM am3 =
rm2.waitForNewAMToLaunchAndRegister(app1.getApplicationId(), 3, nm1);
MockRM.finishAMAndVerifyAppState(app1, rm2, nm1, am3);
RMAppAttempt attempt3 =
rm2.getRMContext().getRMApps().get(app1.getApplicationId())
.getCurrentAppAttempt();
Assert.assertTrue(attempt3.shouldCountTowardsMaxAttemptRetry());
Assert.assertEquals(ContainerExitStatus.INVALID,
appState.getAttempt(am2.getApplicationAttemptId())
appState.getAttempt(am3.getApplicationAttemptId())
.getAMContainerExitStatus());
rm1.stop();