Index: hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java =================================================================== --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java (revision 1571131) +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java (working copy) @@ -224,11 +224,16 @@ new Thread() { @Override public void run() { - LOG.info("Notifying ContainerManager to block new container-requests"); - containerManager.setBlockNewContainerRequests(true); - LOG.info("Cleaning up running containers on resync"); - containerManager.cleanupContainersOnNMResync(); - ((NodeStatusUpdaterImpl) nodeStatusUpdater ).rebootNodeStatusUpdater(); + try { + LOG.info("Notifying ContainerManager to block new container-requests"); + containerManager.setBlockNewContainerRequests(true); + LOG.info("Cleaning up running containers on resync"); + containerManager.cleanupContainersOnNMResync(); + ((NodeStatusUpdaterImpl) nodeStatusUpdater).rebootNodeStatusUpdater(); + } catch (YarnRuntimeException e) { + LOG.fatal("Error while rebooting NodeStatusUpdater.", e); + shutDown(); + } } }.start(); } Index: hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeManagerResync.java =================================================================== --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeManagerResync.java (revision 1571131) +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestNodeManagerResync.java (working copy) @@ -40,6 +40,7 @@ import org.apache.hadoop.yarn.event.Dispatcher; import org.apache.hadoop.yarn.exceptions.NMNotYetReadyException; import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.hadoop.yarn.factories.RecordFactory; import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider; import org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl; @@ -65,6 +66,7 @@ private FileContext localFS; private CyclicBarrier syncBarrier; private AtomicBoolean assertionFailedInThread = new AtomicBoolean(false); + private AtomicBoolean isNMShutdownCalled = new AtomicBoolean(false); @Before public void setup() throws UnsupportedFileSystemException { @@ -137,6 +139,30 @@ Assert.assertFalse(assertionFailedInThread.get()); nm.stop(); } + + @SuppressWarnings("unchecked") + @Test + public void testNMshutdownWhenResyncThrowException() throws IOException, + InterruptedException, YarnException { + NodeManager nm = new TestNodeManager3(); + YarnConfiguration conf = createNMConfig(); + nm.init(conf); + nm.start(); + Assert.assertEquals(1, ((TestNodeManager3) nm).getNMRegistrationCount()); + nm.getNMDispatcher().getEventHandler() + .handle(new NodeManagerEvent(NodeManagerEventType.RESYNC)); + + synchronized (isNMShutdownCalled) { + while (isNMShutdownCalled.get() == false) { + try { + isNMShutdownCalled.wait(); + } catch (InterruptedException e) { + } + } + } + + Assert.assertTrue("NM shutdown not called.",isNMShutdownCalled.get()); + } private YarnConfiguration createNMConfig() { YarnConfiguration conf = new YarnConfiguration(); @@ -322,4 +348,44 @@ } } } + + class TestNodeManager3 extends NodeManager { + + private int registrationCount = 0; + + @Override + protected NodeStatusUpdater createNodeStatusUpdater(Context context, + Dispatcher dispatcher, NodeHealthCheckerService healthChecker) { + return new TestNodeStatusUpdaterImpl3(context, dispatcher, healthChecker, + metrics); + } + + public int getNMRegistrationCount() { + return registrationCount; + } + + @Override + protected void shutDown() { + synchronized (isNMShutdownCalled) { + isNMShutdownCalled.set(true); + isNMShutdownCalled.notify(); + } + } + + class TestNodeStatusUpdaterImpl3 extends MockNodeStatusUpdater { + + public TestNodeStatusUpdaterImpl3(Context context, Dispatcher dispatcher, + NodeHealthCheckerService healthChecker, NodeManagerMetrics metrics) { + super(context, dispatcher, healthChecker, metrics); + } + + @Override + protected void registerWithRM() throws YarnException, IOException { + super.registerWithRM(); + registrationCount++; + if (registrationCount > 1) { + throw new YarnRuntimeException("Registration with RM failed."); + } + } + }} }