SAMZA-2296: Metadata streams not created in Standalone

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

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;
   }
 }

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);

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);
   }
 
   /**

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

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

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);
+    }
+  }
 }

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);