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 68e2f77..6295ed6 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
@@ -19,29 +19,19 @@
 package org.apache.samza.clustermanager;
 
 import com.google.common.annotations.VisibleForTesting;
-import java.io.IOException;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.SamzaException;
 import org.apache.samza.clustermanager.container.placement.ContainerPlacementMetadataStore;
-import org.apache.samza.application.ApplicationUtil;
-import org.apache.samza.classloader.IsolatingClassLoaderFactory;
 import org.apache.samza.clustermanager.container.placement.ContainerPlacementRequestAllocator;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.ClusterManagerConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
-import org.apache.samza.config.MapConfig;
-import org.apache.samza.config.ShellCommandConfig;
 import org.apache.samza.config.StorageConfig;
 import org.apache.samza.config.TaskConfig;
 import org.apache.samza.container.LocalityManager;
@@ -52,7 +42,6 @@
 import org.apache.samza.coordinator.PartitionChangeException;
 import org.apache.samza.coordinator.StreamPartitionCountMonitor;
 import org.apache.samza.coordinator.StreamRegexMonitor;
-import org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore;
 import org.apache.samza.coordinator.metadatastore.NamespaceAwareCoordinatorStreamStore;
 import org.apache.samza.coordinator.stream.messages.SetChangelogMapping;
 import org.apache.samza.coordinator.stream.messages.SetContainerHostMapping;
@@ -63,16 +52,12 @@
 import org.apache.samza.metadatastore.MetadataStore;
 import org.apache.samza.metrics.JmxServer;
 import org.apache.samza.metrics.MetricsRegistryMap;
-import org.apache.samza.serializers.model.SamzaObjectMapper;
 import org.apache.samza.startpoint.StartpointManager;
 import org.apache.samza.storage.ChangelogStreamManager;
 import org.apache.samza.system.StreamMetadataCache;
 import org.apache.samza.system.SystemAdmins;
 import org.apache.samza.system.SystemStream;
-import org.apache.samza.util.ConfigUtil;
-import org.apache.samza.util.CoordinatorStreamUtil;
 import org.apache.samza.util.DiagnosticsUtil;
-import org.apache.samza.util.SplitDeploymentUtil;
 import org.apache.samza.util.SystemClock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -465,142 +450,4 @@
   ContainerProcessManager createContainerProcessManager() {
     return new ContainerProcessManager(config, state, metrics, containerPlacementMetadataStore, localityManager);
   }
