Merge -c 1361813 from trunk to branch-2 to fix MAPREDUCE-4427. Added an 'unmanaged' mode for AMs so as to ease development of new applications. Contributed by Bikas Saha.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1361815 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Arun Murthy 2012-07-15 21:47:53 +00:00
parent 5685eb30bb
commit 4902c6bdc9
18 changed files with 363 additions and 233 deletions

View File

@ -24,6 +24,9 @@ Release 2.0.1-alpha - UNRELEASED
MAPREDUCE-4355. Add RunningJob.getJobStatus() (kkambatl via tucu) MAPREDUCE-4355. Add RunningJob.getJobStatus() (kkambatl via tucu)
MAPREDUCE-4427. Added an 'unmanaged' mode for AMs so as to ease
development of new applications. (Bikas Saha via acmurthy)
OPTIMIZATIONS OPTIMIZATIONS
BUG FIXES BUG FIXES

View File

@ -383,6 +383,7 @@ public class TypeConverter {
switch (yarnApplicationState) { switch (yarnApplicationState) {
case NEW: case NEW:
case SUBMITTED: case SUBMITTED:
case ACCEPTED:
return State.PREP; return State.PREP;
case RUNNING: case RUNNING:
return State.RUNNING; return State.RUNNING;

View File

@ -232,8 +232,9 @@ public class ClientServiceDelegate {
if (user == null) { if (user == null) {
throw RPCUtil.getRemoteException("User is not set in the application report"); throw RPCUtil.getRemoteException("User is not set in the application report");
} }
if (application.getYarnApplicationState() == YarnApplicationState.NEW || if (application.getYarnApplicationState() == YarnApplicationState.NEW
application.getYarnApplicationState() == YarnApplicationState.SUBMITTED) { || application.getYarnApplicationState() == YarnApplicationState.SUBMITTED
|| application.getYarnApplicationState() == YarnApplicationState.ACCEPTED) {
realProxy = null; realProxy = null;
return getNotRunningJob(application, JobState.NEW); return getNotRunningJob(application, JobState.NEW);
} }

View File

@ -56,6 +56,7 @@ import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptCompletionEvent;
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId; import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId;
import org.apache.hadoop.mapreduce.v2.api.records.TaskReport; import org.apache.hadoop.mapreduce.v2.api.records.TaskReport;
import org.apache.hadoop.mapreduce.v2.api.records.TaskState; import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.ApplicationReport;
@ -75,13 +76,17 @@ public class NotRunningJob implements MRClientProtocol {
private ApplicationReport getUnknownApplicationReport() { private ApplicationReport getUnknownApplicationReport() {
ApplicationId unknownAppId = recordFactory.newRecordInstance(ApplicationId.class); ApplicationId unknownAppId = recordFactory
.newRecordInstance(ApplicationId.class);
ApplicationAttemptId unknownAttemptId = recordFactory
.newRecordInstance(ApplicationAttemptId.class);
// Setting AppState to NEW and finalStatus to UNDEFINED as they are never used // Setting AppState to NEW and finalStatus to UNDEFINED as they are never
// used
// for a non running job // for a non running job
return BuilderUtils.newApplicationReport(unknownAppId, "N/A", "N/A", "N/A", "N/A", 0, "", return BuilderUtils.newApplicationReport(unknownAppId, unknownAttemptId,
YarnApplicationState.NEW, "N/A", "N/A", 0, 0, "N/A", "N/A", "N/A", "N/A", 0, "", YarnApplicationState.NEW, "N/A",
FinalApplicationStatus.UNDEFINED, null, "N/A"); "N/A", 0, 0, FinalApplicationStatus.UNDEFINED, null, "N/A");
} }
NotRunningJob(ApplicationReport applicationReport, JobState jobState) { NotRunningJob(ApplicationReport applicationReport, JobState jobState) {

View File

@ -50,6 +50,7 @@ import org.apache.hadoop.mapreduce.v2.api.records.Counters;
import org.apache.hadoop.mapreduce.v2.api.records.JobReport; import org.apache.hadoop.mapreduce.v2.api.records.JobReport;
import org.apache.hadoop.mapreduce.v2.api.records.JobState; import org.apache.hadoop.mapreduce.v2.api.records.JobState;
import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils; import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
@ -404,17 +405,23 @@ public class TestClientServiceDelegate {
} }
private ApplicationReport getFinishedApplicationReport() { private ApplicationReport getFinishedApplicationReport() {
return BuilderUtils.newApplicationReport(BuilderUtils.newApplicationId( ApplicationId appId = BuilderUtils.newApplicationId(1234, 5);
1234, 5), "user", "queue", "appname", "host", 124, null, ApplicationAttemptId attemptId = BuilderUtils.newApplicationAttemptId(
YarnApplicationState.FINISHED, "diagnostics", "url", 0, 0, appId, 0);
FinalApplicationStatus.SUCCEEDED, null, "N/A"); return BuilderUtils.newApplicationReport(appId, attemptId, "user", "queue",
"appname", "host", 124, null, YarnApplicationState.FINISHED,
"diagnostics", "url", 0, 0, FinalApplicationStatus.SUCCEEDED, null,
"N/A");
} }
private ApplicationReport getRunningApplicationReport(String host, int port) { private ApplicationReport getRunningApplicationReport(String host, int port) {
return BuilderUtils.newApplicationReport(BuilderUtils.newApplicationId( ApplicationId appId = BuilderUtils.newApplicationId(1234, 5);
1234, 5), "user", "queue", "appname", host, port, null, ApplicationAttemptId attemptId = BuilderUtils.newApplicationAttemptId(
YarnApplicationState.RUNNING, "diagnostics", "url", 0, 0, appId, 0);
FinalApplicationStatus.UNDEFINED, null, "N/A"); return BuilderUtils.newApplicationReport(appId, attemptId, "user", "queue",
"appname", host, port, null, YarnApplicationState.RUNNING,
"diagnostics", "url", 0, 0, FinalApplicationStatus.UNDEFINED, null,
"N/A");
} }
private ResourceMgrDelegate getRMDelegate() throws YarnRemoteException { private ResourceMgrDelegate getRMDelegate() throws YarnRemoteException {

View File

@ -42,6 +42,12 @@ public interface ApplicationConstants {
*/ */
public static final String AM_CONTAINER_ID_ENV = "AM_CONTAINER_ID"; public static final String AM_CONTAINER_ID_ENV = "AM_CONTAINER_ID";
/**
* The environment variable for APPLICATION_ATTEMPT_ID. Set in AppMaster
* environment only
*/
public static final String AM_APP_ATTEMPT_ID_ENV = "AM_APP_ATTEMPT_ID";
/** /**
* The environment variable for the NM_HOST. Set in the AppMaster environment * The environment variable for the NM_HOST. Set in the AppMaster environment
* only * only

View File

@ -61,6 +61,19 @@ public interface ApplicationReport {
@Unstable @Unstable
void setApplicationId(ApplicationId applicationId); void setApplicationId(ApplicationId applicationId);
/**
* Get the <code>ApplicationAttemptId</code> of the current
* attempt of the application
* @return <code>ApplicationAttemptId</code> of the attempt
*/
@Private
@Unstable
ApplicationAttemptId getCurrentApplicationAttemptId();
@Private
@Unstable
void setCurrentApplicationAttemptId(ApplicationAttemptId applicationAttemptId);
/** /**
* Get the <em>user</em> who submitted the application. * Get the <em>user</em> who submitted the application.
* @return <em>user</em> who submitted the application * @return <em>user</em> who submitted the application

View File

@ -19,7 +19,6 @@
package org.apache.hadoop.yarn.api.records; package org.apache.hadoop.yarn.api.records;
import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate; import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceAudience.Public; import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Stable; import org.apache.hadoop.classification.InterfaceStability.Stable;
import org.apache.hadoop.classification.InterfaceStability.Unstable; import org.apache.hadoop.classification.InterfaceStability.Unstable;
@ -152,6 +151,28 @@ public interface ApplicationSubmissionContext {
@Stable @Stable
public void setAMContainerSpec(ContainerLaunchContext amContainer); public void setAMContainerSpec(ContainerLaunchContext amContainer);
/**
* Get if the RM should manage the execution of the AM.
* If true, then the RM
* will not allocate a container for the AM and start it. It will expect the
* AM to be launched and connect to the RM within the AM liveliness period and
* fail the app otherwise. The client should launch the AM only after the RM
* has ACCEPTED the application and changed the <code>YarnApplicationState</code>.
* Such apps will not be retried by the RM on app attempt failure.
* The default value is false.
* @return true if the AM is not managed by the RM
*/
@Public
@Unstable
public boolean getUnmanagedAM();
/**
* @param value true if RM should not manage the AM
*/
@Public
@Unstable
public void setUnmanagedAM(boolean value);
/** /**
* @return true if tokens should be canceled when the app completes. * @return true if tokens should be canceled when the app completes.
*/ */

View File

@ -43,5 +43,8 @@ public enum YarnApplicationState {
FAILED, FAILED,
/** Application which was terminated by a user or admin. */ /** Application which was terminated by a user or admin. */
KILLED KILLED,
/** Application has been accepted by the scheduler */
ACCEPTED
} }

View File

@ -18,12 +18,14 @@
package org.apache.hadoop.yarn.api.records.impl.pb; package org.apache.hadoop.yarn.api.records.impl.pb;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.YarnApplicationState; import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.api.records.ProtoBase; import org.apache.hadoop.yarn.api.records.ProtoBase;
import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport; import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationAttemptIdProto;
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationIdProto; import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationIdProto;
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationReportProto; import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationReportProto;
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationReportProtoOrBuilder; import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationReportProtoOrBuilder;
@ -39,6 +41,7 @@ implements ApplicationReport {
boolean viaProto = false; boolean viaProto = false;
ApplicationId applicationId; ApplicationId applicationId;
ApplicationAttemptId currentApplicationAttemptId;
public ApplicationReportPBImpl() { public ApplicationReportPBImpl() {
builder = ApplicationReportProto.newBuilder(); builder = ApplicationReportProto.newBuilder();
@ -72,6 +75,20 @@ implements ApplicationReport {
builder.setAppResourceUsage(convertToProtoFormat(appInfo)); builder.setAppResourceUsage(convertToProtoFormat(appInfo));
} }
@Override
public ApplicationAttemptId getCurrentApplicationAttemptId() {
if (this.currentApplicationAttemptId != null) {
return this.currentApplicationAttemptId;
}
ApplicationReportProtoOrBuilder p = viaProto ? proto : builder;
if (!p.hasCurrentApplicationAttemptId()) {
return null;
}
this.currentApplicationAttemptId = convertFromProtoFormat(p.getCurrentApplicationAttemptId());
return this.currentApplicationAttemptId;
}
@Override @Override
public ApplicationResourceUsageReport getApplicationResourceUsageReport() { public ApplicationResourceUsageReport getApplicationResourceUsageReport() {
ApplicationReportProtoOrBuilder p = viaProto ? proto : builder; ApplicationReportProtoOrBuilder p = viaProto ? proto : builder;
@ -198,6 +215,14 @@ implements ApplicationReport {
this.applicationId = applicationId; this.applicationId = applicationId;
} }
@Override
public void setCurrentApplicationAttemptId(ApplicationAttemptId applicationAttemptId) {
maybeInitBuilder();
if (applicationId == null)
builder.clearStatus();
this.currentApplicationAttemptId = applicationAttemptId;
}
@Override @Override
public void setTrackingUrl(String url) { public void setTrackingUrl(String url) {
maybeInitBuilder(); maybeInitBuilder();
@ -330,6 +355,11 @@ implements ApplicationReport {
builder.getApplicationId())) { builder.getApplicationId())) {
builder.setApplicationId(convertToProtoFormat(this.applicationId)); builder.setApplicationId(convertToProtoFormat(this.applicationId));
} }
if (this.currentApplicationAttemptId != null
&& !((ApplicationAttemptIdPBImpl) this.currentApplicationAttemptId).getProto().equals(
builder.getCurrentApplicationAttemptId())) {
builder.setCurrentApplicationAttemptId(convertToProtoFormat(this.currentApplicationAttemptId));
}
} }
private void mergeLocalToProto() { private void mergeLocalToProto() {
@ -351,6 +381,10 @@ implements ApplicationReport {
return ((ApplicationIdPBImpl) t).getProto(); return ((ApplicationIdPBImpl) t).getProto();
} }
private ApplicationAttemptIdProto convertToProtoFormat(ApplicationAttemptId t) {
return ((ApplicationAttemptIdPBImpl) t).getProto();
}
private ApplicationResourceUsageReport convertFromProtoFormat(ApplicationResourceUsageReportProto s) { private ApplicationResourceUsageReport convertFromProtoFormat(ApplicationResourceUsageReportProto s) {
return ProtoUtils.convertFromProtoFormat(s); return ProtoUtils.convertFromProtoFormat(s);
} }
@ -364,6 +398,11 @@ implements ApplicationReport {
return new ApplicationIdPBImpl(applicationId); return new ApplicationIdPBImpl(applicationId);
} }
private ApplicationAttemptIdPBImpl convertFromProtoFormat(
ApplicationAttemptIdProto applicationAttemptId) {
return new ApplicationAttemptIdPBImpl(applicationAttemptId);
}
private YarnApplicationState convertFromProtoFormat(YarnApplicationStateProto s) { private YarnApplicationState convertFromProtoFormat(YarnApplicationStateProto s) {
return ProtoUtils.convertFromProtoFormat(s); return ProtoUtils.convertFromProtoFormat(s);
} }

View File

@ -207,6 +207,19 @@ implements ApplicationSubmissionContext {
this.amContainer = amContainer; this.amContainer = amContainer;
} }
@Override
public boolean getUnmanagedAM() {
ApplicationSubmissionContextProtoOrBuilder p = viaProto ? proto : builder;
//There is a default so cancelTokens should never be null
return p.getUnmanagedAm();
}
@Override
public void setUnmanagedAM(boolean value) {
maybeInitBuilder();
builder.setUnmanagedAm(value);
}
@Override @Override
public boolean getCancelTokensWhenComplete() { public boolean getCancelTokensWhenComplete() {
ApplicationSubmissionContextProtoOrBuilder p = viaProto ? proto : builder; ApplicationSubmissionContextProtoOrBuilder p = viaProto ? proto : builder;

View File

@ -90,6 +90,7 @@ enum YarnApplicationStateProto {
FINISHED = 4; FINISHED = 4;
FAILED = 5; FAILED = 5;
KILLED = 6; KILLED = 6;
ACCEPTED = 7;
} }
enum FinalApplicationStatusProto { enum FinalApplicationStatusProto {
@ -170,6 +171,7 @@ message ApplicationReportProto {
optional FinalApplicationStatusProto final_application_status = 15; optional FinalApplicationStatusProto final_application_status = 15;
optional ApplicationResourceUsageReportProto app_resource_Usage = 16; optional ApplicationResourceUsageReportProto app_resource_Usage = 16;
optional string originalTrackingUrl = 17; optional string originalTrackingUrl = 17;
optional ApplicationAttemptIdProto currentApplicationAttemptId = 18;
} }
enum NodeStateProto { enum NodeStateProto {
@ -235,6 +237,7 @@ message ApplicationSubmissionContextProto {
optional PriorityProto priority = 5; optional PriorityProto priority = 5;
optional ContainerLaunchContextProto am_container_spec = 6; optional ContainerLaunchContextProto am_container_spec = 6;
optional bool cancel_tokens_when_complete = 7 [default = true]; optional bool cancel_tokens_when_complete = 7 [default = true];
optional bool unmanaged_am = 8 [default = false];
} }
enum ApplicationAccessTypeProto { enum ApplicationAccessTypeProto {

View File

@ -331,14 +331,16 @@ public class BuilderUtils {
} }
public static ApplicationReport newApplicationReport( public static ApplicationReport newApplicationReport(
ApplicationId applicationId, String user, String queue, String name, ApplicationId applicationId, ApplicationAttemptId applicationAttemptId,
String host, int rpcPort, String clientToken, YarnApplicationState state, String user, String queue, String name, String host, int rpcPort,
String diagnostics, String url, long startTime, long finishTime, String clientToken, YarnApplicationState state, String diagnostics,
FinalApplicationStatus finalStatus, ApplicationResourceUsageReport appResources, String url, long startTime, long finishTime,
String origTrackingUrl) { FinalApplicationStatus finalStatus,
ApplicationResourceUsageReport appResources, String origTrackingUrl) {
ApplicationReport report = recordFactory ApplicationReport report = recordFactory
.newRecordInstance(ApplicationReport.class); .newRecordInstance(ApplicationReport.class);
report.setApplicationId(applicationId); report.setApplicationId(applicationId);
report.setCurrentApplicationAttemptId(applicationAttemptId);
report.setUser(user); report.setUser(user);
report.setQueue(queue); report.setQueue(queue);
report.setName(name); report.setName(name);

View File

@ -19,18 +19,13 @@
package org.apache.hadoop.yarn; package org.apache.hadoop.yarn;
import java.util.Iterator; import java.util.Iterator;
import java.util.List;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.YarnApplicationState; import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
import org.apache.hadoop.yarn.util.Records; import org.apache.hadoop.yarn.util.Records;
import com.google.common.collect.Iterators; import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
/** /**
* Utilities to generate fake test apps * Utilities to generate fake test apps
@ -66,137 +61,6 @@ public class MockApps {
} }
} }
public static List<ApplicationReport> genApps(int n) {
List<ApplicationReport> list = Lists.newArrayList();
for (int i = 0; i < n; ++i) {
list.add(newApp(i));
}
return list;
}
public static ApplicationReport newApp(int i) {
final ApplicationId id = newAppID(i);
final YarnApplicationState state = newAppState();
final String user = newUserName();
final String name = newAppName();
final String queue = newQueue();
final FinalApplicationStatus finishState = FinalApplicationStatus.UNDEFINED;
return new ApplicationReport() {
private ApplicationResourceUsageReport appUsageReport;
@Override public ApplicationId getApplicationId() { return id; }
@Override public String getUser() { return user; }
@Override public String getName() { return name; }
@Override public YarnApplicationState getYarnApplicationState() { return state; }
@Override public String getQueue() { return queue; }
@Override public String getTrackingUrl() { return ""; }
@Override public String getOriginalTrackingUrl() { return ""; }
@Override public FinalApplicationStatus getFinalApplicationStatus() { return finishState; }
@Override
public ApplicationResourceUsageReport getApplicationResourceUsageReport() {
return this.appUsageReport;
}
public void setApplicationId(ApplicationId applicationId) {
// TODO Auto-generated method stub
}
@Override
public void setTrackingUrl(String url) {
// TODO Auto-generated method stub
}
@Override public void setOriginalTrackingUrl(String url) { }
@Override
public void setApplicationResourceUsageReport(ApplicationResourceUsageReport appResources) {
this.appUsageReport = appResources;
}
@Override
public void setName(String name) {
// TODO Auto-generated method stub
}
@Override
public void setQueue(String queue) {
// TODO Auto-generated method stub
}
@Override
public void setYarnApplicationState(YarnApplicationState state) {
// TODO Auto-generated method stub
}
@Override
public void setUser(String user) {
// TODO Auto-generated method stub
}
@Override
public String getDiagnostics() {
// TODO Auto-generated method stub
return null;
}
@Override
public void setDiagnostics(String diagnostics) {
// TODO Auto-generated method stub
}
@Override
public String getHost() {
// TODO Auto-generated method stub
return null;
}
@Override
public void setHost(String host) {
// TODO Auto-generated method stub
}
@Override
public int getRpcPort() {
// TODO Auto-generated method stub
return 0;
}
@Override
public void setRpcPort(int rpcPort) {
// TODO Auto-generated method stub
}
@Override
public String getClientToken() {
// TODO Auto-generated method stub
return null;
}
@Override
public void setClientToken(String clientToken) {
// TODO Auto-generated method stub
}
@Override
public long getStartTime() {
// TODO Auto-generated method stub
return 0;
}
@Override
public void setStartTime(long startTime) {
// TODO Auto-generated method stub
}
@Override
public long getFinishTime() {
// TODO Auto-generated method stub
return 0;
}
@Override
public void setFinishTime(long finishTime) {
// TODO Auto-generated method stub
}
@Override
public void setFinalApplicationStatus(FinalApplicationStatus finishState) {
// TODO Auto-generated method stub
}
};
}
public static ApplicationId newAppID(int i) { public static ApplicationId newAppID(int i) {
ApplicationId id = Records.newRecord(ApplicationId.class); ApplicationId id = Records.newRecord(ApplicationId.class);
id.setClusterTimestamp(TS); id.setClusterTimestamp(TS);

View File

@ -333,8 +333,9 @@ public class RMAppImpl implements RMApp {
case NEW: case NEW:
return YarnApplicationState.NEW; return YarnApplicationState.NEW;
case SUBMITTED: case SUBMITTED:
case ACCEPTED:
return YarnApplicationState.SUBMITTED; return YarnApplicationState.SUBMITTED;
case ACCEPTED:
return YarnApplicationState.ACCEPTED;
case RUNNING: case RUNNING:
return YarnApplicationState.RUNNING; return YarnApplicationState.RUNNING;
case FINISHED: case FINISHED:
@ -403,12 +404,12 @@ public class RMAppImpl implements RMApp {
} else { } else {
appUsageReport = DUMMY_APPLICATION_RESOURCE_USAGE_REPORT; appUsageReport = DUMMY_APPLICATION_RESOURCE_USAGE_REPORT;
} }
return BuilderUtils.newApplicationReport(this.applicationId, this.user, return BuilderUtils.newApplicationReport(this.applicationId,
this.queue, this.name, host, rpcPort, clientToken, this.currentAttempt.getAppAttemptId(), this.user, this.queue,
createApplicationState(this.stateMachine.getCurrentState()), this.name, host, rpcPort, clientToken,
diags, trackingUrl, createApplicationState(this.stateMachine.getCurrentState()), diags,
this.startTime, this.finishTime, finishState, appUsageReport, trackingUrl, this.startTime, this.finishTime, finishState,
origTrackingUrl); appUsageReport, origTrackingUrl);
} finally { } finally {
this.readLock.unlock(); this.readLock.unlock();
} }
@ -599,21 +600,32 @@ public class RMAppImpl implements RMApp {
@Override @Override
public RMAppState transition(RMAppImpl app, RMAppEvent event) { public RMAppState transition(RMAppImpl app, RMAppEvent event) {
RMAppFailedAttemptEvent failedEvent = ((RMAppFailedAttemptEvent)event); RMAppFailedAttemptEvent failedEvent = ((RMAppFailedAttemptEvent) event);
if (app.attempts.size() == app.maxRetries) { boolean retryApp = true;
String msg = "Application " + app.getApplicationId() String msg = null;
+ " failed " + app.maxRetries if (app.submissionContext.getUnmanagedAM()) {
+ " times due to " + failedEvent.getDiagnostics() // RM does not manage the AM. Do not retry
retryApp = false;
msg = "Unmanaged application " + app.getApplicationId()
+ " failed due to " + failedEvent.getDiagnostics()
+ ". Failing the application."; + ". Failing the application.";
} else if (app.attempts.size() == app.maxRetries) {
retryApp = false;
msg = "Application " + app.getApplicationId() + " failed "
+ app.maxRetries + " times due to " + failedEvent.getDiagnostics()
+ ". Failing the application.";
}
if (retryApp) {
app.createNewAttempt();
return initialState;
} else {
LOG.info(msg); LOG.info(msg);
app.diagnostics.append(msg); app.diagnostics.append(msg);
// Inform the node for app-finish // Inform the node for app-finish
FINAL_TRANSITION.transition(app, event); FINAL_TRANSITION.transition(app, event);
return RMAppState.FAILED; return RMAppState.FAILED;
} }
app.createNewAttempt();
return initialState;
} }
} }

View File

@ -143,15 +143,23 @@ public class RMAppAttemptImpl implements RMAppAttempt {
.addTransition(RMAppAttemptState.NEW, RMAppAttemptState.KILLED, .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.KILLED,
RMAppAttemptEventType.KILL, RMAppAttemptEventType.KILL,
new BaseFinalTransition(RMAppAttemptState.KILLED)) new BaseFinalTransition(RMAppAttemptState.KILLED))
.addTransition(RMAppAttemptState.NEW, RMAppAttemptState.FAILED,
RMAppAttemptEventType.REGISTERED,
new UnexpectedAMRegisteredTransition())
// Transitions from SUBMITTED state // Transitions from SUBMITTED state
.addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.FAILED, .addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.FAILED,
RMAppAttemptEventType.APP_REJECTED, new AppRejectedTransition()) RMAppAttemptEventType.APP_REJECTED, new AppRejectedTransition())
.addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.SCHEDULED, .addTransition(RMAppAttemptState.SUBMITTED,
RMAppAttemptEventType.APP_ACCEPTED, new ScheduleTransition()) EnumSet.of(RMAppAttemptState.LAUNCHED, RMAppAttemptState.SCHEDULED),
RMAppAttemptEventType.APP_ACCEPTED,
new ScheduleTransition())
.addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.KILLED, .addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.KILLED,
RMAppAttemptEventType.KILL, RMAppAttemptEventType.KILL,
new BaseFinalTransition(RMAppAttemptState.KILLED)) new BaseFinalTransition(RMAppAttemptState.KILLED))
.addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.FAILED,
RMAppAttemptEventType.REGISTERED,
new UnexpectedAMRegisteredTransition())
// Transitions from SCHEDULED State // Transitions from SCHEDULED State
.addTransition(RMAppAttemptState.SCHEDULED, .addTransition(RMAppAttemptState.SCHEDULED,
@ -583,9 +591,11 @@ public class RMAppAttemptImpl implements RMAppAttempt {
private static final List<ResourceRequest> EMPTY_CONTAINER_REQUEST_LIST = private static final List<ResourceRequest> EMPTY_CONTAINER_REQUEST_LIST =
new ArrayList<ResourceRequest>(); new ArrayList<ResourceRequest>();
private static final class ScheduleTransition extends BaseTransition { private static final class ScheduleTransition
implements
MultipleArcTransition<RMAppAttemptImpl, RMAppAttemptEvent, RMAppAttemptState> {
@Override @Override
public void transition(RMAppAttemptImpl appAttempt, public RMAppAttemptState transition(RMAppAttemptImpl appAttempt,
RMAppAttemptEvent event) { RMAppAttemptEvent event) {
// Send the acceptance to the app // Send the acceptance to the app
@ -593,17 +603,27 @@ public class RMAppAttemptImpl implements RMAppAttempt {
.getApplicationAttemptId().getApplicationId(), .getApplicationAttemptId().getApplicationId(),
RMAppEventType.APP_ACCEPTED)); RMAppEventType.APP_ACCEPTED));
if (!appAttempt.submissionContext.getUnmanagedAM()) {
// Request a container for the AM. // Request a container for the AM.
ResourceRequest request = BuilderUtils.newResourceRequest( ResourceRequest request = BuilderUtils.newResourceRequest(
AM_CONTAINER_PRIORITY, "*", appAttempt.submissionContext AM_CONTAINER_PRIORITY, "*", appAttempt.submissionContext
.getAMContainerSpec().getResource(), 1); .getAMContainerSpec().getResource(), 1);
Allocation amContainerAllocation = Allocation amContainerAllocation = appAttempt.scheduler.allocate(
appAttempt.scheduler.allocate(appAttempt.applicationAttemptId, appAttempt.applicationAttemptId,
Collections.singletonList(request), EMPTY_CONTAINER_RELEASE_LIST); Collections.singletonList(request), EMPTY_CONTAINER_RELEASE_LIST);
if (amContainerAllocation != null if (amContainerAllocation != null
&& amContainerAllocation.getContainers() != null) { && amContainerAllocation.getContainers() != null) {
assert(amContainerAllocation.getContainers().size() == 0); assert (amContainerAllocation.getContainers().size() == 0);
}
return RMAppAttemptState.SCHEDULED;
} else {
// RM not allocating container. AM is self launched.
// Directly go to LAUNCHED state
// Register with AMLivelinessMonitor
appAttempt.rmContext.getAMLivelinessMonitor().register(
appAttempt.applicationAttemptId);
return RMAppAttemptState.LAUNCHED;
} }
} }
} }
@ -811,11 +831,30 @@ public class RMAppAttemptImpl implements RMAppAttempt {
appAttempt.rmContext.getAMLivelinessMonitor().unregister( appAttempt.rmContext.getAMLivelinessMonitor().unregister(
appAttempt.getAppAttemptId()); appAttempt.getAppAttemptId());
if(!appAttempt.submissionContext.getUnmanagedAM()) {
// Tell the launcher to cleanup. // Tell the launcher to cleanup.
appAttempt.eventHandler.handle(new AMLauncherEvent( appAttempt.eventHandler.handle(new AMLauncherEvent(
AMLauncherEventType.CLEANUP, appAttempt)); AMLauncherEventType.CLEANUP, appAttempt));
} }
} }
}
private static class UnexpectedAMRegisteredTransition extends
BaseFinalTransition {
public UnexpectedAMRegisteredTransition() {
super(RMAppAttemptState.FAILED);
}
@Override
public void transition(RMAppAttemptImpl appAttempt, RMAppAttemptEvent event) {
assert appAttempt.submissionContext.getUnmanagedAM();
appAttempt
.setDiagnostics("Unmanaged AM must register after AM attempt reaches LAUNCHED state.");
super.transition(appAttempt, event);
}
}
private static final class StatusUpdateTransition extends private static final class StatusUpdateTransition extends
BaseTransition { BaseTransition {
@ -884,8 +923,11 @@ public class RMAppAttemptImpl implements RMAppAttempt {
// Is this container the AmContainer? If the finished container is same as // Is this container the AmContainer? If the finished container is same as
// the AMContainer, AppAttempt fails // the AMContainer, AppAttempt fails
if (appAttempt.masterContainer.getId().equals( if (appAttempt.masterContainer != null
&& appAttempt.masterContainer.getId().equals(
containerStatus.getContainerId())) { containerStatus.getContainerId())) {
// container associated with AM. must not be unmanaged
assert appAttempt.submissionContext.getUnmanagedAM() == false;
// Setup diagnostic message // Setup diagnostic message
appAttempt.diagnostics.append("AM Container for " + appAttempt.diagnostics.append("AM Container for " +
appAttempt.getAppAttemptId() + " exited with " + appAttempt.getAppAttemptId() + " exited with " +

View File

@ -31,6 +31,7 @@ import org.apache.hadoop.yarn.MockApps;
import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationSubmissionContextPBImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.AsyncDispatcher; import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.EventHandler; import org.apache.hadoop.yarn.event.EventHandler;
@ -131,7 +132,7 @@ public class TestRMAppTransitions {
rmDispatcher.start(); rmDispatcher.start();
} }
protected RMApp createNewTestApp() { protected RMApp createNewTestApp(ApplicationSubmissionContext submissionContext) {
ApplicationId applicationId = MockApps.newAppID(appId++); ApplicationId applicationId = MockApps.newAppID(appId++);
String user = MockApps.newUserName(); String user = MockApps.newUserName();
String name = MockApps.newAppName(); String name = MockApps.newAppName();
@ -139,13 +140,16 @@ public class TestRMAppTransitions {
Configuration conf = new YarnConfiguration(); Configuration conf = new YarnConfiguration();
// ensure max retries set to known value // ensure max retries set to known value
conf.setInt(YarnConfiguration.RM_AM_MAX_RETRIES, maxRetries); conf.setInt(YarnConfiguration.RM_AM_MAX_RETRIES, maxRetries);
ApplicationSubmissionContext submissionContext = null;
String clientTokenStr = "bogusstring"; String clientTokenStr = "bogusstring";
ApplicationStore appStore = mock(ApplicationStore.class); ApplicationStore appStore = mock(ApplicationStore.class);
YarnScheduler scheduler = mock(YarnScheduler.class); YarnScheduler scheduler = mock(YarnScheduler.class);
ApplicationMasterService masterService = ApplicationMasterService masterService =
new ApplicationMasterService(rmContext, scheduler); new ApplicationMasterService(rmContext, scheduler);
if(submissionContext == null) {
submissionContext = new ApplicationSubmissionContextPBImpl();
}
RMApp application = new RMAppImpl(applicationId, rmContext, RMApp application = new RMAppImpl(applicationId, rmContext,
conf, name, user, conf, name, user,
queue, submissionContext, clientTokenStr, queue, submissionContext, clientTokenStr,
@ -235,8 +239,9 @@ public class TestRMAppTransitions {
diag.toString().matches(regex)); diag.toString().matches(regex));
} }
protected RMApp testCreateAppSubmitted() throws IOException { protected RMApp testCreateAppSubmitted(
RMApp application = createNewTestApp(); ApplicationSubmissionContext submissionContext) throws IOException {
RMApp application = createNewTestApp(submissionContext);
// NEW => SUBMITTED event RMAppEventType.START // NEW => SUBMITTED event RMAppEventType.START
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), RMAppEventType.START); new RMAppEvent(application.getApplicationId(), RMAppEventType.START);
@ -246,8 +251,9 @@ public class TestRMAppTransitions {
return application; return application;
} }
protected RMApp testCreateAppAccepted() throws IOException { protected RMApp testCreateAppAccepted(
RMApp application = testCreateAppSubmitted(); ApplicationSubmissionContext submissionContext) throws IOException {
RMApp application = testCreateAppSubmitted(submissionContext);
// SUBMITTED => ACCEPTED event RMAppEventType.APP_ACCEPTED // SUBMITTED => ACCEPTED event RMAppEventType.APP_ACCEPTED
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), new RMAppEvent(application.getApplicationId(),
@ -258,8 +264,9 @@ public class TestRMAppTransitions {
return application; return application;
} }
protected RMApp testCreateAppRunning() throws IOException { protected RMApp testCreateAppRunning(
RMApp application = testCreateAppAccepted(); ApplicationSubmissionContext submissionContext) throws IOException {
RMApp application = testCreateAppAccepted(submissionContext);
// ACCEPTED => RUNNING event RMAppEventType.ATTEMPT_REGISTERED // ACCEPTED => RUNNING event RMAppEventType.ATTEMPT_REGISTERED
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), new RMAppEvent(application.getApplicationId(),
@ -271,8 +278,9 @@ public class TestRMAppTransitions {
return application; return application;
} }
protected RMApp testCreateAppFinished() throws IOException { protected RMApp testCreateAppFinished(
RMApp application = testCreateAppRunning(); ApplicationSubmissionContext submissionContext) throws IOException {
RMApp application = testCreateAppRunning(submissionContext);
// RUNNING => FINISHED event RMAppEventType.ATTEMPT_FINISHED // RUNNING => FINISHED event RMAppEventType.ATTEMPT_FINISHED
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), new RMAppEvent(application.getApplicationId(),
@ -285,17 +293,38 @@ public class TestRMAppTransitions {
return application; return application;
} }
@Test
public void testUnmanagedApp() throws IOException {
ApplicationSubmissionContext subContext = new ApplicationSubmissionContextPBImpl();
subContext.setUnmanagedAM(true);
// test success path
LOG.info("--- START: testUnmanagedAppSuccessPath ---");
testCreateAppFinished(subContext);
// test app fails after 1 app attempt failure
LOG.info("--- START: testUnmanagedAppFailPath ---");
RMApp application = testCreateAppRunning(subContext);
RMAppEvent event = new RMAppFailedAttemptEvent(
application.getApplicationId(), RMAppEventType.ATTEMPT_FAILED, "");
application.handle(event);
RMAppAttempt appAttempt = application.getCurrentAppAttempt();
Assert.assertEquals(1, appAttempt.getAppAttemptId().getAttemptId());
assertFailed(application,
".*Unmanaged application.*Failing the application.*");
}
@Test @Test
public void testAppSuccessPath() throws IOException { public void testAppSuccessPath() throws IOException {
LOG.info("--- START: testAppSuccessPath ---"); LOG.info("--- START: testAppSuccessPath ---");
testCreateAppFinished(); testCreateAppFinished(null);
} }
@Test @Test
public void testAppNewKill() throws IOException { public void testAppNewKill() throws IOException {
LOG.info("--- START: testAppNewKill ---"); LOG.info("--- START: testAppNewKill ---");
RMApp application = createNewTestApp(); RMApp application = createNewTestApp(null);
// NEW => KILLED event RMAppEventType.KILL // NEW => KILLED event RMAppEventType.KILL
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL); new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL);
@ -307,7 +336,7 @@ public class TestRMAppTransitions {
public void testAppNewReject() throws IOException { public void testAppNewReject() throws IOException {
LOG.info("--- START: testAppNewReject ---"); LOG.info("--- START: testAppNewReject ---");
RMApp application = createNewTestApp(); RMApp application = createNewTestApp(null);
// NEW => FAILED event RMAppEventType.APP_REJECTED // NEW => FAILED event RMAppEventType.APP_REJECTED
String rejectedText = "Test Application Rejected"; String rejectedText = "Test Application Rejected";
RMAppEvent event = RMAppEvent event =
@ -320,7 +349,7 @@ public class TestRMAppTransitions {
public void testAppSubmittedRejected() throws IOException { public void testAppSubmittedRejected() throws IOException {
LOG.info("--- START: testAppSubmittedRejected ---"); LOG.info("--- START: testAppSubmittedRejected ---");
RMApp application = testCreateAppSubmitted(); RMApp application = testCreateAppSubmitted(null);
// SUBMITTED => FAILED event RMAppEventType.APP_REJECTED // SUBMITTED => FAILED event RMAppEventType.APP_REJECTED
String rejectedText = "app rejected"; String rejectedText = "app rejected";
RMAppEvent event = RMAppEvent event =
@ -333,7 +362,7 @@ public class TestRMAppTransitions {
public void testAppSubmittedKill() throws IOException { public void testAppSubmittedKill() throws IOException {
LOG.info("--- START: testAppSubmittedKill---"); LOG.info("--- START: testAppSubmittedKill---");
RMApp application = testCreateAppAccepted(); RMApp application = testCreateAppAccepted(null);
// SUBMITTED => KILLED event RMAppEventType.KILL // SUBMITTED => KILLED event RMAppEventType.KILL
RMAppEvent event = new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL); RMAppEvent event = new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL);
this.rmContext.getRMApps().putIfAbsent(application.getApplicationId(), application); this.rmContext.getRMApps().putIfAbsent(application.getApplicationId(), application);
@ -345,7 +374,7 @@ public class TestRMAppTransitions {
public void testAppAcceptedFailed() throws IOException { public void testAppAcceptedFailed() throws IOException {
LOG.info("--- START: testAppAcceptedFailed ---"); LOG.info("--- START: testAppAcceptedFailed ---");
RMApp application = testCreateAppAccepted(); RMApp application = testCreateAppAccepted(null);
// ACCEPTED => ACCEPTED event RMAppEventType.RMAppEventType.ATTEMPT_FAILED // ACCEPTED => ACCEPTED event RMAppEventType.RMAppEventType.ATTEMPT_FAILED
for (int i=1; i<maxRetries; i++) { for (int i=1; i<maxRetries; i++) {
RMAppEvent event = RMAppEvent event =
@ -374,7 +403,7 @@ public class TestRMAppTransitions {
public void testAppAcceptedKill() throws IOException { public void testAppAcceptedKill() throws IOException {
LOG.info("--- START: testAppAcceptedKill ---"); LOG.info("--- START: testAppAcceptedKill ---");
RMApp application = testCreateAppAccepted(); RMApp application = testCreateAppAccepted(null);
// ACCEPTED => KILLED event RMAppEventType.KILL // ACCEPTED => KILLED event RMAppEventType.KILL
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL); new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL);
@ -386,7 +415,7 @@ public class TestRMAppTransitions {
public void testAppRunningKill() throws IOException { public void testAppRunningKill() throws IOException {
LOG.info("--- START: testAppRunningKill ---"); LOG.info("--- START: testAppRunningKill ---");
RMApp application = testCreateAppRunning(); RMApp application = testCreateAppRunning(null);
// RUNNING => KILLED event RMAppEventType.KILL // RUNNING => KILLED event RMAppEventType.KILL
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL); new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL);
@ -398,7 +427,7 @@ public class TestRMAppTransitions {
public void testAppRunningFailed() throws IOException { public void testAppRunningFailed() throws IOException {
LOG.info("--- START: testAppRunningFailed ---"); LOG.info("--- START: testAppRunningFailed ---");
RMApp application = testCreateAppRunning(); RMApp application = testCreateAppRunning(null);
RMAppAttempt appAttempt = application.getCurrentAppAttempt(); RMAppAttempt appAttempt = application.getCurrentAppAttempt();
int expectedAttemptId = 1; int expectedAttemptId = 1;
Assert.assertEquals(expectedAttemptId, Assert.assertEquals(expectedAttemptId,
@ -444,7 +473,7 @@ public class TestRMAppTransitions {
public void testAppFinishedFinished() throws IOException { public void testAppFinishedFinished() throws IOException {
LOG.info("--- START: testAppFinishedFinished ---"); LOG.info("--- START: testAppFinishedFinished ---");
RMApp application = testCreateAppFinished(); RMApp application = testCreateAppFinished(null);
// FINISHED => FINISHED event RMAppEventType.KILL // FINISHED => FINISHED event RMAppEventType.KILL
RMAppEvent event = RMAppEvent event =
new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL); new RMAppEvent(application.getApplicationId(), RMAppEventType.KILL);
@ -460,7 +489,7 @@ public class TestRMAppTransitions {
public void testAppKilledKilled() throws IOException { public void testAppKilledKilled() throws IOException {
LOG.info("--- START: testAppKilledKilled ---"); LOG.info("--- START: testAppKilledKilled ---");
RMApp application = testCreateAppRunning(); RMApp application = testCreateAppRunning(null);
// RUNNING => KILLED event RMAppEventType.KILL // RUNNING => KILLED event RMAppEventType.KILL
RMAppEvent event = RMAppEvent event =

View File

@ -39,6 +39,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.Container; import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.event.AsyncDispatcher; import org.apache.hadoop.yarn.event.AsyncDispatcher;
@ -56,7 +57,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppFailedAttemptEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppFailedAttemptEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppRejectedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptContainerAcquiredEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptContainerAllocatedEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptContainerAllocatedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptContainerFinishedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptLaunchFailedEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptLaunchFailedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptRegistrationEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptRegistrationEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptRejectedEvent; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptRejectedEvent;
@ -83,6 +86,7 @@ public class TestRMAppAttemptTransitions {
private YarnScheduler scheduler; private YarnScheduler scheduler;
private ApplicationMasterService masterService; private ApplicationMasterService masterService;
private ApplicationMasterLauncher applicationMasterLauncher; private ApplicationMasterLauncher applicationMasterLauncher;
private AMLivelinessMonitor amLivelinessMonitor;
private RMApp application; private RMApp application;
private RMAppAttempt applicationAttempt; private RMAppAttempt applicationAttempt;
@ -136,13 +140,16 @@ public class TestRMAppAttemptTransitions {
private static int appId = 1; private static int appId = 1;
private ApplicationSubmissionContext submissionContext = null;
private boolean unmanagedAM;
@Before @Before
public void setUp() throws Exception { public void setUp() throws Exception {
InlineDispatcher rmDispatcher = new InlineDispatcher(); InlineDispatcher rmDispatcher = new InlineDispatcher();
ContainerAllocationExpirer containerAllocationExpirer = ContainerAllocationExpirer containerAllocationExpirer =
mock(ContainerAllocationExpirer.class); mock(ContainerAllocationExpirer.class);
AMLivelinessMonitor amLivelinessMonitor = mock(AMLivelinessMonitor.class); amLivelinessMonitor = mock(AMLivelinessMonitor.class);
rmContext = rmContext =
new RMContextImpl(new MemStore(), rmDispatcher, new RMContextImpl(new MemStore(), rmDispatcher,
containerAllocationExpirer, amLivelinessMonitor, null, containerAllocationExpirer, amLivelinessMonitor, null,
@ -174,8 +181,7 @@ public class TestRMAppAttemptTransitions {
final String user = MockApps.newUserName(); final String user = MockApps.newUserName();
final String queue = MockApps.newQueue(); final String queue = MockApps.newQueue();
ApplicationSubmissionContext submissionContext = submissionContext = mock(ApplicationSubmissionContext.class);
mock(ApplicationSubmissionContext.class);
when(submissionContext.getUser()).thenReturn(user); when(submissionContext.getUser()).thenReturn(user);
when(submissionContext.getQueue()).thenReturn(queue); when(submissionContext.getQueue()).thenReturn(queue);
ContainerLaunchContext amContainerSpec = mock(ContainerLaunchContext.class); ContainerLaunchContext amContainerSpec = mock(ContainerLaunchContext.class);
@ -183,6 +189,8 @@ public class TestRMAppAttemptTransitions {
when(amContainerSpec.getResource()).thenReturn(resource); when(amContainerSpec.getResource()).thenReturn(resource);
when(submissionContext.getAMContainerSpec()).thenReturn(amContainerSpec); when(submissionContext.getAMContainerSpec()).thenReturn(amContainerSpec);
unmanagedAM = false;
application = mock(RMApp.class); application = mock(RMApp.class);
applicationAttempt = applicationAttempt =
new RMAppAttemptImpl(applicationAttemptId, null, rmContext, scheduler, new RMAppAttemptImpl(applicationAttemptId, null, rmContext, scheduler,
@ -247,7 +255,8 @@ public class TestRMAppAttemptTransitions {
assertEquals(0, applicationAttempt.getRanNodes().size()); assertEquals(0, applicationAttempt.getRanNodes().size());
assertNull(applicationAttempt.getFinalApplicationStatus()); assertNull(applicationAttempt.getFinalApplicationStatus());
// Check events // this works for unmanaged and managed AM's because this is actually doing
// verify(application).handle(anyObject());
verify(application).handle(any(RMAppRejectedEvent.class)); verify(application).handle(any(RMAppRejectedEvent.class));
} }
@ -269,9 +278,24 @@ public class TestRMAppAttemptTransitions {
/** /**
* {@link RMAppAttemptState#SCHEDULED} * {@link RMAppAttemptState#SCHEDULED}
*/ */
@SuppressWarnings("unchecked")
private void testAppAttemptScheduledState() { private void testAppAttemptScheduledState() {
assertEquals(RMAppAttemptState.SCHEDULED, RMAppAttemptState expectedState;
int expectedAllocateCount;
if(unmanagedAM) {
expectedState = RMAppAttemptState.LAUNCHED;
expectedAllocateCount = 0;
} else {
expectedState = RMAppAttemptState.SCHEDULED;
expectedAllocateCount = 1;
}
assertEquals(expectedState,
applicationAttempt.getAppAttemptState()); applicationAttempt.getAppAttemptState());
verify(scheduler, times(expectedAllocateCount)).
allocate(any(ApplicationAttemptId.class),
any(List.class), any(List.class));
assertEquals(0,applicationAttempt.getJustFinishedContainers().size()); assertEquals(0,applicationAttempt.getJustFinishedContainers().size());
assertNull(applicationAttempt.getMasterContainer()); assertNull(applicationAttempt.getMasterContainer());
assertEquals(0.0, (double)applicationAttempt.getProgress(), 0.0001); assertEquals(0.0, (double)applicationAttempt.getProgress(), 0.0001);
@ -280,9 +304,6 @@ public class TestRMAppAttemptTransitions {
// Check events // Check events
verify(application).handle(any(RMAppEvent.class)); verify(application).handle(any(RMAppEvent.class));
verify(scheduler).
allocate(any(ApplicationAttemptId.class),
any(List.class), any(List.class));
} }
/** /**
@ -351,14 +372,16 @@ public class TestRMAppAttemptTransitions {
private void testAppAttemptFinishedState(Container container, private void testAppAttemptFinishedState(Container container,
FinalApplicationStatus finalStatus, FinalApplicationStatus finalStatus,
String trackingUrl, String trackingUrl,
String diagnostics) { String diagnostics,
int finishedContainerCount) {
assertEquals(RMAppAttemptState.FINISHED, assertEquals(RMAppAttemptState.FINISHED,
applicationAttempt.getAppAttemptState()); applicationAttempt.getAppAttemptState());
assertEquals(diagnostics, applicationAttempt.getDiagnostics()); assertEquals(diagnostics, applicationAttempt.getDiagnostics());
assertEquals(trackingUrl, applicationAttempt.getOriginalTrackingUrl()); assertEquals(trackingUrl, applicationAttempt.getOriginalTrackingUrl());
assertEquals("null/proxy/"+applicationAttempt.getAppAttemptId(). assertEquals("null/proxy/"+applicationAttempt.getAppAttemptId().
getApplicationId()+"/", applicationAttempt.getTrackingUrl()); getApplicationId()+"/", applicationAttempt.getTrackingUrl());
assertEquals(0,applicationAttempt.getJustFinishedContainers().size()); assertEquals(finishedContainerCount, applicationAttempt
.getJustFinishedContainers().size());
assertEquals(container, applicationAttempt.getMasterContainer()); assertEquals(container, applicationAttempt.getMasterContainer());
assertEquals(finalStatus, applicationAttempt.getFinalApplicationStatus()); assertEquals(finalStatus, applicationAttempt.getFinalApplicationStatus());
} }
@ -425,6 +448,49 @@ public class TestRMAppAttemptTransitions {
testAppAttemptRunningState(container, host, rpcPort, trackingUrl); testAppAttemptRunningState(container, host, rpcPort, trackingUrl);
} }
@Test
public void testUnmanagedAMSuccess() {
unmanagedAM = true;
when(submissionContext.getUnmanagedAM()).thenReturn(true);
// submit AM and check it goes to LAUNCHED state
scheduleApplicationAttempt();
testAppAttemptLaunchedState(null);
verify(amLivelinessMonitor, times(1)).register(
applicationAttempt.getAppAttemptId());
// launch AM
runApplicationAttempt(null, "host", 8042, "oldtrackingurl");
// complete a container
applicationAttempt.handle(new RMAppAttemptContainerAcquiredEvent(
applicationAttempt.getAppAttemptId(), mock(Container.class)));
applicationAttempt.handle(new RMAppAttemptContainerFinishedEvent(
applicationAttempt.getAppAttemptId(), mock(ContainerStatus.class)));
// complete AM
String trackingUrl = "mytrackingurl";
String diagnostics = "Successful";
FinalApplicationStatus finalStatus = FinalApplicationStatus.SUCCEEDED;
applicationAttempt.handle(new RMAppAttemptUnregistrationEvent(
applicationAttempt.getAppAttemptId(), trackingUrl, finalStatus,
diagnostics));
testAppAttemptFinishedState(null, finalStatus, trackingUrl, diagnostics, 1);
}
@Test
public void testUnmanagedAMUnexpectedRegistration() {
unmanagedAM = true;
when(submissionContext.getUnmanagedAM()).thenReturn(true);
// submit AM and check it goes to SUBMITTED state
submitApplicationAttempt();
assertEquals(RMAppAttemptState.SUBMITTED,
applicationAttempt.getAppAttemptState());
// launch AM and verify attempt failed
applicationAttempt.handle(new RMAppAttemptRegistrationEvent(
applicationAttempt.getAppAttemptId(), "host", 8042, "oldtrackingurl"));
testAppAttemptSubmittedToFailedState("Unmanaged AM must register after AM attempt reaches LAUNCHED state.");
}
@Test @Test
public void testNewToKilled() { public void testNewToKilled() {
@ -499,7 +565,7 @@ public class TestRMAppAttemptTransitions {
applicationAttempt.getAppAttemptId(), applicationAttempt.getAppAttemptId(),
trackingUrl, finalStatus, diagnostics)); trackingUrl, finalStatus, diagnostics));
testAppAttemptFinishedState(amContainer, finalStatus, testAppAttemptFinishedState(amContainer, finalStatus,
trackingUrl, diagnostics); trackingUrl, diagnostics, 0);
} }
@ -516,7 +582,7 @@ public class TestRMAppAttemptTransitions {
applicationAttempt.getAppAttemptId(), applicationAttempt.getAppAttemptId(),
trackingUrl, finalStatus, diagnostics)); trackingUrl, finalStatus, diagnostics));
testAppAttemptFinishedState(amContainer, finalStatus, testAppAttemptFinishedState(amContainer, finalStatus,
trackingUrl, diagnostics); trackingUrl, diagnostics, 0);
} }
} }