ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public SubmitApplicationResponse submitApplication( SubmitApplicationRequest request) throws YarnException, IOException { ApplicationSubmissionContext submissionContext = request .getApplicationSubmissionContext(); ApplicationId applicationId = submissionContext.getApplicationId(); CallerContext callerContext = CallerContext.getCurrent(); String user = null; try { user = UserGroupInformation.getCurrentUser().getShortUserName(); } catch (IOException ie) { LOG.warn("Unable to get the current user.", ie); RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST, ie.getMessage(), "ClientRMService", "Exception in submitting application", applicationId, callerContext); throw RPCUtil.getRemoteException(ie); } if (YarnConfiguration.timelineServiceV2Enabled(getConfig())) { String value = null; try { for (String tag : submissionContext.getApplicationTags()) { if (tag.startsWith(TimelineUtils.FLOW_RUN_ID_TAG_PREFIX + ":") || tag.startsWith( TimelineUtils.FLOW_RUN_ID_TAG_PREFIX.toLowerCase() + ":")) { value = tag.substring(TimelineUtils.FLOW_RUN_ID_TAG_PREFIX.length() + 1); Long.valueOf(value); } } } catch (NumberFormatException e) { LOG.warn("Invalid to flow run: " + value + ". Flow run should be a long integer", e); RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST, e.getMessage(), "ClientRMService", "Exception in submitting application", applicationId); throw RPCUtil.getRemoteException(e); } } if (rmContext.getRMApps().get(applicationId) != null) { LOG.info("This is an earlier submitted application: " + applicationId); return SubmitApplicationResponse.newInstance(); } ByteBuffer tokenConf = submissionContext.getAMContainerSpec().getTokensConf(); if (tokenConf != null) { int maxSize = getConfig() .getInt(YarnConfiguration.RM_DELEGATION_TOKEN_MAX_CONF_SIZE, YarnConfiguration.DEFAULT_RM_DELEGATION_TOKEN_MAX_CONF_SIZE_BYTES); LOG.info("Using app provided configurations for delegation token renewal," + " total size = " + tokenConf.capacity()); if (tokenConf.capacity() > maxSize) { throw new YarnException( "Exceed " + YarnConfiguration.RM_DELEGATION_TOKEN_MAX_CONF_SIZE + " = " + maxSize + " bytes, current conf size = " + tokenConf.capacity() + " bytes."); } } if (submissionContext.getQueue() == null) { submissionContext.setQueue(YarnConfiguration.DEFAULT_QUEUE_NAME); } if (submissionContext.getApplicationName() == null) { submissionContext.setApplicationName( YarnConfiguration.DEFAULT_APPLICATION_NAME); } if (submissionContext.getApplicationType() == null) { submissionContext .setApplicationType(YarnConfiguration.DEFAULT_APPLICATION_TYPE); } else { if (submissionContext.getApplicationType().length() > YarnConfiguration.APPLICATION_TYPE_LENGTH) { submissionContext.setApplicationType(submissionContext .getApplicationType().substring(0, YarnConfiguration.APPLICATION_TYPE_LENGTH)); } } ReservationId reservationId = request.getApplicationSubmissionContext() .getReservationID(); checkReservationACLs(submissionContext.getQueue(), AuditConstants .SUBMIT_RESERVATION_REQUEST, reservationId); try { rmAppManager.submitApplication(submissionContext, System.currentTimeMillis(), user); LOG.info("Application with id " + applicationId.getId() + " submitted by user " + user); RMAuditLogger.logSuccess(user, AuditConstants.SUBMIT_APP_REQUEST, "ClientRMService", applicationId, callerContext); } catch (YarnException e) { LOG.info("Exception in submitting " + applicationId, e); RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST, e.getMessage(), "ClientRMService", "Exception in submitting application", applicationId, callerContext); throw e; } return recordFactory .newRecordInstance(SubmitApplicationResponse.class); } ClientRMService(RMContext rmContext, YarnScheduler scheduler,
      RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
      QueueACLsManager queueACLsManager,
      RMDelegationTokenSecretManager rmDTSecretManager);  ClientRMService(RMContext rmContext, YarnScheduler scheduler,
      RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
      QueueACLsManager queueACLsManager,
      RMDelegationTokenSecretManager rmDTSecretManager, Clock clock); @Private InetSocketAddress getBindAddress(); @Override GetNewApplicationResponse getNewApplication(
      GetNewApplicationRequest request); @Override GetApplicationReportResponse getApplicationReport(
      GetApplicationReportRequest request); @Override GetApplicationAttemptReportResponse getApplicationAttemptReport(
      GetApplicationAttemptReportRequest request); @Override GetApplicationAttemptsResponse getApplicationAttempts(
      GetApplicationAttemptsRequest request); @Override GetContainerReportResponse getContainerReport(
      GetContainerReportRequest request); @Override GetContainersResponse getContainers(GetContainersRequest request); @Override SubmitApplicationResponse submitApplication(
      SubmitApplicationRequest request); @SuppressWarnings("unchecked") @Override FailApplicationAttemptResponse failApplicationAttempt(
      FailApplicationAttemptRequest request); @SuppressWarnings("unchecked") @Override KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetApplicationsResponse getApplications(GetApplicationsRequest request); @Override GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request); @Override GetQueueInfoResponse getQueueInfo(GetQueueInfoRequest request); @Override GetQueueUserAclsInfoResponse getQueueUserAcls(
      GetQueueUserAclsInfoRequest request); @Override GetDelegationTokenResponse getDelegationToken(
      GetDelegationTokenRequest request); @Override RenewDelegationTokenResponse renewDelegationToken(
      RenewDelegationTokenRequest request); @Override CancelDelegationTokenResponse cancelDelegationToken(
      CancelDelegationTokenRequest request); @SuppressWarnings("unchecked") @Override MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues(
      MoveApplicationAcrossQueuesRequest request); @VisibleForTesting Server getServer(); @Override GetNewReservationResponse getNewReservation(
      GetNewReservationRequest request); @Override ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override ReservationListResponse listReservations(
        ReservationListRequest requestInfo); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetLabelsToNodesResponse getLabelsToNodes(
      GetLabelsToNodesRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request); @Override UpdateApplicationPriorityResponse updateApplicationPriority(
      UpdateApplicationPriorityRequest request); @SuppressWarnings("unchecked") @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 GetAttributesToNodesResponse getAttributesToNodes(
      GetAttributesToNodesRequest request); @Override GetClusterNodeAttributesResponse getClusterNodeAttributes(
      GetClusterNodeAttributesRequest request); @Override GetNodesToAttributesResponse getNodesToAttributes(
      GetNodesToAttributesRequest request); @VisibleForTesting void setDisplayPerUserApps(boolean displayPerUserApps);  }
