FederationInterceptorREST extends AbstractRESTRequestInterceptor { @Override public Response submitApplication(ApplicationSubmissionContextInfo newApp, HttpServletRequest hsr) throws AuthorizationException, IOException, InterruptedException { long startTime = clock.getTime(); if (newApp == null || newApp.getApplicationId() == null) { routerMetrics.incrAppsFailedSubmitted(); String errMsg = "Missing ApplicationSubmissionContextInfo or " + "applicationSubmissionContex information."; return Response .status(Status.BAD_REQUEST) .entity(errMsg) .build(); } ApplicationId applicationId = null; try { applicationId = ApplicationId.fromString(newApp.getApplicationId()); } catch (IllegalArgumentException e) { routerMetrics.incrAppsFailedSubmitted(); return Response .status(Status.BAD_REQUEST) .entity(e.getLocalizedMessage()) .build(); } List<SubClusterId> blacklist = new ArrayList<SubClusterId>(); for (int i = 0; i < numSubmitRetries; ++i) { ApplicationSubmissionContext context = RMWebAppUtil.createAppSubmissionContext(newApp, this.getConf()); SubClusterId subClusterId = null; try { subClusterId = policyFacade.getHomeSubcluster(context, blacklist); } catch (YarnException e) { routerMetrics.incrAppsFailedSubmitted(); return Response .status(Status.SERVICE_UNAVAILABLE) .entity(e.getLocalizedMessage()) .build(); } LOG.info("submitApplication appId {} try #{} on SubCluster {}", applicationId, i, subClusterId); ApplicationHomeSubCluster appHomeSubCluster = ApplicationHomeSubCluster.newInstance(applicationId, subClusterId); if (i == 0) { try { subClusterId = federationFacade.addApplicationHomeSubCluster(appHomeSubCluster); } catch (YarnException e) { routerMetrics.incrAppsFailedSubmitted(); String errMsg = "Unable to insert the ApplicationId " + applicationId + " into the FederationStateStore"; return Response .status(Status.SERVICE_UNAVAILABLE) .entity(errMsg + " " + e.getLocalizedMessage()) .build(); } } else { try { federationFacade.updateApplicationHomeSubCluster(appHomeSubCluster); } catch (YarnException e) { String errMsg = "Unable to update the ApplicationId " + applicationId + " into the FederationStateStore"; SubClusterId subClusterIdInStateStore; try { subClusterIdInStateStore = federationFacade.getApplicationHomeSubCluster(applicationId); } catch (YarnException e1) { routerMetrics.incrAppsFailedSubmitted(); return Response .status(Status.SERVICE_UNAVAILABLE) .entity(e1.getLocalizedMessage()) .build(); } if (subClusterId == subClusterIdInStateStore) { LOG.info("Application {} already submitted on SubCluster {}", applicationId, subClusterId); } else { routerMetrics.incrAppsFailedSubmitted(); return Response .status(Status.SERVICE_UNAVAILABLE) .entity(errMsg) .build(); } } } SubClusterInfo subClusterInfo; try { subClusterInfo = federationFacade.getSubCluster(subClusterId); } catch (YarnException e) { routerMetrics.incrAppsFailedSubmitted(); return Response .status(Status.SERVICE_UNAVAILABLE) .entity(e.getLocalizedMessage()) .build(); } Response response = null; try { response = getOrCreateInterceptorForSubCluster(subClusterId, subClusterInfo.getRMWebServiceAddress()).submitApplication(newApp, hsr); } catch (Exception e) { LOG.warn("Unable to submit the application {} to SubCluster {}", applicationId, subClusterId.getId(), e); } if (response != null && response.getStatus() == HttpServletResponse.SC_ACCEPTED) { LOG.info("Application {} with appId {} submitted on {}", context.getApplicationName(), applicationId, subClusterId); long stopTime = clock.getTime(); routerMetrics.succeededAppsSubmitted(stopTime - startTime); return response; } else { blacklist.add(subClusterId); } } routerMetrics.incrAppsFailedSubmitted(); String errMsg = "Application " + newApp.getApplicationName() + " with appId " + applicationId + " failed to be submitted."; LOG.error(errMsg); return Response .status(Status.SERVICE_UNAVAILABLE) .entity(errMsg) .build(); }  @Override void init(String user); @Override Response createNewApplication(HttpServletRequest hsr); @Override Response submitApplication(ApplicationSubmissionContextInfo newApp,
      HttpServletRequest hsr); @Override AppInfo getApp(HttpServletRequest hsr, String appId,
      Set<String> unselectedFields); @Override Response updateAppState(AppState targetState, HttpServletRequest hsr,
      String appId); @Override AppsInfo getApps(HttpServletRequest hsr, String stateQuery,
      Set<String> statesQuery, String finalStatusQuery, String userQuery,
      String queueQuery, String count, String startedBegin, String startedEnd,
      String finishBegin, String finishEnd, Set<String> applicationTypes,
      Set<String> applicationTags, Set<String> unselectedFields); @Override NodeInfo getNode(String nodeId); @Override NodesInfo getNodes(String states); @Override ClusterMetricsInfo getClusterMetricsInfo(); @Override AppState getAppState(HttpServletRequest hsr, String appId); @Override ClusterInfo get(); @Override ClusterInfo getClusterInfo(); @Override ClusterUserInfo getClusterUserInfo(HttpServletRequest hsr); @Override SchedulerTypeInfo getSchedulerInfo(); @Override String dumpSchedulerLogs(String time, HttpServletRequest hsr); @Override ActivitiesInfo getActivities(HttpServletRequest hsr, String nodeId); @Override AppActivitiesInfo getAppActivities(HttpServletRequest hsr,
      String appId, String time); @Override ApplicationStatisticsInfo getAppStatistics(HttpServletRequest hsr,
      Set<String> stateQueries, Set<String> typeQueries); @Override NodeToLabelsInfo getNodeToLabels(HttpServletRequest hsr); @Override LabelsToNodesInfo getLabelsToNodes(Set<String> labels); @Override Response replaceLabelsOnNodes(NodeToLabelsEntryList newNodeToLabels,
      HttpServletRequest hsr); @Override Response replaceLabelsOnNode(Set<String> newNodeLabelsName,
      HttpServletRequest hsr, String nodeId); @Override NodeLabelsInfo getClusterNodeLabels(HttpServletRequest hsr); @Override Response addToClusterNodeLabels(NodeLabelsInfo newNodeLabels,
      HttpServletRequest hsr); @Override Response removeFromCluserNodeLabels(Set<String> oldNodeLabels,
      HttpServletRequest hsr); @Override NodeLabelsInfo getLabelsOnNode(HttpServletRequest hsr, String nodeId); @Override AppPriority getAppPriority(HttpServletRequest hsr, String appId); @Override Response updateApplicationPriority(AppPriority targetPriority,
      HttpServletRequest hsr, String appId); @Override AppQueue getAppQueue(HttpServletRequest hsr, String appId); @Override Response updateAppQueue(AppQueue targetQueue, HttpServletRequest hsr,
      String appId); @Override Response postDelegationToken(DelegationToken tokenData,
      HttpServletRequest hsr); @Override Response postDelegationTokenExpiration(HttpServletRequest hsr); @Override Response cancelDelegationToken(HttpServletRequest hsr); @Override Response createNewReservation(HttpServletRequest hsr); @Override Response submitReservation(ReservationSubmissionRequestInfo resContext,
      HttpServletRequest hsr); @Override Response updateReservation(ReservationUpdateRequestInfo resContext,
      HttpServletRequest hsr); @Override Response deleteReservation(ReservationDeleteRequestInfo resContext,
      HttpServletRequest hsr); @Override Response listReservation(String queue, String reservationId,
      long startTime, long endTime, boolean includeResourceAllocations,
      HttpServletRequest hsr); @Override AppTimeoutInfo getAppTimeout(HttpServletRequest hsr, String appId,
      String type); @Override AppTimeoutsInfo getAppTimeouts(HttpServletRequest hsr, String appId); @Override Response updateApplicationTimeout(AppTimeoutInfo appTimeout,
      HttpServletRequest hsr, String appId); @Override AppAttemptsInfo getAppAttempts(HttpServletRequest hsr, String appId); @Override RMQueueAclInfo checkUserAccessToQueue(String queue, String username,
      String queueAclType, HttpServletRequest hsr); @Override AppAttemptInfo getAppAttempt(HttpServletRequest req,
      HttpServletResponse res, String appId, String appAttemptId); @Override ContainersInfo getContainers(HttpServletRequest req,
      HttpServletResponse res, String appId, String appAttemptId); @Override ContainerInfo getContainer(HttpServletRequest req,
      HttpServletResponse res, String appId, String appAttemptId,
      String containerId); @Override void setNextInterceptor(RESTRequestInterceptor next); @Override void shutdown();  }
