cameronlee314 commented on a change in pull request #1248: SAMZA-2410: Update 
ClusterBasedJobCoordinator config retrieval logic from loader.
URL: https://github.com/apache/samza/pull/1248#discussion_r367043742
 
 

 ##########
 File path: 
samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java
 ##########
 @@ -172,15 +179,39 @@
    * Creates a new ClusterBasedJobCoordinator instance from a config. Invoke 
run() to actually
    * run the jobcoordinator.
    *
-   * @param coordinatorSystemConfig the coordinator stream config that can be 
used to read the
-   *                                {@link 
org.apache.samza.job.model.JobModel} from.
+   * @param jobCoordinatorConfig job coordinator config that either contains 
coordinator stream properties
+   *                             or config loader properties to load full job 
config.
    */
-  public ClusterBasedJobCoordinator(Config coordinatorSystemConfig) {
+  public ClusterBasedJobCoordinator(Config jobCoordinatorConfig) {
     metrics = new MetricsRegistryMap();
 
-    coordinatorStreamStore = new 
CoordinatorStreamStore(coordinatorSystemConfig, metrics);
-    coordinatorStreamStore.init();
-    config = 
CoordinatorStreamUtil.readConfigFromCoordinatorStream(coordinatorStreamStore);
+    JobConfig jobConfig = new JobConfig(jobCoordinatorConfig);
+
+    if (jobConfig.getConfigLoaderFactory().isPresent()) {
+      // load full job config with ConfigLoader
+      Config originalConfig = ConfigUtil.loadConfig(jobCoordinatorConfig);
+
+      // Execute planning
+      ApplicationDescriptorImpl<? extends ApplicationDescriptor>
+          appDesc = 
ApplicationDescriptorUtil.getAppDescriptor(ApplicationUtil.fromConfig(originalConfig),
 originalConfig);
+      RemoteJobPlanner planner = new RemoteJobPlanner(appDesc);
+      List<JobConfig> jobConfigs = planner.prepareJobs();
+
+      if (jobConfigs.size() != 1) {
+        throw new SamzaException("Only support single remote job is 
supported.");
+      }
+
+      config = jobConfigs.get(0);
+      coordinatorStreamStore = new 
CoordinatorStreamStore(CoordinatorStreamUtil.buildCoordinatorStreamConfig(config),
 metrics);
+      coordinatorStreamStore.init();
+      CoordinatorStreamUtil.writeConfigToCoordinatorStream(config, true);
+      DiagnosticsUtil.createDiagnosticsStream(config);
 
 Review comment:
   Does this need to match the set-up done in `JobRunner.run`? Maybe it would 
be good to put a comment in here and in `JobRunner` to make sure they are 
consistent. I know it will be cleaned up eventually, but there will be some 
time in which both flows exist.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

Reply via email to