diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/event/TestAsyncDispatcher.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/event/TestAsyncDispatcher.java index ba0deff..018096b 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/event/TestAsyncDispatcher.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/event/TestAsyncDispatcher.java @@ -50,7 +50,9 @@ public void testDispatcherOnCloseIfQueueEmpty() throws Exception { disp.waitForEventThreadToWait(); try { disp.getEventHandler().handle(event); + Assert.fail("Expected YarnRuntimeException"); } catch (YarnRuntimeException e) { + Assert.assertTrue(e.getCause() instanceof InterruptedException); } // Queue should be empty and dispatcher should not hang on close Assert.assertTrue("Event Queue should have been empty", diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java index 3eefb8f..bde56cc 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java @@ -1036,6 +1036,13 @@ void continuousSchedulingAttempt() throws InterruptedException { } catch (Throwable ex) { LOG.error("Error while attempting scheduling for node " + node + ": " + ex.toString(), ex); + if ((ex instanceof YarnRuntimeException) && + (ex.getCause() instanceof InterruptedException)) { + // AsyncDispatcher translates InterruptedException to + // YarnRuntimeException with cause InterruptedException. + // Need to throw InterruptedException to stop schedulingThread. + throw (InterruptedException)ex.getCause(); + } } } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java index c352cc9..a02cf18 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java @@ -27,7 +27,10 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; import java.io.File; @@ -4320,6 +4323,34 @@ public void testContinuousSchedulingWithNodeRemoved() throws Exception { } @Test + public void testContinuousSchedulingInterruptedException() + throws Exception { + scheduler.init(conf); + scheduler.start(); + FairScheduler spyScheduler = spy(scheduler); + Assert.assertTrue("Continuous scheduling should be disabled.", + !spyScheduler.isContinuousSchedulingEnabled()); + // Add one nodes + RMNode node1 = + MockNodes.newNodeInfo(1, Resources.createResource(8 * 1024, 8), 1, + "127.0.0.1"); + NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1); + spyScheduler.handle(nodeEvent1); + Assert.assertEquals("We should have one alive node.", + 1, spyScheduler.getNumClusterNodes()); + InterruptedException ie = new InterruptedException(); + doThrow(new YarnRuntimeException(ie)).when(spyScheduler). + attemptScheduling(isA(FSSchedulerNode.class)); + // Invoke the continuous scheduling once + try { + spyScheduler.continuousSchedulingAttempt(); + fail("Expected InterruptedException to stop schedulingThread"); + } catch (InterruptedException e) { + Assert.assertEquals(ie, e); + } + } + + @Test public void testSchedulingOnRemovedNode() throws Exception { // Disable continuous scheduling, will invoke continuous scheduling manually scheduler.init(conf);