-
-  /**
-   * The entry point for the {@link ClusterBasedJobCoordinator}.
-   */
-  public static void main(String[] args) {
-    Thread.setDefaultUncaughtExceptionHandler((thread, exception) -> {
-      LOG.error("Uncaught exception in ClusterBasedJobCoordinator::main. Exiting job coordinator", exception);
-      System.exit(1);
-    });
-    if (!SplitDeploymentUtil.isSplitDeploymentEnabled()) {
-      // no isolation enabled, so can just execute runClusterBasedJobCoordinator directly
-      runClusterBasedJobCoordinator(args);
-    } else {
-      SplitDeploymentUtil.runWithClassLoader(new IsolatingClassLoaderFactory().buildClassLoader(),
-          ClusterBasedJobCoordinator.class, "runClusterBasedJobCoordinator", args);
-    }
-    System.exit(0);
-  }
-
-  /**
-   * This is the actual execution for the {@link ClusterBasedJobCoordinator}. This is separated out from
-   * {@link #main(String[])} so that it can be executed directly or from a separate classloader.
-   */
-  private static void runClusterBasedJobCoordinator(String[] args) {
-    final String coordinatorSystemEnv = System.getenv(ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG);
-    final String submissionEnv = System.getenv(ShellCommandConfig.ENV_SUBMISSION_CONFIG);
-
-    if (!StringUtils.isBlank(submissionEnv)) {
-      Config submissionConfig;
-      try {
-        //Read and parse the coordinator system config.
-        LOG.info("Parsing submission config {}", submissionEnv);
-        submissionConfig =
-            new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(submissionEnv, Config.class));
-        LOG.info("Using the submission config: {}.", submissionConfig);
-      } catch (IOException e) {
-        LOG.error("Exception while reading submission config", e);
-        throw new SamzaException(e);
-      }
-
-      ApplicationConfig appConfig = new ApplicationConfig(submissionConfig);
-
-      /*
-       * Invoke app.main.class with app.main.args when present.
-       * For Beam jobs, app.main.class will be Beam's main class
-       * and app.main.args will be Beam's pipeline options.
-       */
-      if (appConfig.getAppMainClass().isPresent()) {
-        String className = appConfig.getAppMainClass().get();
-        LOG.info("Invoke main {}", className);
-        try {
-          Class<?> cls = Class.forName(className);
-          Method mainMethod = cls.getMethod("main", String[].class);
-          mainMethod.invoke(null, (Object) toArgs(appConfig));
-        } catch (Exception e) {
-          throw new SamzaException(e);
-        }
-      } else {
-        JobConfig jobConfig = new JobConfig(submissionConfig);
-
-        if (!jobConfig.getConfigLoaderFactory().isPresent()) {
-          throw new SamzaException(JobConfig.CONFIG_LOADER_FACTORY + " is required to initialize job coordinator from config loader");
-        }
-
-        // load full job config with ConfigLoader
-        Config originalConfig = ConfigUtil.loadConfig(submissionConfig);
-
-        JobCoordinatorLaunchUtil.run(ApplicationUtil.fromConfig(originalConfig), originalConfig);
-      }
-
-      LOG.info("Finished running ClusterBasedJobCoordinator");
-    } else {
-      // TODO: Clean this up once SAMZA-2405 is completed when legacy flow is removed.
-      Config coordinatorSystemConfig;
-      try {
-        //Read and parse the coordinator system config.
-        LOG.info("Parsing coordinator system config {}", coordinatorSystemEnv);
-        coordinatorSystemConfig =
-            new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(coordinatorSystemEnv, Config.class));
-        LOG.info("Using the coordinator system config: {}.", coordinatorSystemConfig);
-      } catch (IOException e) {
-        LOG.error("Exception while reading coordinator stream config", e);
-        throw new SamzaException(e);
-      }
-      ClusterBasedJobCoordinator jc = createFromMetadataStore(coordinatorSystemConfig);
-      jc.run();
-      LOG.info("Finished running ClusterBasedJobCoordinator");
-    }
-  }
-
-  /**
-   * Initialize {@link ClusterBasedJobCoordinator} with coordinator stream config, full job config will be fetched from
-   * coordinator stream.
-   *
-   * @param metadataStoreConfig to initialize {@link MetadataStore}
-   * @return {@link ClusterBasedJobCoordinator}
-   */
-  // TODO SAMZA-2432: Clean this up once SAMZA-2405 is completed when legacy flow is removed.
-  @VisibleForTesting
-  static ClusterBasedJobCoordinator createFromMetadataStore(Config metadataStoreConfig) {
-    MetricsRegistryMap metrics = new MetricsRegistryMap();
-
-    CoordinatorStreamStore coordinatorStreamStore = new CoordinatorStreamStore(metadataStoreConfig, metrics);
-    coordinatorStreamStore.init();
-    Config config = CoordinatorStreamUtil.readConfigFromCoordinatorStream(coordinatorStreamStore);
-
-    return new ClusterBasedJobCoordinator(metrics, coordinatorStreamStore, config);
-  }
-
-  /**
-   * Convert Samza config to command line arguments to invoke app.main.class
-   *
-   * @param config Samza config to convert.
-   * @return converted command line arguments.
-   */
-  @VisibleForTesting
-  static String[] toArgs(ApplicationConfig config) {
-    List<String> args = new ArrayList<>(config.size() * 2);
-
-    config.forEach((key, value) -> {
-      if (key.equals(ApplicationConfig.APP_MAIN_ARGS)) {
-        /*
-         * Converts native beam pipeline options such as
-         * --runner=SamzaRunner --maxSourceParallelism=1024
-         */
-        args.addAll(Arrays.asList(value.split("\\s")));
-      } else {
-        /*
-         * Converts native Samza configs to config override format such as
-         * --config job.name=test
-         */
-        args.add("--config");
-        args.add(String.format("%s=%s", key, value));
-      }
-    });
-
-    return args.toArray(new String[0]);
-  }
 }
