Merge pull request #636 from kadaan/AllowConfigSpecificationForStateS…

…toreChecker Allow configuration file to be specified when running statestore-checker
apache · Feb 1, 2016 · bd34294 · bd34294
2 parents 8937534 + 14b2981
commit bd34294
Show file tree

Hide file tree

Showing 3 changed files with 45 additions and 20 deletions.
diff --git a/gobblin-runtime/src/main/java/gobblin/runtime/cli/CliOptions.java b/gobblin-runtime/src/main/java/gobblin/runtime/cli/CliOptions.java
@@ -21,13 +21,7 @@
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
-
-import org.apache.commons.configuration.ConfigurationConverter;
 import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.PropertiesConfiguration;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
 
 import gobblin.util.JobConfigurationUtils;
 
@@ -62,7 +56,6 @@ public class CliOptions {
    * @throws IOException
    */
   public static Properties parseArgs(Class<?> caller, String[] args) throws IOException {
-
     try {
       // Parse command-line options
       CommandLine cmd = new DefaultParser().parse(options(), args);
@@ -78,10 +71,8 @@ public static Properties parseArgs(Class<?> caller, String[] args) throws IOExce
       }
 
       // Load system and job configuration properties
-      Properties sysConfig = cmd.hasOption(SYS_CONFIG_OPTION.getLongOpt()) ?
-          fileToProperties(cmd.getOptionValue(SYS_CONFIG_OPTION.getLongOpt())) : new Properties();
-      Properties jobConfig = cmd.hasOption(JOB_CONFIG_OPTION.getLongOpt()) ?
-          fileToProperties(cmd.getOptionValue(JOB_CONFIG_OPTION.getLongOpt())) : new Properties();
+      Properties sysConfig = JobConfigurationUtils.fileToProperties(cmd.getOptionValue(SYS_CONFIG_OPTION.getLongOpt()));
+      Properties jobConfig = JobConfigurationUtils.fileToProperties(cmd.getOptionValue(JOB_CONFIG_OPTION.getLongOpt()));
 
       return JobConfigurationUtils.combineSysAndJobProperties(sysConfig, jobConfig);
     } catch (ParseException pe) {
@@ -107,10 +98,4 @@ private static Options options() {
     return options;
   }
 
-  private static Properties fileToProperties(String fileName) throws IOException, ConfigurationException {
-    Path filePath = new Path(fileName);
-    PropertiesConfiguration propsConfig = new PropertiesConfiguration();
-    propsConfig.load(filePath.getFileSystem(new Configuration()).open(filePath));
-    return ConfigurationConverter.getProperties(propsConfig);
-  }
 }
diff --git a/gobblin-runtime/src/main/java/gobblin/runtime/util/JobStateToJsonConverter.java b/gobblin-runtime/src/main/java/gobblin/runtime/util/JobStateToJsonConverter.java
@@ -19,7 +19,9 @@
 import java.io.StringWriter;
 import java.io.Writer;
 import java.util.List;
+import java.util.Properties;
 
+import gobblin.util.JobConfigurationUtils;
 import org.apache.commons.cli.BasicParser;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
@@ -28,6 +30,9 @@
 import org.apache.commons.cli.OptionBuilder;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,9 +60,14 @@ public class JobStateToJsonConverter {
   private final StateStore<? extends JobState> jobStateStore;
   private final boolean keepConfig;
 
-  public JobStateToJsonConverter(String storeUrl, boolean keepConfig)
+  public JobStateToJsonConverter(Properties props, String storeUrl, boolean keepConfig)
       throws IOException {
-    this.jobStateStore = new FsDatasetStateStore(storeUrl);
+    Configuration conf = new Configuration();
+    JobConfigurationUtils.putPropertiesIntoConfiguration(props, conf);
+    Path storePath = new Path(storeUrl);
+    FileSystem fs = storePath.getFileSystem(conf);
+    String storeRootDir = storePath.toUri().getPath();
+    this.jobStateStore = new FsDatasetStateStore(fs, storeRootDir);
     this.keepConfig = keepConfig;
   }
 
@@ -156,6 +166,12 @@ private void writeJobStates(JsonWriter jsonWriter, List<? extends JobState> jobS
   @SuppressWarnings("all")
   public static void main(String[] args)
       throws Exception {
+    Option sysConfigOption = OptionBuilder
+        .withArgName("system configuration file")
+        .withDescription("Gobblin system configuration file")
+        .withLongOpt("sysconfig")
+        .hasArgs()
+        .create("sc");
     Option storeUrlOption = OptionBuilder
         .withArgName("gobblin state store URL")
         .withDescription("Gobblin state store root path URL")
@@ -192,6 +208,7 @@ public static void main(String[] args)
         .create("t");
 
     Options options = new Options();
+    options.addOption(sysConfigOption);
     options.addOption(storeUrlOption);
     options.addOption(jobNameOption);
     options.addOption(jobIdOption);
@@ -209,7 +226,12 @@ public static void main(String[] args)
       System.exit(1);
     }
 
-    JobStateToJsonConverter converter = new JobStateToJsonConverter(cmd.getOptionValue('u'), cmd.hasOption("kc"));
+    Properties sysConfig = new Properties();
+    if (cmd.hasOption(sysConfigOption.getLongOpt()) ) {
+      sysConfig = JobConfigurationUtils.fileToProperties(cmd.getOptionValue(sysConfigOption.getLongOpt()));
+    }
+
+    JobStateToJsonConverter converter = new JobStateToJsonConverter(sysConfig, cmd.getOptionValue('u'), cmd.hasOption("kc"));
     StringWriter stringWriter = new StringWriter();
     if (cmd.hasOption('i')) {
       converter.convert(cmd.getOptionValue('n'), cmd.getOptionValue('i'), stringWriter);

diff --git a/gobblin-utility/src/main/java/gobblin/util/JobConfigurationUtils.java b/gobblin-utility/src/main/java/gobblin/util/JobConfigurationUtils.java
@@ -12,11 +12,16 @@
 
 package gobblin.util;
 
+import java.io.IOException;
 import java.util.Properties;
 
+import org.apache.commons.configuration.ConfigurationConverter;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
 import org.apache.hadoop.conf.Configuration;
 
 import gobblin.configuration.State;
+import org.apache.hadoop.fs.Path;
 
 
 /**
@@ -66,4 +71,17 @@ public static void putStateIntoConfiguration(State state, Configuration configur
       configuration.set(key, state.getProp(key));
     }
   }
+
+  /**
+   * Load the properties from the specified file into a {@link Properties} object.
+   *
+   * @param fileName the name of the file to load properties from
+   * @return a new {@link Properties} instance
+   */
+  public static Properties fileToProperties(String fileName) throws IOException, ConfigurationException {
+    Path filePath = new Path(fileName);
+    PropertiesConfiguration propsConfig = new PropertiesConfiguration();
+    propsConfig.load(filePath.getFileSystem(new Configuration()).open(filePath));
+    return ConfigurationConverter.getProperties(propsConfig);
+  }
 }