ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @SuppressWarnings("unchecked") @Override public KillApplicationResponse forceKillApplication( KillApplicationRequest request) throws YarnException { ApplicationId applicationId = request.getApplicationId(); CallerContext callerContext = CallerContext.getCurrent(); 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, callerContext); 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, callerContext); 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, callerContext); throw RPCUtil.getRemoteException(new AccessControlException("User " + callerUGI.getShortUserName() + " cannot perform operation " + ApplicationAccessType.MODIFY_APP.name() + " on " + applicationId)); } if (application.isAppFinalStateStored()) { return KillApplicationResponse.newInstance(true); } StringBuilder message = new StringBuilder(); message.append("Application ").append(applicationId) .append(" was killed by user ").append(callerUGI.getShortUserName()); InetAddress remoteAddress = Server.getRemoteIp(); if (null != remoteAddress) { message.append(" at ").append(remoteAddress.getHostAddress()); } String diagnostics = org.apache.commons.lang3.StringUtils .trimToNull(request.getDiagnostics()); if (diagnostics != null) { message.append(" with diagnostic message: ") .append(diagnostics); } this.rmContext.getDispatcher().getEventHandler() .handle(new RMAppKillByClientEvent(applicationId, message.toString(), callerUGI, remoteAddress)); 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 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 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(); conf.setBoolean(MockRM.ENABLE_WEBAPP, true); MockRM rm = new MockRM(conf); rm.init(conf); rm.start(); ClientRMService rmService = rm.getClientRMService(); GetApplicationsRequest getRequest = GetApplicationsRequest.newInstance( EnumSet.of(YarnApplicationState.KILLED)); RMApp app1 = MockRMAppSubmitter.submitWithMemory(1024, rm); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(1024, rm) .withUnmanagedAM(true) .build(); RMApp app2 = MockRMAppSubmitter.submit(rm, data); assertEquals("Incorrect number of apps in the RM", 0, rmService.getApplications(getRequest).getApplicationList().size()); KillApplicationRequest killRequest1 = KillApplicationRequest.newInstance(app1.getApplicationId()); String diagnostic = "message1"; killRequest1.setDiagnostics(diagnostic); 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()); assertTrue("Diagnostic message is incorrect", app1.getDiagnostics().toString().contains(diagnostic)); 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()); }