YARN-5999. AMRMClientAsync will stop if any exceptions thrown on allocate call. Contributed by Jian He
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/64a2d5be Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/64a2d5be Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/64a2d5be Branch: refs/heads/YARN-5085 Commit: 64a2d5be91a7f344aadf820481c542c967bc46a8 Parents: f5e0bd3 Author: Xuan <[email protected]> Authored: Wed Dec 14 14:33:23 2016 -0800 Committer: Xuan <[email protected]> Committed: Wed Dec 14 14:33:23 2016 -0800 ---------------------------------------------------------------------- .../api/async/impl/AMRMClientAsyncImpl.java | 30 ++++++++------------ .../api/async/impl/TestAMRMClientAsync.java | 2 +- 2 files changed, 13 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/64a2d5be/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java index bc6cadd..3dd53d3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java @@ -61,7 +61,7 @@ extends AMRMClientAsync<T> { private final HeartbeatThread heartbeatThread; private final CallbackHandlerThread handlerThread; - private final BlockingQueue<AllocateResponse> responseQueue; + private final BlockingQueue<Object> responseQueue; private final Object unregisterHeartbeatLock = new Object(); @@ -70,8 +70,6 @@ extends AMRMClientAsync<T> { private volatile String collectorAddr; - private volatile Throwable savedException; - /** * * @param intervalMs heartbeat interval in milliseconds between AM and RM @@ -90,7 +88,6 @@ extends AMRMClientAsync<T> { handlerThread = new CallbackHandlerThread(); responseQueue = new LinkedBlockingQueue<>(); keepRunning = true; - savedException = null; } /** @@ -111,9 +108,8 @@ extends AMRMClientAsync<T> { super(client, intervalMs, callbackHandler); heartbeatThread = new HeartbeatThread(); handlerThread = new CallbackHandlerThread(); - responseQueue = new LinkedBlockingQueue<AllocateResponse>(); + responseQueue = new LinkedBlockingQueue<Object>(); keepRunning = true; - savedException = null; } @Override @@ -265,7 +261,7 @@ extends AMRMClientAsync<T> { public void run() { while (true) { - AllocateResponse response = null; + Object response = null; // synchronization ensures we don't send heartbeats after unregistering synchronized (unregisterHeartbeatLock) { if (!keepRunning) { @@ -280,10 +276,7 @@ extends AMRMClientAsync<T> { return; } catch (Throwable ex) { LOG.error("Exception on heartbeat", ex); - savedException = ex; - // interrupt handler thread in case it waiting on the queue - handlerThread.interrupt(); - return; + response = ex; } if (response != null) { while (true) { @@ -316,19 +309,20 @@ extends AMRMClientAsync<T> { return; } try { - AllocateResponse response; - if(savedException != null) { - LOG.error("Stopping callback due to: ", savedException); - handler.onError(savedException); - return; - } + Object object; try { - response = responseQueue.take(); + object = responseQueue.take(); } catch (InterruptedException ex) { LOG.info("Interrupted while waiting for queue", ex); continue; } + if (object instanceof Throwable) { + progress = handler.getProgress(); + handler.onError((Throwable) object); + continue; + } + AllocateResponse response = (AllocateResponse) object; String collectorAddress = response.getCollectorAddr(); TimelineClient timelineClient = client.getRegisteredTimelineClient(); if (timelineClient != null && collectorAddress != null http://git-wip-us.apache.org/repos/asf/hadoop/blob/64a2d5be/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/async/impl/TestAMRMClientAsync.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/async/impl/TestAMRMClientAsync.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/async/impl/TestAMRMClientAsync.java index dac82e4..ba38340 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/async/impl/TestAMRMClientAsync.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/async/impl/TestAMRMClientAsync.java @@ -213,7 +213,7 @@ public class TestAMRMClientAsync { asyncClient.stop(); // stopping should have joined all threads and completed all callbacks - Assert.assertTrue(callbackHandler.callbackCount == 0); + Assert.assertTrue(callbackHandler.callbackCount > 0); } @Test (timeout = 10000) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