diff --git a/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java b/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java
new file mode 100644
index 0000000..107c473
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinatorRunner.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.clustermanager;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.SamzaException;
+import org.apache.samza.application.ApplicationUtil;
+import org.apache.samza.classloader.IsolatingClassLoaderFactory;
+import org.apache.samza.config.ApplicationConfig;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.ShellCommandConfig;
+import org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore;
+import org.apache.samza.metrics.MetricsRegistryMap;
+import org.apache.samza.serializers.model.SamzaObjectMapper;
+import org.apache.samza.util.ConfigUtil;
+import org.apache.samza.util.CoordinatorStreamUtil;
+import org.apache.samza.util.SplitDeploymentUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class ClusterBasedJobCoordinatorRunner {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ClusterBasedJobCoordinatorRunner.class);
+
+  /**
+   * The entry point for the {@link ClusterBasedJobCoordinator}.
+   */
+  public static void main(String[] args) {
+    Thread.setDefaultUncaughtExceptionHandler((thread, exception) -> {
+      LOG.error("Uncaught exception in ClusterBasedJobCoordinator::main. Exiting job coordinator", exception);
+      System.exit(1);
+    });
+    if (!SplitDeploymentUtil.isSplitDeploymentEnabled()) {
+      // no isolation enabled, so can just execute runClusterBasedJobCoordinator directly
+      runClusterBasedJobCoordinator(args);
+    } else {
+      SplitDeploymentUtil.runWithClassLoader(new IsolatingClassLoaderFactory().buildClassLoader(),
+          ClusterBasedJobCoordinatorRunner.class, "runClusterBasedJobCoordinator", args);
+    }
+    System.exit(0);
+  }
+
+  /**
+   * This is the actual execution for the {@link ClusterBasedJobCoordinator}. This is separated out from
+   * {@link #main(String[])} so that it can be executed directly or from a separate classloader.
+   */
+  @VisibleForTesting
+  static void runClusterBasedJobCoordinator(String[] args) {
+    final String coordinatorSystemEnv = System.getenv(ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG);
+    final String submissionEnv = System.getenv(ShellCommandConfig.ENV_SUBMISSION_CONFIG);
+
+    if (!StringUtils.isBlank(submissionEnv)) {
+      Config submissionConfig;
+      try {
+        //Read and parse the coordinator system config.
+        LOG.info("Parsing submission config {}", submissionEnv);
+        submissionConfig =
+            new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(submissionEnv, Config.class));
+        LOG.info("Using the submission config: {}.", submissionConfig);
+      } catch (IOException e) {
+        LOG.error("Exception while reading submission config", e);
+        throw new SamzaException(e);
+      }
+
+      ApplicationConfig appConfig = new ApplicationConfig(submissionConfig);
+
+      /*
+       * Invoke app.main.class with app.main.args when present.
+       * For Beam jobs, app.main.class will be Beam's main class
+       * and app.main.args will be Beam's pipeline options.
+       */
+      if (appConfig.getAppMainClass().isPresent()) {
+        String className = appConfig.getAppMainClass().get();
+        LOG.info("Invoke main {}", className);
+        try {
+          Class<?> cls = Class.forName(className);
+          Method mainMethod = cls.getMethod("main", String[].class);
+          mainMethod.invoke(null, (Object) toArgs(appConfig));
+        } catch (Exception e) {
+          throw new SamzaException(e);
+        }
+      } else {
+        JobConfig jobConfig = new JobConfig(submissionConfig);
+
+        if (!jobConfig.getConfigLoaderFactory().isPresent()) {
+          throw new SamzaException(JobConfig.CONFIG_LOADER_FACTORY + " is required to initialize job coordinator from config loader");
+        }
+
+        // load full job config with ConfigLoader
+        Config originalConfig = ConfigUtil.loadConfig(submissionConfig);
+
+        JobCoordinatorLaunchUtil.run(ApplicationUtil.fromConfig(originalConfig), originalConfig);
+      }
+
+      LOG.info("Finished running ClusterBasedJobCoordinator");
+    } else {
+      // TODO: Clean this up once SAMZA-2405 is completed when legacy flow is removed.
+      Config coordinatorSystemConfig;
+      try {
+        //Read and parse the coordinator system config.
+        LOG.info("Parsing coordinator system config {}", coordinatorSystemEnv);
+        coordinatorSystemConfig =
+            new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(coordinatorSystemEnv, Config.class));
+        LOG.info("Using the coordinator system config: {}.", coordinatorSystemConfig);
+      } catch (IOException e) {
+        LOG.error("Exception while reading coordinator stream config", e);
+        throw new SamzaException(e);
+      }
+      ClusterBasedJobCoordinator jc = createFromMetadataStore(coordinatorSystemConfig);
+      jc.run();
+      LOG.info("Finished running ClusterBasedJobCoordinator");
+    }
+  }
+
+  /**
+   * Initialize {@link ClusterBasedJobCoordinator} with coordinator stream config, full job config will be fetched from
+   * coordinator stream.
+   *
+   * @param metadataStoreConfig to initialize {@link org.apache.samza.metadatastore.MetadataStore}
+   * @return {@link ClusterBasedJobCoordinator}
+   */
+  // TODO SAMZA-2432: Clean this up once SAMZA-2405 is completed when legacy flow is removed.
+  @VisibleForTesting
+  static ClusterBasedJobCoordinator createFromMetadataStore(Config metadataStoreConfig) {
+    MetricsRegistryMap metrics = new MetricsRegistryMap();
+
+    CoordinatorStreamStore coordinatorStreamStore = new CoordinatorStreamStore(metadataStoreConfig, metrics);
+    coordinatorStreamStore.init();
+    Config config = CoordinatorStreamUtil.readConfigFromCoordinatorStream(coordinatorStreamStore);
+
+    return new ClusterBasedJobCoordinator(metrics, coordinatorStreamStore, config);
+  }
+
+  /**
+   * Convert Samza config to command line arguments to invoke app.main.class
+   *
+   * @param config Samza config to convert.
+   * @return converted command line arguments.
+   */
+  @VisibleForTesting
+  static String[] toArgs(ApplicationConfig config) {
+    List<String> args = new ArrayList<>(config.size() * 2);
+
+    config.forEach((key, value) -> {
+      if (key.equals(ApplicationConfig.APP_MAIN_ARGS)) {
+        /*
+         * Converts native beam pipeline options such as
+         * --runner=SamzaRunner --maxSourceParallelism=1024
+         */
+        args.addAll(Arrays.asList(value.split("\\s")));
+      } else {
+        /*
+         * Converts native Samza configs to config override format such as
+         * --config job.name=test
+         */
+        args.add("--config");
+        args.add(String.format("%s=%s", key, value));
+      }
+    });
+
+    return args.toArray(new String[0]);
+  }
+}
diff --git a/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java b/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
index 6444451..caa1ffe 100644
--- a/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
+++ b/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinator.java
@@ -69,7 +69,7 @@
 @RunWith(PowerMockRunner.class)
 @PrepareForTest({
     CoordinatorStreamUtil.class,
-    ClusterBasedJobCoordinator.class,
+    ClusterBasedJobCoordinatorRunner.class,
     CoordinatorStreamStore.class,
     RemoteJobPlanner.class})
 public class TestClusterBasedJobCoordinator {
@@ -112,7 +112,7 @@
     CoordinatorStreamSystemProducer producer = new CoordinatorStreamSystemProducer(config, mock(MetricsRegistry.class));
     producer.writeConfig("test-job", config);
 
-    ClusterBasedJobCoordinator clusterCoordinator = ClusterBasedJobCoordinator.createFromMetadataStore(config);
+    ClusterBasedJobCoordinator clusterCoordinator = ClusterBasedJobCoordinatorRunner.createFromMetadataStore(config);
 
     // change the input system stream metadata
     MockSystemFactory.MSG_QUEUES.put(new SystemStreamPartition("kafka", "topic1", new Partition(1)), new ArrayList<>());
@@ -132,7 +132,7 @@
     CoordinatorStreamSystemProducer producer = new CoordinatorStreamSystemProducer(config, mock(MetricsRegistry.class));
     producer.writeConfig("test-job", config);
 
-    ClusterBasedJobCoordinator clusterCoordinator = ClusterBasedJobCoordinator.createFromMetadataStore(config);
+    ClusterBasedJobCoordinator clusterCoordinator = ClusterBasedJobCoordinatorRunner.createFromMetadataStore(config);
 
     // change the input system stream metadata
     MockSystemFactory.MSG_QUEUES.put(new SystemStreamPartition("kafka", "topic1", new Partition(1)), new ArrayList<>());
@@ -150,7 +150,7 @@
     Config config = new MapConfig(configMap);
     MockitoException stopException = new MockitoException("Stop");
 
-    ClusterBasedJobCoordinator clusterCoordinator = Mockito.spy(ClusterBasedJobCoordinator.createFromMetadataStore(config));
+    ClusterBasedJobCoordinator clusterCoordinator = Mockito.spy(ClusterBasedJobCoordinatorRunner.createFromMetadataStore(config));
     ContainerProcessManager mockContainerProcessManager = mock(ContainerProcessManager.class);
     doReturn(true).when(mockContainerProcessManager).shouldShutdown();
     StartpointManager mockStartpointManager = mock(StartpointManager.class);
@@ -186,7 +186,7 @@
         "--config", "app.class=class1",
         "--runner=SamzaRunner",
         "--maxSourceParallelism=1024");
