From 90c0ecdef721a2b041c30ea22be032a93ada4a0d Mon Sep 17 00:00:00 2001 From: PanTheMan Date: Fri, 27 Mar 2020 00:07:28 -0700 Subject: [PATCH] Close MetadataResourceUtil when job is stopped --- .../ClusterBasedJobCoordinator.java | 8 +++++- .../coordinator/MetadataResourceUtil.java | 4 +++ .../standalone/PassthroughJobCoordinator.java | 4 ++- .../org/apache/samza/zk/ZkJobCoordinator.java | 3 ++- .../apache/samza/job/local/ProcessJob.scala | 6 +++-- .../samza/job/local/ProcessJobFactory.scala | 2 +- .../samza/job/local/ThreadJobFactory.scala | 1 + .../samza/job/local/TestProcessJob.scala | 26 ++++++++++++++++--- 8 files changed, 45 insertions(+), 9 deletions(-) diff --git a/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java index e7647f40ff..ef921a1ef9 100644 --- a/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java +++ b/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java @@ -175,6 +175,11 @@ public class ClusterBasedJobCoordinator { private final SystemAdmins systemAdmins; + /** + * Util to close the KafkaCheckpointManager later on + */ + private MetadataResourceUtil metadataResourceUtil; + /** * Internal variable for the instance of {@link JmxServer} */ @@ -259,7 +264,7 @@ public void run() { //create necessary checkpoint and changelog streams, if not created JobModel jobModel = jobModelManager.jobModel(); - MetadataResourceUtil metadataResourceUtil = new MetadataResourceUtil(jobModel, this.metrics, config); + metadataResourceUtil = new MetadataResourceUtil(jobModel, this.metrics, config); metadataResourceUtil.createResources(); // fan out the startpoints @@ -341,6 +346,7 @@ private void onShutDown() { systemAdmins.stop(); shutDowncontainerPlacementRequestAllocatorAndUtils(); containerProcessManager.stop(); + metadataResourceUtil.stop(); metadataStore.close(); } catch (Throwable e) { LOG.error("Exception while stopping cluster based job coordinator", e); diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java b/samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java index 1050662920..c2081c67e3 100644 --- a/samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java +++ b/samza-core/src/main/java/org/apache/samza/coordinator/MetadataResourceUtil.java @@ -59,6 +59,10 @@ public void createResources() { createChangelogStreams(); } + public void stop() { + checkpointManager.stop(); + } + @VisibleForTesting void createChangelogStreams() { ChangelogStreamManager.createChangelogStreams(config, jobModel.maxChangeLogStreamPartitions); diff --git a/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java index 9fb9975ea2..6fb4141517 100644 --- a/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java +++ b/samza-core/src/main/java/org/apache/samza/standalone/PassthroughJobCoordinator.java @@ -67,6 +67,7 @@ public class PassthroughJobCoordinator implements JobCoordinator { private final String processorId; private final Config config; private final LocationId locationId; + private MetadataResourceUtil metadataResourceUtil; private JobCoordinatorListener coordinatorListener = null; public PassthroughJobCoordinator(String processorId, Config config, MetricsRegistry metricsRegistry) { @@ -85,7 +86,7 @@ public void start() { try { jobModel = getJobModel(); // TODO metrics registry has been null here for a while; is it safe? - MetadataResourceUtil metadataResourceUtil = new MetadataResourceUtil(jobModel, null, config); + metadataResourceUtil = new MetadataResourceUtil(jobModel, null, config); metadataResourceUtil.createResources(); } catch (Exception e) { LOGGER.error("Exception while trying to getJobModel.", e); @@ -111,6 +112,7 @@ public void stop() { coordinatorListener.onJobModelExpired(); coordinatorListener.onCoordinatorStop(); } + metadataResourceUtil.stop(); } @Override diff --git a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java index feaabba34b..8175ffa73e 100644 --- a/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java +++ b/samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java @@ -114,6 +114,7 @@ public class ZkJobCoordinator implements JobCoordinator { private JobModel newJobModel; private boolean hasLoadedMetadataResources = false; private String cachedJobModelVersion = null; + private MetadataResourceUtil metadataResourceUtil; @VisibleForTesting ZkSessionMetrics zkSessionMetrics; @@ -312,7 +313,7 @@ JobModel readJobModelFromMetadataStore(String zkJobModelVersion) { @VisibleForTesting void loadMetadataResources(JobModel jobModel) { try { - MetadataResourceUtil metadataResourceUtil = createMetadataResourceUtil(jobModel, config); + metadataResourceUtil = createMetadataResourceUtil(jobModel, config); metadataResourceUtil.createResources(); if (coordinatorStreamStore != null) { diff --git a/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala b/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala index 9e0285254e..dd606d1943 100644 --- a/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala +++ b/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJob.scala @@ -21,7 +21,7 @@ package org.apache.samza.job.local import java.util.concurrent.CountDownLatch -import org.apache.samza.coordinator.JobModelManager +import org.apache.samza.coordinator.{JobModelManager, MetadataResourceUtil} import org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore import org.apache.samza.job.ApplicationStatus.{New, Running, SuccessfulFinish, UnsuccessfulFinish} import org.apache.samza.job.{ApplicationStatus, CommandBuilder, StreamJob} @@ -45,7 +45,8 @@ object ProcessJob { class ProcessJob( commandBuilder: CommandBuilder, val jobModelManager: JobModelManager, - val coordinatorStreamStore: CoordinatorStreamStore) extends StreamJob with Logging { + val coordinatorStreamStore: CoordinatorStreamStore, + val metadataResourceUtil: MetadataResourceUtil) extends StreamJob with Logging { import ProcessJob._ @@ -76,6 +77,7 @@ class ProcessJob( case e: Exception => error("Encountered an error during job start: %s".format(e.getMessage)) } finally { jobModelManager.stop + metadataResourceUtil.stop coordinatorStreamStore.close setStatus(if (processExitCode == 0) SuccessfulFinish else UnsuccessfulFinish) } diff --git a/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJobFactory.scala b/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJobFactory.scala index fab8c6ee69..c4f40e03e3 100644 --- a/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJobFactory.scala +++ b/samza-core/src/main/scala/org/apache/samza/job/local/ProcessJobFactory.scala @@ -113,6 +113,6 @@ class ProcessJobFactory extends StreamJobFactory with Logging { .setId("0") .setUrl(jobModelManager.server.getUrl) - new ProcessJob(commandBuilder, jobModelManager, coordinatorStreamStore) + new ProcessJob(commandBuilder, jobModelManager, coordinatorStreamStore, metadataResourceUtil) } } 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 c1a36837f5..e70510851e 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 @@ -158,6 +158,7 @@ class ThreadJobFactory extends StreamJobFactory with Logging { threadJob } finally { jobModelManager.stop + metadataResourceUtil.stop if (jmxServer != null) { jmxServer.stop } diff --git a/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala b/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala index d4f4d41fed..70a0194763 100644 --- a/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala +++ b/samza-core/src/test/scala/org/apache/samza/job/local/TestProcessJob.scala @@ -21,7 +21,7 @@ package org.apache.samza.job.local import com.google.common.collect.ImmutableMap import org.apache.samza.config.MapConfig -import org.apache.samza.coordinator.JobModelManager +import org.apache.samza.coordinator.{JobModelManager, MetadataResourceUtil} import org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore import org.apache.samza.coordinator.stream.MockCoordinatorStreamSystemFactory import org.apache.samza.job.ApplicationStatus.{Running, SuccessfulFinish, UnsuccessfulFinish} @@ -59,7 +59,8 @@ object TestProcessJob { new ProcessJob( commandBuilder, new MockJobModelManager, - new MockCoordinateStreamStore(MockConfigs, systemProducer, systemConsumer, systemAdmin)) + new MockCoordinateStreamStore(MockConfigs, systemProducer, systemConsumer, systemAdmin), + new MockMetadataResourceUtil) } private def getMockJobModelManager(processJob: ProcessJob): MockJobModelManager = { @@ -69,6 +70,10 @@ object TestProcessJob { private def getMockCoordinatorStreamStore(processJob: ProcessJob): MockCoordinateStreamStore = { processJob.coordinatorStreamStore.asInstanceOf[MockCoordinateStreamStore] } + + private def getMockMetadataResourceUtil(processJob: ProcessJob): MockMetadataResourceUtil = { + processJob.metadataResourceUtil.asInstanceOf[MockMetadataResourceUtil] + } } class TestProcessJob { @@ -84,6 +89,7 @@ class TestProcessJob { assertEquals(SuccessfulFinish, status) assertTrue(getMockJobModelManager(processJob).stopped) assertTrue(getMockCoordinatorStreamStore(processJob).closed) + assertTrue(getMockMetadataResourceUtil(processJob).stopped) } @Test @@ -95,6 +101,7 @@ class TestProcessJob { assertEquals(UnsuccessfulFinish, status) assertTrue(getMockJobModelManager(processJob).stopped) assertTrue(getMockCoordinatorStreamStore(processJob).closed) + assertTrue(getMockMetadataResourceUtil(processJob).stopped) } @Test @@ -116,6 +123,7 @@ class TestProcessJob { assertEquals(UnsuccessfulFinish, processJob.getStatus) assertTrue(getMockJobModelManager(processJob).stopped) assertTrue(getMockCoordinatorStreamStore(processJob).closed) + assertTrue(getMockMetadataResourceUtil(processJob).stopped) } @Test @@ -127,6 +135,7 @@ class TestProcessJob { assertEquals(UnsuccessfulFinish, processJob.getStatus) assertTrue(getMockJobModelManager(processJob).stopped) assertTrue(getMockCoordinatorStreamStore(processJob).closed) + assertTrue(getMockMetadataResourceUtil(processJob).stopped) } @Test @@ -138,6 +147,7 @@ class TestProcessJob { assertEquals(SuccessfulFinish, processJob.getStatus) assertTrue(getMockJobModelManager(processJob).stopped) assertTrue(getMockCoordinatorStreamStore(processJob).closed) + assertTrue(getMockMetadataResourceUtil(processJob).stopped) } @Test @@ -186,4 +196,14 @@ class MockCoordinateStreamStore( override def close: Unit = { closed = true } -} \ No newline at end of file +} + +class MockMetadataResourceUtil extends MetadataResourceUtil(null, null, null) { + var stopped: Boolean = false + + override def createResources: Unit = {} + + override def stop: Unit = { + stopped = true + } +}