YARN-3697. FairScheduler: ContinuousSchedulingThread can fail to shutdown. (Zhihai Xu via kasha)
(cherry picked from commit 332b520a48
)
Conflicts:
hadoop-yarn-project/CHANGES.txt
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
This commit is contained in:
parent
30ebb792f9
commit
f3fd5e30cd
|
@ -73,6 +73,9 @@ Release 2.7.2 - UNRELEASED
|
||||||
|
|
||||||
YARN-4153. TestAsyncDispatcher failed at branch-2.7 (Zhihai Xu via jianhe)
|
YARN-4153. TestAsyncDispatcher failed at branch-2.7 (Zhihai Xu via jianhe)
|
||||||
|
|
||||||
|
YARN-3697. FairScheduler: ContinuousSchedulingThread can fail to shutdown.
|
||||||
|
(Zhihai Xu via kasha)
|
||||||
|
|
||||||
YARN-4158. Remove duplicate close for LogWriter in
|
YARN-4158. Remove duplicate close for LogWriter in
|
||||||
AppLogAggregatorImpl#uploadLogsForContainers (Zhihai Xu via jlowe)
|
AppLogAggregatorImpl#uploadLogsForContainers (Zhihai Xu via jlowe)
|
||||||
|
|
||||||
|
|
|
@ -50,7 +50,9 @@ public class TestAsyncDispatcher {
|
||||||
disp.waitForEventThreadToWait();
|
disp.waitForEventThreadToWait();
|
||||||
try {
|
try {
|
||||||
disp.getEventHandler().handle(event);
|
disp.getEventHandler().handle(event);
|
||||||
|
Assert.fail("Expected YarnRuntimeException");
|
||||||
} catch (YarnRuntimeException e) {
|
} catch (YarnRuntimeException e) {
|
||||||
|
Assert.assertTrue(e.getCause() instanceof InterruptedException);
|
||||||
}
|
}
|
||||||
// Queue should be empty and dispatcher should not hang on close
|
// Queue should be empty and dispatcher should not hang on close
|
||||||
Assert.assertTrue("Event Queue should have been empty",
|
Assert.assertTrue("Event Queue should have been empty",
|
||||||
|
|
|
@ -1010,6 +1010,13 @@ public class FairScheduler extends
|
||||||
} catch (Throwable ex) {
|
} catch (Throwable ex) {
|
||||||
LOG.error("Error while attempting scheduling for node " + node +
|
LOG.error("Error while attempting scheduling for node " + node +
|
||||||
": " + ex.toString(), ex);
|
": " + 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();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -27,7 +27,10 @@ import static org.junit.Assert.assertNull;
|
||||||
import static org.junit.Assert.assertSame;
|
import static org.junit.Assert.assertSame;
|
||||||
import static org.junit.Assert.assertTrue;
|
import static org.junit.Assert.assertTrue;
|
||||||
import static org.junit.Assert.fail;
|
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.mock;
|
||||||
|
import static org.mockito.Mockito.spy;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
|
@ -3950,6 +3953,34 @@ public class TestFairScheduler extends FairSchedulerTestBase {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@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
|
@Test
|
||||||
public void testDontAllowUndeclaredPools() throws Exception {
|
public void testDontAllowUndeclaredPools() throws Exception {
|
||||||
conf.setBoolean(FairSchedulerConfiguration.ALLOW_UNDECLARED_POOLS, false);
|
conf.setBoolean(FairSchedulerConfiguration.ALLOW_UNDECLARED_POOLS, false);
|
||||||
|
|
Loading…
Reference in New Issue