-    List<String> actual = Arrays.asList(ClusterBasedJobCoordinator.toArgs(appConfig));
+    List<String> actual = Arrays.asList(ClusterBasedJobCoordinatorRunner.toArgs(appConfig));
 
     // cannot assert expected equals to actual as the order can be different.
     assertEquals(expected.size(), actual.size());
diff --git a/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinatorRunner.java b/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinatorRunner.java
new file mode 100644
index 0000000..548cb4a
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/clustermanager/TestClusterBasedJobCoordinatorRunner.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.clustermanager;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.samza.application.ApplicationUtil;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.ShellCommandConfig;
+import org.apache.samza.config.loaders.PropertiesConfigLoaderFactory;
+import org.apache.samza.serializers.model.SamzaObjectMapper;
+import org.apache.samza.util.ConfigUtil;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
+import static org.powermock.api.mockito.PowerMockito.mock;
+
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({
+    System.class,
+    ClusterBasedJobCoordinatorRunner.class,
+    ApplicationUtil.class,
+    JobCoordinatorLaunchUtil.class})
+public class TestClusterBasedJobCoordinatorRunner {
+
+  @Test
+  public void testRunClusterBasedJobCoordinator() throws Exception  {
+    Config submissionConfig = new MapConfig(ImmutableMap.of(
+        JobConfig.CONFIG_LOADER_FACTORY,
+        PropertiesConfigLoaderFactory.class.getName(),
+        PropertiesConfigLoaderFactory.CONFIG_LOADER_PROPERTIES_PREFIX + "path",
+        getClass().getResource("/test.properties").getPath()));
+    Config fullConfig = ConfigUtil.loadConfig(submissionConfig);
+    StreamApplication mockApplication = mock(StreamApplication.class);
+    PowerMockito.mockStatic(System.class, ApplicationUtil.class, JobCoordinatorLaunchUtil.class);
+    PowerMockito
+        .when(System.getenv(eq(ShellCommandConfig.ENV_SUBMISSION_CONFIG)))
+        .thenReturn(SamzaObjectMapper.getObjectMapper().writeValueAsString(submissionConfig));
+    PowerMockito
+        .when(ApplicationUtil.fromConfig(any()))
+        .thenReturn(mockApplication);
+    PowerMockito.doNothing().when(JobCoordinatorLaunchUtil.class, "run", mockApplication, fullConfig);
+
+    ClusterBasedJobCoordinatorRunner.runClusterBasedJobCoordinator(null);
+
+    PowerMockito.verifyStatic(times(1));
+    JobCoordinatorLaunchUtil.run(mockApplication, fullConfig);
+  }
+}
diff --git a/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java b/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java
index d1dd8f8..72772ba 100644
--- a/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java
+++ b/samza-core/src/test/java/org/apache/samza/util/TestSplitDeploymentUtil.java
@@ -18,7 +18,7 @@
  */
 package org.apache.samza.util;
 
