FederationClientInterceptor extends AbstractClientRequestInterceptor { @Override public SubmitApplicationResponse submitApplication( SubmitApplicationRequest request) throws YarnException, IOException { long startTime = clock.getTime(); if (request == null || request.getApplicationSubmissionContext() == null || request.getApplicationSubmissionContext() .getApplicationId() == null) { routerMetrics.incrAppsFailedSubmitted(); RouterServerUtil .logAndThrowException("Missing submitApplication request or " + "applicationSubmissionContex information.", null); } ApplicationId applicationId = request.getApplicationSubmissionContext().getApplicationId(); List<SubClusterId> blacklist = new ArrayList<SubClusterId>(); for (int i = 0; i < numSubmitRetries; ++i) { SubClusterId subClusterId = policyFacade.getHomeSubcluster( request.getApplicationSubmissionContext(), blacklist); LOG.info("submitApplication appId" + applicationId + " try #" + i + " on SubCluster " + subClusterId); ApplicationHomeSubCluster appHomeSubCluster = ApplicationHomeSubCluster.newInstance(applicationId, subClusterId); if (i == 0) { try { subClusterId = federationFacade.addApplicationHomeSubCluster(appHomeSubCluster); } catch (YarnException e) { routerMetrics.incrAppsFailedSubmitted(); String message = "Unable to insert the ApplicationId " + applicationId + " into the FederationStateStore"; RouterServerUtil.logAndThrowException(message, e); } } else { try { federationFacade.updateApplicationHomeSubCluster(appHomeSubCluster); } catch (YarnException e) { String message = "Unable to update the ApplicationId " + applicationId + " into the FederationStateStore"; SubClusterId subClusterIdInStateStore = federationFacade.getApplicationHomeSubCluster(applicationId); if (subClusterId == subClusterIdInStateStore) { LOG.info("Application " + applicationId + " already submitted on SubCluster " + subClusterId); } else { routerMetrics.incrAppsFailedSubmitted(); RouterServerUtil.logAndThrowException(message, e); } } } ApplicationClientProtocol clientRMProxy = getClientRMProxyForSubCluster(subClusterId); SubmitApplicationResponse response = null; try { response = clientRMProxy.submitApplication(request); } catch (Exception e) { LOG.warn("Unable to submit the application " + applicationId + "to SubCluster " + subClusterId.getId(), e); } if (response != null) { LOG.info("Application " + request.getApplicationSubmissionContext().getApplicationName() + " with appId " + applicationId + " submitted on " + subClusterId); long stopTime = clock.getTime(); routerMetrics.succeededAppsSubmitted(stopTime - startTime); return response; } else { blacklist.add(subClusterId); } } routerMetrics.incrAppsFailedSubmitted(); String errMsg = "Application " + request.getApplicationSubmissionContext().getApplicationName() + " with appId " + applicationId + " failed to be submitted."; LOG.error(errMsg); throw new YarnException(errMsg); }  @Override void init(String userName); @Override void setNextInterceptor(ClientRequestInterceptor next); @Override GetNewApplicationResponse getNewApplication(
      GetNewApplicationRequest request); @Override SubmitApplicationResponse submitApplication(
      SubmitApplicationRequest request); @Override KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetApplicationReportResponse getApplicationReport(
      GetApplicationReportRequest request); @Override GetApplicationsResponse getApplications(GetApplicationsRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request); @Override GetQueueInfoResponse getQueueInfo(GetQueueInfoRequest request); @Override GetQueueUserAclsInfoResponse getQueueUserAcls(
      GetQueueUserAclsInfoRequest request); @Override MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues(
      MoveApplicationAcrossQueuesRequest request); @Override GetNewReservationResponse getNewReservation(
      GetNewReservationRequest request); @Override ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationListResponse listReservations(
      ReservationListRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetLabelsToNodesResponse getLabelsToNodes(
      GetLabelsToNodesRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request); @Override GetApplicationAttemptReportResponse getApplicationAttemptReport(
      GetApplicationAttemptReportRequest request); @Override GetApplicationAttemptsResponse getApplicationAttempts(
      GetApplicationAttemptsRequest request); @Override GetContainerReportResponse getContainerReport(
      GetContainerReportRequest request); @Override GetContainersResponse getContainers(GetContainersRequest request); @Override GetDelegationTokenResponse getDelegationToken(
      GetDelegationTokenRequest request); @Override RenewDelegationTokenResponse renewDelegationToken(
      RenewDelegationTokenRequest request); @Override CancelDelegationTokenResponse cancelDelegationToken(
      CancelDelegationTokenRequest request); @Override FailApplicationAttemptResponse failApplicationAttempt(
      FailApplicationAttemptRequest request); @Override UpdateApplicationPriorityResponse updateApplicationPriority(
      UpdateApplicationPriorityRequest request); @Override SignalContainerResponse signalToContainer(
      SignalContainerRequest request); @Override UpdateApplicationTimeoutsResponse updateApplicationTimeouts(
      UpdateApplicationTimeoutsRequest request); @Override GetAllResourceProfilesResponse getResourceProfiles(
      GetAllResourceProfilesRequest request); @Override GetResourceProfileResponse getResourceProfile(
      GetResourceProfileRequest request); @Override GetAllResourceTypeInfoResponse getResourceTypeInfo(
      GetAllResourceTypeInfoRequest request); @Override void shutdown(); @Override GetAttributesToNodesResponse getAttributesToNodes(
      GetAttributesToNodesRequest request); @Override GetClusterNodeAttributesResponse getClusterNodeAttributes(
      GetClusterNodeAttributesRequest request); @Override GetNodesToAttributesResponse getNodesToAttributes(
      GetNodesToAttributesRequest request);  }
@Test public void testSubmitApplication() throws YarnException, IOException, InterruptedException { LOG.info("Test FederationClientInterceptor: Submit Application"); ApplicationId appId = ApplicationId.newInstance(System.currentTimeMillis(), 1); ApplicationSubmissionContext context = ApplicationSubmissionContext .newInstance(appId, "", "", null, null, false, false, -1, null, null); SubmitApplicationRequest request = SubmitApplicationRequest.newInstance(context); SubmitApplicationResponse response = interceptor.submitApplication(request); Assert.assertNotNull(response); SubClusterId scIdResult = stateStoreUtil.queryApplicationHomeSC(appId); Assert.assertNotNull(scIdResult); Assert.assertTrue(subClusters.contains(scIdResult)); }
@Test public void testSubmitApplicationMultipleSubmission() throws YarnException, IOException, InterruptedException { LOG.info( "Test FederationClientInterceptor: Submit Application - Multiple"); ApplicationId appId = ApplicationId.newInstance(System.currentTimeMillis(), 1); ApplicationSubmissionContext context = ApplicationSubmissionContext .newInstance(appId, "", "", null, null, false, false, -1, null, null); SubmitApplicationRequest request = SubmitApplicationRequest.newInstance(context); SubmitApplicationResponse response = interceptor.submitApplication(request); Assert.assertNotNull(response); SubClusterId scIdResult = stateStoreUtil.queryApplicationHomeSC(appId); Assert.assertNotNull(scIdResult); response = interceptor.submitApplication(request); Assert.assertNotNull(response); SubClusterId scIdResult2 = stateStoreUtil.queryApplicationHomeSC(appId); Assert.assertNotNull(scIdResult2); Assert.assertEquals(scIdResult, scIdResult); }