SAMZA-2527:MetadataStore is not initialized before reading in JobCoordinatorLaunchUtil (#1365)
Symtom: Job failed to launch.
Cause: Metadata store is not initialized before invoking readLaunchConfigFromCoordinatorStream()
Changes: Init metadata store before invoking readLaunchConfigFromCoordinatorStream
Tests: None
API Changes: None
Upgrade Instructions: None
Usage Instructions: None
Co-authored-by: Ke Wu <kwu@linkedin.com>
diff --git a/samza-core/src/main/java/org/apache/samza/clustermanager/JobCoordinatorLaunchUtil.java b/samza-core/src/main/java/org/apache/samza/clustermanager/JobCoordinatorLaunchUtil.java
index 6915614..c943c0c 100644
--- a/samza-core/src/main/java/org/apache/samza/clustermanager/JobCoordinatorLaunchUtil.java
+++ b/samza-core/src/main/java/org/apache/samza/clustermanager/JobCoordinatorLaunchUtil.java
@@ -62,6 +62,10 @@
MetricsRegistryMap metrics = new MetricsRegistryMap();
MetadataStore
metadataStore = new CoordinatorStreamStore(CoordinatorStreamUtil.buildCoordinatorStreamConfig(fullConfig), metrics);
+ // MetadataStore will be closed in ClusterBasedJobCoordinator#onShutDown
+ // initialization of MetadataStore can be moved to ClusterBasedJobCoordinator after we clean up
+ // ClusterBasedJobCoordinator#createFromMetadataStore
+ metadataStore.init();
// Reads extra launch config from metadata store.
Config launchConfig = CoordinatorStreamUtil.readLaunchConfigFromCoordinatorStream(fullConfig, metadataStore);
Config finalConfig = new MapConfig(launchConfig, fullConfig);
@@ -69,10 +73,6 @@
// This needs to be consistent with RemoteApplicationRunner#run where JobRunner#submit to be called instead of JobRunner#run
CoordinatorStreamUtil.writeConfigToCoordinatorStream(finalConfig, true);
DiagnosticsUtil.createDiagnosticsStream(finalConfig);
- // MetadataStore will be closed in ClusterBasedJobCoordinator#onShutDown
- // initialization of MetadataStore can be moved to ClusterBasedJobCoordinator after we clean up
- // ClusterBasedJobCoordinator#createFromMetadataStore
- metadataStore.init();
ClusterBasedJobCoordinator jc = new ClusterBasedJobCoordinator(
metrics,
