private void handleYarnContainerChange(String containerCountAsString) throws IOException, YarnException {
  String applicationId=yarnUtil.getRunningAppId(jobName,jobID);
  int containerCount=Integer.valueOf(containerCountAsString);
  int currentNumTask=getCurrentNumTasks();
  int currentNumContainers=getCurrentNumContainers();
  if (containerCount == currentNumContainers) {
    log.error("The new number of containers is equal to the current number of containers, skipping this message");
    return;
  }
  if (containerCount <= 0) {
    log.error("The number of containers cannot be zero or less, skipping this message");
    return;
  }
  if (containerCount > currentNumTask) {
    log.error("The number of containers cannot be more than the number of task, skipping this message");
    return;
  }
  log.info("Killing the current job");
  yarnUtil.killApplication(applicationId);
  coordinatorServerURL=null;
  try {
    String state=yarnUtil.getApplicationState(applicationId);
    Thread.sleep(1000);
    int countSleep=1;
    while (!state.equals("KILLED")) {
      state=yarnUtil.getApplicationState(applicationId);
      log.info("Job kill signal sent, but job not killed yet for " + applicationId + ". Sleeping for another 1000ms");
      Thread.sleep(1000);
      countSleep++;
      if (countSleep > 10) {
        throw new IllegalStateException("Job has not been killed after 10 attempts.");
      }
    }
  }
 catch (  InterruptedException e) {
    e.printStackTrace();
  }
  log.info("Killed the current job successfully");
  log.info("Staring the job again");
  skipUnreadMessages();
  JobRunner jobRunner=new JobRunner(config);
  jobRunner.run(false);
}

private void handleYarnContainerChange ( String containerCountAsString ) throws IOException , YarnException { String applicationId = yarnUtil . getRunningAppId ( jobName , jobID ) ; int containerCount = Integer . valueOf ( containerCountAsString ) ; int currentNumTask = getCurrentNumTasks ( ) ; int currentNumContainers = getCurrentNumContainers ( ) ; if ( containerCount == currentNumContainers ) { log . error ( "The new number of containers is equal to the current number of containers, skipping this message" ) ; return ; } if ( containerCount <= 0 ) { log . error ( "The number of containers cannot be zero or less, skipping this message" ) ; return ; } if ( containerCount > currentNumTask ) { log . error ( "The number of containers cannot be more than the number of task, skipping this message" ) ; return ; } log . info ( "Killing the current job" ) ; yarnUtil . killApplication ( applicationId ) ; coordinatorServerURL = null ; try { String state = yarnUtil . getApplicationState ( applicationId ) ; Thread . sleep ( 1000 ) ; int countSleep = 1 ; while ( ! state . equals ( "KILLED" ) ) { state = yarnUtil . getApplicationState ( applicationId ) ; log . info ( "Job kill signal sent, but job not killed yet for " + applicationId + ". Sleeping for another 1000ms" ) ; Thread . sleep ( 1000 ) ; countSleep ++ ; if ( countSleep > 10 ) { throw new IllegalStateException ( "Job has not been killed after 10 attempts." ) ; } } } catch ( InterruptedException e ) { e . printStackTrace ( ) ; } log . info ( "Killed the current job successfully" ) ; log . info ( "Staring the job again" ) ; skipUnreadMessages ( ) ; JobRunner jobRunner = new JobRunner ( config ) ; jobRunner . run ( false ) ; }
this method handles setconfig messages that want to change the number of containers of a job
