/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.samza.runtime;

import joptsimple.OptionSet;
import joptsimple.OptionSpec;
import org.apache.samza.config.Config;
import org.apache.samza.config.MapConfig;
import org.apache.samza.util.CommandLine;


/**
 * This class contains the main() method used by run-app.sh.
 * It creates the {@link ApplicationRunner} based on the config, and then run the application.
 */
public class ApplicationRunnerMain {

  public static class ApplicationRunnerCommandLine extends CommandLine {
    OptionSpec<String> operationOpt =
        parser().accepts("operation", "The operation to perform; run, status, kill.")
            .withRequiredArg()
            .ofType(String.class)
            .describedAs("operation=run")
            .defaultsTo("run");

    ApplicationRunnerOperation getOperation(OptionSet options) {
      String rawOp = options.valueOf(operationOpt);
      return ApplicationRunnerOperation.fromString(rawOp);
    }

    @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));
    }
  }

  public static void main(String[] args) {
    ApplicationRunnerCommandLine cmdLine = new ApplicationRunnerCommandLine();
    OptionSet options = cmdLine.parser().parse(args);
    Config orgConfig = cmdLine.loadConfig(options);
    ApplicationRunnerOperation op = cmdLine.getOperation(options);
    ApplicationRunnerUtil.invoke(orgConfig, op);
  }
}

@@ -837,16 +837,64 @@
 ntime;%0A%0A
+import java.util.HashMap;%0Aimport java.util.Map;%0A
 import j
@@ -1014,25 +1014,133 @@
 .config.
-Map
 Config
+Loader;%0Aimport org.apache.samza.config.MapConfig;%0Aimport org.apache.samza.config.loaders.PropertiesConfigLoader
 ;%0Aimport
@@ -1716,16 +1716,239 @@
 run%22);%0A%0A
+    OptionSpec%3CString%3E configPathOpt =%0A        parser().accepts(%22config-path%22, %22File path to submission properties file.%22)%0A            .withOptionalArg()%0A            .ofType(String.class)%0A            .describedAs(%22path%22);%0A%0A
     Appl
@@ -2196,155 +2196,361 @@
     
-// ConfigLoader is not supposed to be invoked to load full job config during job submission.%0A      return new MapConfig(getConfigOverrides(options)
+Map%3CString, String%3E submissionConfig = new HashMap%3C%3E();%0A%0A      if (options.has(configPathOpt)) %7B%0A        ConfigLoader loader = new PropertiesConfigLoader(options.valueOf(configPathOpt));%0A        submissionConfig.putAll(loader.getConfig());%0A      %7D%0A%0A      submissionConfig.putAll(getConfigOverrides(options));%0A%0A      return new MapConfig(submissionConfig
 );%0A 

Add submission config file support (#1336)