-import org.apache.samza.clustermanager.ClusterBasedJobCoordinator;
+import org.apache.samza.clustermanager.ClusterBasedJobCoordinatorRunner;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.powermock.api.mockito.PowerMockito;
@@ -33,18 +33,18 @@
 
 
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({ClusterBasedJobCoordinator.class})
+@PrepareForTest({ClusterBasedJobCoordinatorRunner.class})
 public class TestSplitDeploymentUtil {
 
   @Test
   public void testRunWithIsolatingClassLoader() throws Exception {
     // partially mock ClusterBasedJobCoordinator (mock runClusterBasedJobCoordinator method only)
-    PowerMockito.spy(ClusterBasedJobCoordinator.class);
+    PowerMockito.spy(ClusterBasedJobCoordinatorRunner.class);
     // save the context classloader to make sure that it gets set properly once the test is finished
     ClassLoader previousContextClassLoader = Thread.currentThread().getContextClassLoader();
     ClassLoader classLoader = mock(ClassLoader.class);
     String[] args = new String[]{"arg0", "arg1"};
-    doReturn(ClusterBasedJobCoordinator.class).when(classLoader).loadClass(ClusterBasedJobCoordinator.class.getName());
+    doReturn(ClusterBasedJobCoordinatorRunner.class).when(classLoader).loadClass(ClusterBasedJobCoordinatorRunner.class.getName());
 
     // stub the private static method which is called by reflection
     PowerMockito.doAnswer(invocation -> {
@@ -53,19 +53,19 @@
       // checks that the context classloader is set correctly
       assertEquals(classLoader, Thread.currentThread().getContextClassLoader());
       return null;
-    }).when(ClusterBasedJobCoordinator.class, "runClusterBasedJobCoordinator", any());
+    }).when(ClusterBasedJobCoordinatorRunner.class, "runClusterBasedJobCoordinator", any());
 
     try {
       SplitDeploymentUtil.runWithClassLoader(classLoader,
-          ClusterBasedJobCoordinator.class, "runClusterBasedJobCoordinator", args);
+          ClusterBasedJobCoordinatorRunner.class, "runClusterBasedJobCoordinator", args);
       assertEquals(previousContextClassLoader, Thread.currentThread().getContextClassLoader());
     } finally {
       // reset it explicitly just in case runWithClassLoader throws an exception
       Thread.currentThread().setContextClassLoader(previousContextClassLoader);
     }
     // make sure that the classloader got used
