@Test public void testRMConnectionRetry() throws Exception { Configuration conf = new Configuration(); LocalContainerAllocator lca = new StubbedLocalContainerAllocator(); lca.init(conf); lca.start(); try { lca.heartbeat(); Assert.fail("heartbeat was supposed to throw"); } catch (YarnException e) { } finally { lca.stop(); } conf.setLong(MRJobConfig.MR_AM_TO_RM_WAIT_INTERVAL_MS, 0); lca = new StubbedLocalContainerAllocator(); lca.init(conf); lca.start(); try { lca.heartbeat(); Assert.fail("heartbeat was supposed to throw"); } catch (YarnRuntimeException e) { } finally { lca.stop(); } }
@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; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } 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.getAMCommand() != null) { switch(allocateResponse.getAMCommand()) { case AM_RESYNC: LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resyncing."); this.lastResponseID = 0; register(); break; case AM_SHUTDOWN: 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()); default: String msg = "Unhandled value of AMCommand: " + allocateResponse.getAMCommand(); LOG.error(msg); throw new YarnRuntimeException(msg); } } }
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; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } 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.getAMCommand() != null) { switch(allocateResponse.getAMCommand()) { case AM_RESYNC: LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resyncing."); this.lastResponseID = 0; register(); break; case AM_SHUTDOWN: 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()); default: String msg = "Unhandled value of AMCommand: " + allocateResponse.getAMCommand(); LOG.error(msg); throw new YarnRuntimeException(msg); } } } }
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; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } 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.getAMCommand() != null) { switch(allocateResponse.getAMCommand()) { case AM_RESYNC: LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resyncing."); this.lastResponseID = 0; register(); break; case AM_SHUTDOWN: 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()); default: String msg = "Unhandled value of AMCommand: " + allocateResponse.getAMCommand(); LOG.error(msg); throw new YarnRuntimeException(msg); } } } 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; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } 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.getAMCommand() != null) { switch(allocateResponse.getAMCommand()) { case AM_RESYNC: LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resyncing."); this.lastResponseID = 0; register(); break; case AM_SHUTDOWN: 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()); default: String msg = "Unhandled value of AMCommand: " + allocateResponse.getAMCommand(); LOG.error(msg); throw new YarnRuntimeException(msg); } } } 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; try { allocateResponse = scheduler.allocate(allocateRequest); retrystartTime = System.currentTimeMillis(); } 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.getAMCommand() != null) { switch(allocateResponse.getAMCommand()) { case AM_RESYNC: LOG.info("ApplicationMaster is out of sync with ResourceManager," + " hence resyncing."); this.lastResponseID = 0; register(); break; case AM_SHUTDOWN: 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()); default: String msg = "Unhandled value of AMCommand: " + allocateResponse.getAMCommand(); LOG.error(msg); throw new YarnRuntimeException(msg); } } } LocalContainerAllocator(ClientService clientService,
    AppContext context, String nmHost, int nmPort, int nmHttpPort
    , ContainerId cId); @SuppressWarnings("unchecked") @Override void handle(ContainerAllocatorEvent event);  }