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:
[email protected]
With regards,
Apache Git Services
