From b314d69de1d692643f1c1ef1aaed27c62ab22664 Mon Sep 17 00:00:00 2001 From: Daniel Nishimura Date: Thu, 8 Aug 2019 11:30:28 -0700 Subject: [PATCH 1/2] SAMZA-2296: Metadata streams not created in Standalone --- .../ClusterBasedJobCoordinator.java | 2 +- .../coordinator/MetadataResourceUtil.java | 21 +++++---- .../standalone/PassthroughJobCoordinator.java | 2 +- .../org/apache/samza/zk/ZkJobCoordinator.java | 6 +-- .../samza/job/local/ProcessJobFactory.scala | 2 +- .../samza/job/local/ThreadJobFactory.scala | 2 +- .../coordinator/TestMetadataResourceUtil.java | 44 ++++++++++++++++--- .../apache/samza/zk/TestZkJobCoordinator.java | 2 +- 8 files changed, 57 insertions(+), 24 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 ef46205f2f..b22bf2370c 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 @@ -234,7 +234,7 @@ public void run() { //create necessary checkpoint and changelog streams, if not created JobModel jobModel = jobModelManager.jobModel(); MetadataResourceUtil metadataResourceUtil = - new MetadataResourceUtil(jobModel, this.metrics, getClass().getClassLoader()); + new MetadataResourceUtil(jobModel, this.metrics, getClass().getClassLoader(), config); metadataResourceUtil.createResources(); // fan out the startpoints 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 148e9427b6..ee59ca0cd3 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 @@ -21,6 +21,7 @@ import com.google.common.annotations.VisibleForTesting; import org.apache.samza.checkpoint.CheckpointManager; +import org.apache.samza.config.Config; import org.apache.samza.config.TaskConfig; import org.apache.samza.job.model.JobModel; import org.apache.samza.metrics.MetricsRegistry; @@ -33,24 +34,20 @@ // TODO: Replace with a metadata admin interface when the {@link MetadataStore} is fully augmented to handle all metadata sources. public class MetadataResourceUtil { private final CheckpointManager checkpointManager; + private final Config config; private final JobModel jobModel; // TODO: Should be loaded by metadata store in the future /** * @param jobModel the loaded {@link JobModel} * @param metricsRegistry the registry for reporting metrics. */ - public MetadataResourceUtil(JobModel jobModel, MetricsRegistry metricsRegistry, ClassLoader classLoader) { + public MetadataResourceUtil(JobModel jobModel, MetricsRegistry metricsRegistry, ClassLoader classLoader, Config config) { + this.config = config; this.jobModel = jobModel; - TaskConfig taskConfig = new TaskConfig(jobModel.getConfig()); + TaskConfig taskConfig = new TaskConfig(config); this.checkpointManager = taskConfig.getCheckpointManager(metricsRegistry, classLoader).orElse(null); } - @VisibleForTesting - MetadataResourceUtil(CheckpointManager checkpointManager, JobModel jobModel) { - this.jobModel = jobModel; - this.checkpointManager = checkpointManager; - } - /** * Creates and loads the required metadata resources for checkpoints, changelog stream and other * resources related to the metadata system @@ -62,7 +59,13 @@ public void createResources() { createChangelogStreams(); } + @VisibleForTesting void createChangelogStreams() { - ChangelogStreamManager.createChangelogStreams(jobModel.getConfig(), jobModel.maxChangeLogStreamPartitions); + ChangelogStreamManager.createChangelogStreams(config, jobModel.maxChangeLogStreamPartitions); + } + + @VisibleForTesting + CheckpointManager getCheckpointManager() { + return checkpointManager; } } 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 b7d43f65d6..716b3671ef 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 @@ -86,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, getClass().getClassLoader()); + MetadataResourceUtil metadataResourceUtil = new MetadataResourceUtil(jobModel, null, getClass().getClassLoader(), config); metadataResourceUtil.createResources(); } catch (Exception e) { LOGGER.error("Exception while trying to getJobModel.", e); 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 b4a2481bd2..80bf92ec7b 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 @@ -313,7 +313,7 @@ JobModel readJobModelFromMetadataStore(String zkJobModelVersion) { @VisibleForTesting void loadMetadataResources(JobModel jobModel) { try { - MetadataResourceUtil metadataResourceUtil = createMetadataResourceUtil(jobModel, getClass().getClassLoader()); + MetadataResourceUtil metadataResourceUtil = createMetadataResourceUtil(jobModel, getClass().getClassLoader(), config); metadataResourceUtil.createResources(); if (coordinatorStreamStore != null) { @@ -343,8 +343,8 @@ void loadMetadataResources(JobModel jobModel) { } @VisibleForTesting - MetadataResourceUtil createMetadataResourceUtil(JobModel jobModel, ClassLoader classLoader) { - return new MetadataResourceUtil(jobModel, metrics.getMetricsRegistry(), classLoader); + MetadataResourceUtil createMetadataResourceUtil(JobModel jobModel, ClassLoader classLoader, Config config) { + return new MetadataResourceUtil(jobModel, metrics.getMetricsRegistry(), classLoader, config); } /** 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 b1cfbe6129..746c234bab 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 @@ -70,7 +70,7 @@ class ProcessJobFactory extends StreamJobFactory with Logging { changelogStreamManager.writePartitionMapping(taskPartitionMappings) //create necessary checkpoint and changelog streams - val metadataResourceUtil = new MetadataResourceUtil(jobModel, metricsRegistry, classLoader) + val metadataResourceUtil = new MetadataResourceUtil(jobModel, metricsRegistry, classLoader, config) metadataResourceUtil.createResources() // fan out the startpoints 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 f9116fefa2..fc8296b02f 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 @@ -71,7 +71,7 @@ class ThreadJobFactory extends StreamJobFactory with Logging { changelogStreamManager.writePartitionMapping(taskPartitionMappings) //create necessary checkpoint and changelog streams - val metadataResourceUtil = new MetadataResourceUtil(jobModel, metricsRegistry, classLoader) + val metadataResourceUtil = new MetadataResourceUtil(jobModel, metricsRegistry, classLoader, config) metadataResourceUtil.createResources() // fan out the startpoints diff --git a/samza-core/src/test/java/org/apache/samza/coordinator/TestMetadataResourceUtil.java b/samza-core/src/test/java/org/apache/samza/coordinator/TestMetadataResourceUtil.java index dc8670415e..7ff85e1246 100644 --- a/samza-core/src/test/java/org/apache/samza/coordinator/TestMetadataResourceUtil.java +++ b/samza-core/src/test/java/org/apache/samza/coordinator/TestMetadataResourceUtil.java @@ -19,30 +19,60 @@ package org.apache.samza.coordinator; +import com.google.common.collect.ImmutableMap; import org.apache.samza.checkpoint.CheckpointManager; +import org.apache.samza.checkpoint.CheckpointManagerFactory; +import org.apache.samza.config.Config; +import org.apache.samza.config.MapConfig; +import org.apache.samza.config.TaskConfig; import org.apache.samza.job.model.JobModel; +import org.apache.samza.metrics.MetricsRegistry; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; public class TestMetadataResourceUtil { - private CheckpointManager checkpointManager; - private JobModel jobModel; + private JobModel mockJobModel; + private MetricsRegistry mockMetricsRegistry; @Before public void setUp() { - checkpointManager = Mockito.mock(CheckpointManager.class); - jobModel = Mockito.mock(JobModel.class); + mockJobModel = Mockito.mock(JobModel.class); + mockMetricsRegistry = Mockito.mock(MetricsRegistry.class); } @Test - public void testLoad() { - MetadataResourceUtil metadataResourceUtil = Mockito.spy(new MetadataResourceUtil(checkpointManager, jobModel)); + public void testLoadWithCheckpointConfigured() { + MapConfig mapConfig = new MapConfig(ImmutableMap.of(TaskConfig.CHECKPOINT_MANAGER_FACTORY, + TestCheckpointManagerFactory.class.getName())); + MetadataResourceUtil metadataResourceUtil = Mockito.spy(new MetadataResourceUtil(mockJobModel, mockMetricsRegistry, getClass().getClassLoader(), mapConfig)); Mockito.doNothing().when(metadataResourceUtil).createChangelogStreams(); metadataResourceUtil.createResources(); - Mockito.verify(checkpointManager).createResources(); + Mockito.verify(mockJobModel, Mockito.never()).getConfig(); // Never get config from job model. In standalone, this is empty. + Mockito.verify(metadataResourceUtil.getCheckpointManager()).createResources(); Mockito.verify(metadataResourceUtil).createChangelogStreams(); } + + + @Test + public void testLoadWithoutCheckpointConfigured() { + MapConfig mapConfig = new MapConfig(); + MetadataResourceUtil metadataResourceUtil = Mockito.spy(new MetadataResourceUtil(mockJobModel, mockMetricsRegistry, getClass().getClassLoader(), mapConfig)); + Mockito.doNothing().when(metadataResourceUtil).createChangelogStreams(); + + metadataResourceUtil.createResources(); + Mockito.verify(mockJobModel, Mockito.never()).getConfig(); // Never get config from job model. In standalone, this is empty. + Assert.assertNull(metadataResourceUtil.getCheckpointManager()); + Mockito.verify(metadataResourceUtil).createChangelogStreams(); + } + + public static class TestCheckpointManagerFactory implements CheckpointManagerFactory { + @Override + public CheckpointManager getCheckpointManager(Config config, MetricsRegistry registry) { + return Mockito.mock(CheckpointManager.class); + } + } } \ No newline at end of file diff --git a/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java b/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java index 1b3ee12e92..3492035c1e 100644 --- a/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java +++ b/samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java @@ -272,7 +272,7 @@ public void testLoadMetadataResources() throws IOException { MetadataResourceUtil mockMetadataResourceUtil = mock(MetadataResourceUtil.class); doReturn(mockMetadataResourceUtil).when(zkJobCoordinator) - .createMetadataResourceUtil(any(), eq(getClass().getClassLoader())); + .createMetadataResourceUtil(any(), eq(getClass().getClassLoader()), any(Config.class)); verifyZeroInteractions(mockStartpointManager); From 858a7adc37ad260d3358e45e64d67f0975378cc9 Mon Sep 17 00:00:00 2001 From: Daniel Nishimura Date: Thu, 8 Aug 2019 13:29:14 -0700 Subject: [PATCH 2/2] Trigger