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_r368067913
##########
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();
Review comment:
It looks like you already call `init` in the shared constructor. Please call
it only once.
----------------------------------------------------------------
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
