SLIDER-302: propagation working with correct ordering
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/3d02e1cb Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/3d02e1cb Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/3d02e1cb Branch: refs/heads/feature/SLIDER-149_Support_a_YARN_service_registry Commit: 3d02e1cbb072a795b58ac7e263b30d7370f5b133 Parents: 7ef046b Author: Steve Loughran <ste...@apache.org> Authored: Fri Aug 8 17:07:24 2014 +0100 Committer: Steve Loughran <ste...@apache.org> Committed: Fri Aug 8 17:07:24 2014 +0100 ---------------------------------------------------------------------- .../slider/server/appmaster/actions/QueueService.java | 10 +++++++--- .../slider/server/appmaster/actions/TestActions.groovy | 12 ++++++------ 2 files changed, 13 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/3d02e1cb/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java index dd39de0..69cb9f7 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java @@ -41,11 +41,13 @@ public class QueueService extends AbstractService super("action queue"); } - public void putImmediately(AsyncAction action) { + public void put(AsyncAction action) { + log.debug("Queueing {}", action); actionQueue.add(action); } public void putDelayed(AsyncAction action) { + log.debug("Delayed Queueing {}", action); delayedActions.add(action); } @@ -56,13 +58,15 @@ public class QueueService extends AbstractService public void run() { try { - log.info("ActionExecutor Queue terminated"); + log.info("QueueService processor started"); AsyncAction take; do { take = delayedActions.take(); - actionQueue.push(take); + log.debug("Propagating {}", take); + actionQueue.put(take); } while (!(take instanceof ActionStopQueue)); + log.info("QueueService processor terminated"); } catch (InterruptedException e) { //game over } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/3d02e1cb/slider-core/src/test/groovy/org/apache/slider/server/appmaster/actions/TestActions.groovy ---------------------------------------------------------------------- diff --git a/slider-core/src/test/groovy/org/apache/slider/server/appmaster/actions/TestActions.groovy b/slider-core/src/test/groovy/org/apache/slider/server/appmaster/actions/TestActions.groovy index 24d4c40..7e6338b 100644 --- a/slider-core/src/test/groovy/org/apache/slider/server/appmaster/actions/TestActions.groovy +++ b/slider-core/src/test/groovy/org/apache/slider/server/appmaster/actions/TestActions.groovy @@ -103,8 +103,8 @@ class TestActions { @Test public void testImmediateQueue() throws Throwable { ActionNoteExecuted noteExecuted = new ActionNoteExecuted("executed", 0) - queues.putImmediately(noteExecuted) - queues.putImmediately(new ActionStopQueue(0)) + queues.put(noteExecuted) + queues.put(new ActionStopQueue(0)) QueueExecutor ex = new QueueExecutor(queues) ex.run(); assert queues.actionQueue.empty @@ -120,9 +120,9 @@ class TestActions { List<AsyncAction> actions = [note1, stop, note2] Collections.sort(actions) - assert actions.get(0) == note1 - assert actions.get(1) == note2 - assert actions.get(3) == stop + assert actions[0] == note1 + assert actions[1] == note2 + assert actions[2] == stop } @Test @@ -136,8 +136,8 @@ class TestActions { assert note1.nanos < note2.nanos assert note2.nanos < stop.nanos queues.putDelayed(note1) - queues.putDelayed(stop) queues.putDelayed(note2) + queues.putDelayed(stop) // async to sync expected to run in order queues.run(); assert queues.delayedActions.empty
