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_r368068551
##########
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) {
+ MetricsRegistryMap metrics = new MetricsRegistryMap();
+
+ CoordinatorStreamStore coordinatorStreamStore = new
CoordinatorStreamStore(metadataStoreConfig, metrics);
+ coordinatorStreamStore.init();
+ Config config =
CoordinatorStreamUtil.readConfigFromCoordinatorStream(coordinatorStreamStore);
+
+ return new ClusterBasedJobCoordinator(metrics, coordinatorStreamStore,
config);
+ }
+
+ /**
+ * Initialize {@link ClusterBasedJobCoordinator} with submission config,
full job config will be fetched using
+ * specified {@link org.apache.samza.config.ConfigLoaderFactory}
+ *
+ * @param submissionConfig specifies {@link
org.apache.samza.config.ConfigLoaderFactory}
+ * @return {@link ClusterBasedJobCoordinator}
+ */
+ public static ClusterBasedJobCoordinator createFromConfigLoader(Config
submissionConfig) {
Review comment:
Does this need to be public? If you need in tests, could you make it package
private and mark as `VisibleForTesting`?
----------------------------------------------------------------
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
