kw2542 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_r367138308
##########
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:
Added a comment
----------------------------------------------------------------
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
