ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @SuppressWarnings("unchecked") @Override public KillApplicationResponse forceKillApplication( KillApplicationRequest request) throws YarnException { ApplicationId applicationId = request.getApplicationId(); UserGroupInformation callerUGI; try { callerUGI = UserGroupInformation.getCurrentUser(); } catch (IOException ie) { LOG.info("Error getting UGI ", ie); RMAuditLogger.logFailure("UNKNOWN", AuditConstants.KILL_APP_REQUEST, "UNKNOWN", "ClientRMService" , "Error getting UGI", applicationId); throw RPCUtil.getRemoteException(ie); } RMApp application = this.rmContext.getRMApps().get(applicationId); if (application == null) { RMAuditLogger.logFailure(callerUGI.getUserName(), AuditConstants.KILL_APP_REQUEST, "UNKNOWN", "ClientRMService", "Trying to kill an absent application", applicationId); throw new ApplicationNotFoundException("Trying to kill an absent" + " application " + applicationId); } if (!checkAccess(callerUGI, application.getUser(), ApplicationAccessType.MODIFY_APP, application)) { RMAuditLogger.logFailure(callerUGI.getShortUserName(), AuditConstants.KILL_APP_REQUEST, "User doesn't have permissions to " + ApplicationAccessType.MODIFY_APP.toString(), "ClientRMService", AuditConstants.UNAUTHORIZED_USER, applicationId); throw RPCUtil.getRemoteException(new AccessControlException("User " + callerUGI.getShortUserName() + " cannot perform operation " + ApplicationAccessType.MODIFY_APP.name() + " on " + applicationId)); } if (application.isAppFinalStateStored()) { RMAuditLogger.logSuccess(callerUGI.getShortUserName(), AuditConstants.KILL_APP_REQUEST, "ClientRMService", applicationId); return KillApplicationResponse.newInstance(true); } this.rmContext.getDispatcher().getEventHandler() .handle(new RMAppEvent(applicationId, RMAppEventType.KILL)); return KillApplicationResponse.newInstance( application.getApplicationSubmissionContext().getUnmanagedAM()); } 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 KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetApplicationsResponse getApplications(
      GetApplicationsRequest request); @Private GetApplicationsResponse getApplications(
      GetApplicationsRequest request, boolean caseSensitive); @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 ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request);  }
@Test public void testForceKillNonExistingApplication() throws YarnException { RMContext rmContext = mock(RMContext.class); when(rmContext.getRMApps()).thenReturn( new ConcurrentHashMap<ApplicationId, RMApp>()); ClientRMService rmService = new ClientRMService(rmContext, null, null, null, null, null); ApplicationId applicationId = BuilderUtils.newApplicationId(System.currentTimeMillis(), 0); KillApplicationRequest request = KillApplicationRequest.newInstance(applicationId); try { rmService.forceKillApplication(request); Assert.fail(); } catch (ApplicationNotFoundException ex) { Assert.assertEquals(ex.getMessage(), "Trying to kill an absent " + "application " + request.getApplicationId()); } }
@Test public void testForceKillApplication() throws Exception { YarnConfiguration conf = new YarnConfiguration(); MockRM rm = new MockRM(); rm.init(conf); rm.start(); ClientRMService rmService = rm.getClientRMService(); GetApplicationsRequest getRequest = GetApplicationsRequest.newInstance( EnumSet.of(YarnApplicationState.KILLED)); RMApp app1 = rm.submitApp(1024); RMApp app2 = rm.submitApp(1024, true); assertEquals("Incorrect number of apps in the RM", 0, rmService.getApplications(getRequest).getApplicationList().size()); KillApplicationRequest killRequest1 = KillApplicationRequest.newInstance(app1.getApplicationId()); KillApplicationRequest killRequest2 = KillApplicationRequest.newInstance(app2.getApplicationId()); int killAttemptCount = 0; for (int i = 0; i < 100; i++) { KillApplicationResponse killResponse1 = rmService.forceKillApplication(killRequest1); killAttemptCount++; if (killResponse1.getIsKillCompleted()) { break; } Thread.sleep(10); } assertTrue("Kill attempt count should be greater than 1 for managed AMs", killAttemptCount > 1); assertEquals("Incorrect number of apps in the RM", 1, rmService.getApplications(getRequest).getApplicationList().size()); KillApplicationResponse killResponse2 = rmService.forceKillApplication(killRequest2); assertTrue("Killing UnmanagedAM should falsely acknowledge true", killResponse2.getIsKillCompleted()); for (int i = 0; i < 100; i++) { if (2 == rmService.getApplications(getRequest).getApplicationList().size()) { break; } Thread.sleep(10); } assertEquals("Incorrect number of apps in the RM", 2, rmService.getApplications(getRequest).getApplicationList().size()); }