Skip to content

YARN-11180. Refactor some code of getNewApplication, submitApplication etc. #4618

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
goiri merged 13 commits into from
Jul 29, 2022
Merged

YARN-11180. Refactor some code of getNewApplication, submitApplication etc. #4618

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
bfef1dc
YARN-11180. Refactor some code of getNewApplication, submitApplicatio…
Jul 24, 2022
a8150e4
YARN-11180. Fix CheckStyle.
Jul 24, 2022
975d9bf
YARN-11180. Fix CheckStyle.
Jul 24, 2022
24234ac
YARN-11180. Fix CheckStyle.
Jul 24, 2022
15ab9a9
YARN-11180. Fix CheckStyle.
Jul 24, 2022
390f1f2
YARN-11180. Fix CheckStyle.
Jul 25, 2022
fe41b76
YARN-11180. Fix CheckStyle.
Jul 25, 2022
41b7d8a
YARN-11180. Fix CheckStyle.
Jul 25, 2022
ea6a4a3
Merge branch 'trunk' into YARN-11180
slfan1989 Jul 26, 2022
07a9043
YARN-11180. Fix CheckStyle.
Jul 26, 2022
730ea24
YARN-11180. Fix CheckStyle.
Jul 26, 2022
515f422
YARN-11180. Fix CheckStyle.
Jul 27, 2022
b2aa446
Merge branch 'apache:trunk' into YARN-11180
slfan1989 Jul 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 46 additions & 49 deletions .../main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,35 +227,29 @@ protected ApplicationClientProtocol getClientRMProxyForSubCluster(
ApplicationClientProtocol clientRMProxy = null;
try {
boolean serviceAuthEnabled = getConf().getBoolean(
CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, false);
CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, false);
UserGroupInformation realUser = user;
if (serviceAuthEnabled) {
realUser = UserGroupInformation.createProxyUser(
user.getShortUserName(), UserGroupInformation.getLoginUser());
user.getShortUserName(), UserGroupInformation.getLoginUser());
}
clientRMProxy = FederationProxyProviderUtil.createRMProxy(getConf(),
ApplicationClientProtocol.class, subClusterId, realUser);
} catch (Exception e) {
RouterServerUtil.logAndThrowException(
"Unable to create the interface to reach the SubCluster "
+ subClusterId,
e);
"Unable to create the interface to reach the SubCluster " + subClusterId, e);
}

clientRMProxies.put(subClusterId, clientRMProxy);
return clientRMProxy;
}

