Repository: samza Updated Branches: refs/heads/master 242002ae0 -> 0548e6af9
SAMZA-842: Job does not quit when task shutdown is requested using ThreadJobFactory Project: http://git-wip-us.apache.org/repos/asf/samza/repo Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/0548e6af Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/0548e6af Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/0548e6af Branch: refs/heads/master Commit: 0548e6af9f1a31beff0f137af4eb2c5020a51a4a Parents: 242002a Author: Tommy Becker <[email protected]> Authored: Mon Sep 12 15:26:10 2016 -0700 Committer: Yi Pan (Data Infrastructure) <[email protected]> Committed: Wed Sep 14 01:40:09 2016 -0700 ---------------------------------------------------------------------- .../org/apache/samza/job/local/ThreadJobFactory.scala | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/samza/blob/0548e6af/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala ---------------------------------------------------------------------- diff --git a/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala b/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala index 56881d4..eaab3a6 100644 --- a/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala +++ b/samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala @@ -48,9 +48,18 @@ class ThreadJobFactory extends StreamJobFactory with Logging { try { coordinator.start - new ThreadJob(SamzaContainer(containerModel, coordinator.jobModel, new JmxServer)) + new ThreadJob(new Runnable { + override def run(): Unit = { + val jmxServer = new JmxServer + try { + SamzaContainer(containerModel, coordinator.jobModel, jmxServer).run() + } finally { + jmxServer.stop + } + } + }) } finally { coordinator.stop } } -} \ No newline at end of file +}