@Test public void testSubmitApplication() throws YarnException, IOException, InterruptedException { ApplicationId appId = ApplicationId.newInstance(System.currentTimeMillis(), 1); ApplicationSubmissionContextInfo context = new ApplicationSubmissionContextInfo(); context.setApplicationId(appId.toString()); Response response = interceptor.submitApplication(context, null); Assert.assertEquals(ACCEPTED, response.getStatus()); SubClusterId ci = (SubClusterId) response.getEntity(); Assert.assertNotNull(response); SubClusterId scIdResult = stateStoreUtil.queryApplicationHomeSC(appId); Assert.assertNotNull(scIdResult); Assert.assertTrue(subClusters.contains(scIdResult)); Assert.assertEquals(ci, scIdResult); }
@Test public void testSubmitApplicationMultipleSubmission() throws YarnException, IOException, InterruptedException { ApplicationId appId = ApplicationId.newInstance(System.currentTimeMillis(), 1); ApplicationSubmissionContextInfo context = new ApplicationSubmissionContextInfo(); context.setApplicationId(appId.toString()); Response response = interceptor.submitApplication(context, null); Assert.assertNotNull(response); Assert.assertEquals(ACCEPTED, response.getStatus()); SubClusterId scIdResult = stateStoreUtil.queryApplicationHomeSC(appId); Assert.assertNotNull(scIdResult); response = interceptor.submitApplication(context, null); Assert.assertNotNull(response); Assert.assertEquals(ACCEPTED, response.getStatus()); SubClusterId scIdResult2 = stateStoreUtil.queryApplicationHomeSC(appId); Assert.assertNotNull(scIdResult2); Assert.assertEquals(scIdResult, scIdResult2); }
@Test public void testSubmitApplicationEmptyRequest() throws YarnException, IOException, InterruptedException { Response response = interceptor.submitApplication(null, null); Assert.assertEquals(BAD_REQUEST, response.getStatus()); response = interceptor .submitApplication(new ApplicationSubmissionContextInfo(), null); Assert.assertEquals(BAD_REQUEST, response.getStatus()); ApplicationSubmissionContextInfo context = new ApplicationSubmissionContextInfo(); response = interceptor.submitApplication(context, null); Assert.assertEquals(BAD_REQUEST, response.getStatus()); }
@Test public void testSubmitApplicationWrongFormat() throws YarnException, IOException, InterruptedException { ApplicationSubmissionContextInfo context = new ApplicationSubmissionContextInfo(); context.setApplicationId("Application_wrong_id"); Response response = interceptor.submitApplication(context, null); Assert.assertEquals(BAD_REQUEST, response.getStatus()); }