commit a086ae0a88b114109b8f3da6d487009505132c82
author Ke Wu <ke.wu.cs@icloud.com> Thu Apr 02 14:13:02 2020 -0700
committer GitHub <noreply@github.com> Thu Apr 02 14:13:02 2020 -0700
tree 2a842ee58b11f6ba53f062b32bec3dbe229e5c28
parent 26d9eecac76f3b87610312d0066d96a5b36eef35

SAMZA-2499: Add submission config file support (#1336)

API Changes:
1. Add --config-path option support when running run-app.sh

Upgrade Instructions:
None

Usage Instructions:
Alternative way when submitting job,

deploy/samza/bin/run-app.sh \
  --config app.class=A \
  --config job.stream.factory=B \
  --config yarn.package.path=C \
  --config job.config.loader.factory=D \
  --config job.config.loader.properties.xx=E \
  --config job.name=F

can be simplied to

deploy/samza/bin/run-app.sh --config-path=/path/to/submission/properties/file/submission.properties

where submission.properties contains

app.class=A
job.stream.factory=B
yarn.package.path=C
job.config.loader.factory=D
job.config.loader.properties.xx=E
job.name=F

samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java

diff --git a/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java b/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
index 220b68b..b917d4d 100644
--- a/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
+++ b/samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java
@@ -19,10 +19,14 @@
 
 package org.apache.samza.runtime;
 
+import java.util.HashMap;
+import java.util.Map;
 import joptsimple.OptionSet;
 import joptsimple.OptionSpec;
 import org.apache.samza.config.Config;
+import org.apache.samza.config.ConfigLoader;
 import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.loaders.PropertiesConfigLoader;
 import org.apache.samza.util.CommandLine;
 
 
@@ -40,6 +44,12 @@
             .describedAs("operation=run")
             .defaultsTo("run");
 
+    OptionSpec<String> configPathOpt =
+        parser().accepts("config-path", "File path to submission properties file.")
+            .withOptionalArg()
+            .ofType(String.class)
+            .describedAs("path");
+
     ApplicationRunnerOperation getOperation(OptionSet options) {
       String rawOp = options.valueOf(operationOpt);
       return ApplicationRunnerOperation.fromString(rawOp);
@@ -47,8 +57,16 @@
 
     @Override
     public Config loadConfig(OptionSet options) {
-      // ConfigLoader is not supposed to be invoked to load full job config during job submission.
-      return new MapConfig(getConfigOverrides(options));
+      Map<String, String> submissionConfig = new HashMap<>();
+
+      if (options.has(configPathOpt)) {
+        ConfigLoader loader = new PropertiesConfigLoader(options.valueOf(configPathOpt));
+        submissionConfig.putAll(loader.getConfig());
+      }
+
+      submissionConfig.putAll(getConfigOverrides(options));
+
+      return new MapConfig(submissionConfig);
     }
   }