@Test(timeout=4000) public void testConcurrentAppSubmit() throws IOException, InterruptedException, BrokenBarrierException, YarnException { ResourceScheduler scheduler = mockResourceScheduler(); RMContext rmContext = mock(RMContext.class); mockRMContext(scheduler, rmContext); RMStateStore stateStore = mock(RMStateStore.class); when(rmContext.getStateStore()).thenReturn(stateStore); RMAppManager appManager = new RMAppManager(rmContext, scheduler, null, mock(ApplicationACLsManager.class), new Configuration()); final ApplicationId appId1 = getApplicationId(100); final ApplicationId appId2 = getApplicationId(101); final SubmitApplicationRequest submitRequest1 = mockSubmitAppRequest( appId1, null, null); final SubmitApplicationRequest submitRequest2 = mockSubmitAppRequest( appId2, null, null); final CyclicBarrier startBarrier = new CyclicBarrier(2); final CyclicBarrier endBarrier = new CyclicBarrier(2); EventHandler<Event> eventHandler = new EventHandler<Event>() { @Override public void handle(Event rawEvent) { if (rawEvent instanceof RMAppEvent) { RMAppEvent event = (RMAppEvent) rawEvent; if (event.getApplicationId().equals(appId1)) { try { startBarrier.await(); endBarrier.await(); } catch (BrokenBarrierException e) { LOG.warn("Broken Barrier", e); } catch (InterruptedException e) { LOG.warn("Interrupted while awaiting barriers", e); } } } } }; when(rmContext.getDispatcher().getEventHandler()).thenReturn(eventHandler); doReturn(mock(RMTimelineCollectorManager.class)).when(rmContext) .getRMTimelineCollectorManager(); final ClientRMService rmService = new ClientRMService(rmContext, scheduler, appManager, null, null, null); rmService.init(new Configuration()); Thread t = new Thread() { @Override public void run() { try { rmService.submitApplication(submitRequest1); } catch (YarnException | IOException e) {} } }; t.start(); startBarrier.await(); rmService.submitApplication(submitRequest2); endBarrier.await(); t.join(); }