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_r368077678
##########
File path:
samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java
##########
@@ -480,20 +485,101 @@ private static void
executeRunClusterBasedJobCoordinatorForClass(Class<?> cluste
* {@link #main(String[])} so that it can be executed directly or from a
separate classloader.
*/
private static void runClusterBasedJobCoordinator(String[] args) {
- Config coordinatorSystemConfig;
final String coordinatorSystemEnv =
System.getenv(ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG());
- 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);
+ final String submissionEnv =
System.getenv(ShellCommandConfig.ENV_SUBMISSION_CONFIG());
+
+ if (submissionEnv != null) {
+ 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);
+ }
+
+ ClusterBasedJobCoordinator jc = createFromConfigLoader(submissionConfig);
+ jc.run();
+ 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.
+ public static ClusterBasedJobCoordinator createFromMetadataStore(Config
metadataStoreConfig) {
Review comment:
Update to private.
----------------------------------------------------------------
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
