@Test public void testAllocResponseId() throws Exception { ApplicationMasterProtocol scheduler = new MockScheduler(); Configuration conf = new Configuration(); LocalContainerAllocator lca = new StubbedLocalContainerAllocator(scheduler); lca.init(conf); lca.start(); lca.heartbeat(); lca.heartbeat(); lca.close(); }
@SuppressWarnings("unchecked") @Override protected synchronized void heartbeat() throws Exception { AllocateRequest allocateRequest = AllocateRequest.newInstance(this.lastResponseID, super.getApplicationProgress(), new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>(), null); AllocateResponse allocateResponse = null; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } catch (ApplicationAttemptNotFoundException e) { LOG.info("Event from RM: shutting down Application Master"); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.JOB_AM_REBOOT)); throw new YarnRuntimeException( "Resource Manager doesn't recognize AttemptId: " + this.getContext().getApplicationID(), e); } catch (ApplicationMasterNotRegisteredException e) { LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resync and send outstanding requests."); this.lastResponseID = 0; register(); } catch (Exception e) { if (System.currentTimeMillis() - retrystartTime >= retryInterval) { LOG.error("Could not contact RM after " + retryInterval + " milliseconds."); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.INTERNAL_ERROR)); throw new YarnRuntimeException("Could not contact RM after " + retryInterval + " milliseconds."); } throw e; } if (allocateResponse != null) { this.lastResponseID = allocateResponse.getResponseId(); Token token = allocateResponse.getAMRMToken(); if (token != null) { updateAMRMToken(token); } } }
LocalContainerAllocator extends RMCommunicator implements ContainerAllocator { @SuppressWarnings("unchecked") @Override protected synchronized void heartbeat() throws Exception { AllocateRequest allocateRequest = AllocateRequest.newInstance(this.lastResponseID, super.getApplicationProgress(), new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>(), null); AllocateResponse allocateResponse = null; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } catch (ApplicationAttemptNotFoundException e) { LOG.info("Event from RM: shutting down Application Master"); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.JOB_AM_REBOOT)); throw new YarnRuntimeException( "Resource Manager doesn't recognize AttemptId: " + this.getContext().getApplicationID(), e); } catch (ApplicationMasterNotRegisteredException e) { LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resync and send outstanding requests."); this.lastResponseID = 0; register(); } catch (Exception e) { if (System.currentTimeMillis() - retrystartTime >= retryInterval) { LOG.error("Could not contact RM after " + retryInterval + " milliseconds."); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.INTERNAL_ERROR)); throw new YarnRuntimeException("Could not contact RM after " + retryInterval + " milliseconds."); } throw e; } if (allocateResponse != null) { this.lastResponseID = allocateResponse.getResponseId(); Token token = allocateResponse.getAMRMToken(); if (token != null) { updateAMRMToken(token); } } } }
LocalContainerAllocator extends RMCommunicator implements ContainerAllocator { @SuppressWarnings("unchecked") @Override protected synchronized void heartbeat() throws Exception { AllocateRequest allocateRequest = AllocateRequest.newInstance(this.lastResponseID, super.getApplicationProgress(), new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>(), null); AllocateResponse allocateResponse = null; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } catch (ApplicationAttemptNotFoundException e) { LOG.info("Event from RM: shutting down Application Master"); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.JOB_AM_REBOOT)); throw new YarnRuntimeException( "Resource Manager doesn't recognize AttemptId: " + this.getContext().getApplicationID(), e); } catch (ApplicationMasterNotRegisteredException e) { LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resync and send outstanding requests."); this.lastResponseID = 0; register(); } catch (Exception e) { if (System.currentTimeMillis() - retrystartTime >= retryInterval) { LOG.error("Could not contact RM after " + retryInterval + " milliseconds."); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.INTERNAL_ERROR)); throw new YarnRuntimeException("Could not contact RM after " + retryInterval + " milliseconds."); } throw e; } if (allocateResponse != null) { this.lastResponseID = allocateResponse.getResponseId(); Token token = allocateResponse.getAMRMToken(); if (token != null) { updateAMRMToken(token); } } } LocalContainerAllocator(ClientService clientService,
    AppContext context, String nmHost, int nmPort, int nmHttpPort
    , ContainerId cId); }
LocalContainerAllocator extends RMCommunicator implements ContainerAllocator { @SuppressWarnings("unchecked") @Override protected synchronized void heartbeat() throws Exception { AllocateRequest allocateRequest = AllocateRequest.newInstance(this.lastResponseID, super.getApplicationProgress(), new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>(), null); AllocateResponse allocateResponse = null; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } catch (ApplicationAttemptNotFoundException e) { LOG.info("Event from RM: shutting down Application Master"); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.JOB_AM_REBOOT)); throw new YarnRuntimeException( "Resource Manager doesn't recognize AttemptId: " + this.getContext().getApplicationID(), e); } catch (ApplicationMasterNotRegisteredException e) { LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resync and send outstanding requests."); this.lastResponseID = 0; register(); } catch (Exception e) { if (System.currentTimeMillis() - retrystartTime >= retryInterval) { LOG.error("Could not contact RM after " + retryInterval + " milliseconds."); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.INTERNAL_ERROR)); throw new YarnRuntimeException("Could not contact RM after " + retryInterval + " milliseconds."); } throw e; } if (allocateResponse != null) { this.lastResponseID = allocateResponse.getResponseId(); Token token = allocateResponse.getAMRMToken(); if (token != null) { updateAMRMToken(token); } } } LocalContainerAllocator(ClientService clientService,
    AppContext context, String nmHost, int nmPort, int nmHttpPort
    , ContainerId cId); @SuppressWarnings("unchecked") @Override void handle(ContainerAllocatorEvent event); }
LocalContainerAllocator extends RMCommunicator implements ContainerAllocator { @SuppressWarnings("unchecked") @Override protected synchronized void heartbeat() throws Exception { AllocateRequest allocateRequest = AllocateRequest.newInstance(this.lastResponseID, super.getApplicationProgress(), new ArrayList<ResourceRequest>(), new ArrayList<ContainerId>(), null); AllocateResponse allocateResponse = null; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } catch (ApplicationAttemptNotFoundException e) { LOG.info("Event from RM: shutting down Application Master"); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.JOB_AM_REBOOT)); throw new YarnRuntimeException( "Resource Manager doesn't recognize AttemptId: " + this.getContext().getApplicationID(), e); } catch (ApplicationMasterNotRegisteredException e) { LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resync and send outstanding requests."); this.lastResponseID = 0; register(); } catch (Exception e) { if (System.currentTimeMillis() - retrystartTime >= retryInterval) { LOG.error("Could not contact RM after " + retryInterval + " milliseconds."); eventHandler.handle(new JobEvent(this.getJob().getID(), JobEventType.INTERNAL_ERROR)); throw new YarnRuntimeException("Could not contact RM after " + retryInterval + " milliseconds."); } throw e; } if (allocateResponse != null) { this.lastResponseID = allocateResponse.getResponseId(); Token token = allocateResponse.getAMRMToken(); if (token != null) { updateAMRMToken(token); } } } LocalContainerAllocator(ClientService clientService,
    AppContext context, String nmHost, int nmPort, int nmHttpPort
    , ContainerId cId); @SuppressWarnings("unchecked") @Override void handle(ContainerAllocatorEvent event);  }