private SubClusterId getRandomActiveSubCluster(
Map<SubClusterId, SubClusterInfo> activeSubclusters)
throws YarnException {

if (activeSubclusters == null || activeSubclusters.size() < 1) {
Map<SubClusterId, SubClusterInfo> activeSubClusters) throws YarnException {
if (activeSubClusters == null || activeSubClusters.size() < 1) {

@goiri goiri Jul 24, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isEmpty()?

@slfan1989 slfan1989 Jul 25, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Your suggestion is right, I will modify the code.

RouterServerUtil.logAndThrowException(
FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE, null);
}
List<SubClusterId> list = new ArrayList<>(activeSubclusters.keySet());

List<SubClusterId> list = new ArrayList<>(activeSubClusters.keySet());
return list.get(rand.nextInt(list.size()));
}

Expand All @@ -280,47 +274,50 @@ private SubClusterId getRandomActiveSubCluster(
public GetNewApplicationResponse getNewApplication(
GetNewApplicationRequest request) throws YarnException, IOException {

long startTime = clock.getTime();
if(request == null) {

@goiri goiri Jul 24, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

if (request == null) {

@slfan1989 slfan1989 Jul 25, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will fix it.

routerMetrics.incrAppsFailedCreated();
String errMsg = "Missing getNewApplication request.";
RouterAuditLogger.logFailure(user.getShortUserName(),
RouterAuditLogger.AuditConstants.GET_NEW_APP, "UNKNOWN",
"RouterClientRMService", errMsg);
RouterServerUtil.logAndThrowException(errMsg, null);
}

long startTime = clock.getTime();
Map<SubClusterId, SubClusterInfo> subClustersActive =
federationFacade.getSubClusters(true);

GetNewApplicationResponse response = null;

@goiri goiri Jul 25, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not leave the initialization where it was?


for (int i = 0; i < numSubmitRetries; ++i) {
SubClusterId subClusterId = getRandomActiveSubCluster(subClustersActive);
LOG.debug(
"getNewApplication try #{} on SubCluster {}", i, subClusterId);
ApplicationClientProtocol clientRMProxy =
getClientRMProxyForSubCluster(subClusterId);
GetNewApplicationResponse response = null;
LOG.info("getNewApplication try #{} on SubCluster {}", i, subClusterId);
ApplicationClientProtocol clientRMProxy = getClientRMProxyForSubCluster(subClusterId);
response = null;
try {
response = clientRMProxy.getNewApplication(request);
} catch (Exception e) {
LOG.warn("Unable to create a new ApplicationId in SubCluster "
+ subClusterId.getId(), e);
LOG.warn("Unable to create a new ApplicationId in SubCluster {}.", subClusterId.getId(), e);
subClustersActive.remove(subClusterId);
}

if (response != null) {

long stopTime = clock.getTime();
routerMetrics.succeededAppsCreated(stopTime - startTime);
RouterAuditLogger.logSuccess(user.getShortUserName(),
RouterAuditLogger.AuditConstants.GET_NEW_APP,
"RouterClientRMService", response.getApplicationId());
return response;
} else {
// Empty response from the ResourceManager.
// Blacklist this subcluster for this request.
subClustersActive.remove(subClusterId);
Comment thread
goiri marked this conversation as resolved.
}

}

routerMetrics.incrAppsFailedCreated();
String errMsg = "Fail to create a new application.";
String errMsg = "Failed to create a new application.";
RouterAuditLogger.logFailure(user.getShortUserName(),
RouterAuditLogger.AuditConstants.GET_NEW_APP, "UNKNOWN",
"RouterClientRMService", errMsg);
throw new YarnException(errMsg);
RouterServerUtil.logAndThrowException(errMsg, null);
return response;

@goiri goiri Jul 25, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

return? Aren't we throwing the exception?

@slfan1989 slfan1989 Jul 26, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Your suggestion is right, I will throw exception directly，thank you.

}

/**
Expand Down Expand Up @@ -392,32 +389,31 @@ public GetNewApplicationResponse getNewApplication(
public SubmitApplicationResponse submitApplication(
SubmitApplicationRequest request) throws YarnException, IOException {

long startTime = clock.getTime();

if (request == null || request.getApplicationSubmissionContext() == null
|| request.getApplicationSubmissionContext()
.getApplicationId() == null) {
|| request.getApplicationSubmissionContext().getApplicationId() == null) {
routerMetrics.incrAppsFailedSubmitted();
String errMsg =
"Missing submitApplication request or applicationSubmissionContext "
+ "information.";
"Missing submitApplication request or applicationSubmissionContext information.";
RouterAuditLogger.logFailure(user.getShortUserName(),
RouterAuditLogger.AuditConstants.SUBMIT_NEW_APP, "UNKNOWN",
"RouterClientRMService", errMsg);
throw new YarnException(errMsg);
RouterServerUtil.logAndThrowException(errMsg, null);
}

long startTime = clock.getTime();

ApplicationId applicationId =
request.getApplicationSubmissionContext().getApplicationId();

List<SubClusterId> blacklist = new ArrayList<SubClusterId>();
List<SubClusterId> blacklist = new ArrayList<>();

for (int i = 0; i < numSubmitRetries; ++i) {

SubClusterId subClusterId = policyFacade.getHomeSubcluster(
request.getApplicationSubmissionContext(), blacklist);
LOG.info("submitApplication appId {} try #{} on SubCluster {}.", applicationId, i,
subClusterId);

LOG.info("submitApplication appId {} try #{} on SubCluster {}.",
applicationId, i, subClusterId);

ApplicationHomeSubCluster appHomeSubCluster =
ApplicationHomeSubCluster.newInstance(applicationId, subClusterId);
Expand All @@ -430,32 +426,34 @@ public SubmitApplicationResponse submitApplication(
federationFacade.addApplicationHomeSubCluster(appHomeSubCluster);
} catch (YarnException e) {
routerMetrics.incrAppsFailedSubmitted();
String message = "Unable to insert the ApplicationId " + applicationId
+ " into the FederationStateStore";
String message =
String.format("Unable to insert the ApplicationId %s into the FederationStateStore.",
applicationId);
RouterAuditLogger.logFailure(user.getShortUserName(),
RouterAuditLogger.AuditConstants.SUBMIT_NEW_APP, "UNKNOWN",
"RouterClientRMService", message, applicationId, subClusterId);
throw new YarnException(message, e);
RouterServerUtil.logAndThrowException(message, e);
}
} else {
try {
// update the mapping of applicationId and the home subClusterId to
// the new subClusterId we have selected
federationFacade.updateApplicationHomeSubCluster(appHomeSubCluster);
} catch (YarnException e) {
String message = "Unable to update the ApplicationId " + applicationId
+ " into the FederationStateStore";
String message =
String.format("Unable to update the ApplicationId %s into the FederationStateStore.",
applicationId);
SubClusterId subClusterIdInStateStore =
federationFacade.getApplicationHomeSubCluster(applicationId);
if (subClusterId == subClusterIdInStateStore) {
LOG.info("Application {} already submitted on SubCluster {}.", applicationId,
subClusterId);
LOG.info("Application {} already submitted on SubCluster {}.",
applicationId, subClusterId);
} else {
routerMetrics.incrAppsFailedSubmitted();
RouterAuditLogger.logFailure(user.getShortUserName(),
RouterAuditLogger.AuditConstants.SUBMIT_NEW_APP, "UNKNOWN",
"RouterClientRMService", message, applicationId, subClusterId);
throw new YarnException(message, e);
RouterServerUtil.logAndThrowException(message, e);
}
}
}
Expand Down Expand Up @@ -489,9 +487,8 @@ public SubmitApplicationResponse submitApplication(
}

routerMetrics.incrAppsFailedSubmitted();
String errMsg = "Application "
+ request.getApplicationSubmissionContext().getApplicationName()
+ " with appId " + applicationId + " failed to be submitted.";
String errMsg = String.format("Application %s with appId %s failed to be submitted.",
request.getApplicationSubmissionContext().getApplicationName(), applicationId);
RouterAuditLogger.logFailure(user.getShortUserName(),
RouterAuditLogger.AuditConstants.SUBMIT_NEW_APP, "UNKNOWN",
"RouterClientRMService", errMsg, applicationId);
Expand Down
5 changes: 2 additions & 3 deletions ...t/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ public void setUp() {
interceptor.setConf(this.getConf());
interceptor.init(user);

subClusters = new ArrayList<SubClusterId>();
subClusters = new ArrayList<>();

try {
for (int i = 0; i < NUM_SUBCLUSTER; i++) {
Expand Down Expand Up @@ -201,8 +201,7 @@ protected YarnConfiguration createConfiguration() {
* ApplicationId has to belong to one of the SubCluster in the cluster.
*/
@Test
public void testGetNewApplication()
throws YarnException, IOException, InterruptedException {
public void testGetNewApplication() throws YarnException, IOException {
LOG.info("Test FederationClientInterceptor: Get New Application");

GetNewApplicationRequest request = GetNewApplicationRequest.newInstance();
Expand Down