-    verify(classLoader).loadClass(ClusterBasedJobCoordinator.class.getName());
+    verify(classLoader).loadClass(ClusterBasedJobCoordinatorRunner.class.getName());
     // make sure runClusterBasedJobCoordinator only got called once
-    verifyPrivate(ClusterBasedJobCoordinator.class).invoke("runClusterBasedJobCoordinator", new Object[]{aryEq(args)});
+    verifyPrivate(ClusterBasedJobCoordinatorRunner.class).invoke("runClusterBasedJobCoordinator", new Object[]{aryEq(args)});
   }
 }
diff --git a/samza-shell/src/main/bash/run-jc.sh b/samza-shell/src/main/bash/run-jc.sh
index cf3e734..fc8a3c4 100755
--- a/samza-shell/src/main/bash/run-jc.sh
+++ b/samza-shell/src/main/bash/run-jc.sh
@@ -22,4 +22,4 @@
 # Set container name system properties for use in Log4J
 [[ $JAVA_OPTS != *-Dsamza.container.name* ]] && export JAVA_OPTS="$JAVA_OPTS -Dsamza.container.name=samza-job-coordinator"
 
-exec $(dirname $0)/run-class.sh org.apache.samza.clustermanager.ClusterBasedJobCoordinator "$@"
+exec $(dirname $0)/run-class.sh org.apache.samza.clustermanager.ClusterBasedJobCoordinatorRunner "$@"