From 18f5b2a77e95e412e806cabf4080a991f79aca92 Mon Sep 17 00:00:00 2001
From: Venkat Ranganathan <venkat@hortonworks.com>
Date: Mon, 19 May 2014 10:23:34 -0700
Subject: [PATCH] SQOOP-1321: Add ability to serialize SqoopOption into JobConf
   (Jarek Jarcec Cecho via Venkat Ranganathan)

---
 conf/sqoop-site-template.xml                  | 14 +++++++
 .../apache/sqoop/mapreduce/ExportJobBase.java |  2 +-
 .../apache/sqoop/mapreduce/ImportJobBase.java |  2 +-
 .../org/apache/sqoop/mapreduce/JobBase.java   | 37 +++++++++++++++++++
 .../org/apache/sqoop/mapreduce/MergeJob.java  |  2 +-
 5 files changed, 54 insertions(+), 3 deletions(-)

diff --git a/conf/sqoop-site-template.xml b/conf/sqoop-site-template.xml
index eccd168e..368af262 100644
--- a/conf/sqoop-site-template.xml
+++ b/conf/sqoop-site-template.xml
@@ -127,6 +127,20 @@ under the License.
   </property>
   -->
 
+  <!--
+    Enabling this option will instruct Sqoop to put all options that
+    were used in the invocation into created mapreduce job(s). This
+    become handy when one needs to investigate what exact options were
+    used in the Sqoop invocation.
+  -->
+  <!--
+  <property>
+    <name>sqoop.jobbase.serialize.sqoopoptions</name>
+    <value>true</value>
+    <description>If true, then all options will be serialized into job.xml
+    </description>
+  </property>
+  -->
 
   <!--
     SERVER CONFIGURATION: If you plan to run a Sqoop metastore on this machine,
diff --git a/src/java/org/apache/sqoop/mapreduce/ExportJobBase.java b/src/java/org/apache/sqoop/mapreduce/ExportJobBase.java
index 53de9c9f..9f510b9b 100644
--- a/src/java/org/apache/sqoop/mapreduce/ExportJobBase.java
+++ b/src/java/org/apache/sqoop/mapreduce/ExportJobBase.java
@@ -379,7 +379,7 @@ public void runExport() throws ExportException, IOException {
       }
     }
 
-    Job job = new Job(conf);
+    Job job = createJob(conf);
     try {
       // Set the external jar to use for the job.
       job.getConfiguration().set("mapred.jar", ormJarFile);
diff --git a/src/java/org/apache/sqoop/mapreduce/ImportJobBase.java b/src/java/org/apache/sqoop/mapreduce/ImportJobBase.java
index 8b1493d0..0ac35de1 100644
--- a/src/java/org/apache/sqoop/mapreduce/ImportJobBase.java
+++ b/src/java/org/apache/sqoop/mapreduce/ImportJobBase.java
@@ -227,7 +227,7 @@ public void runImport(String tableName, String ormJarFile, String splitByCol,
 
     loadJars(conf, ormJarFile, tableClassName);
 
-    Job job = new Job(conf);
+    Job job = createJob(conf);
     try {
       // Set the external jar to use for the job.
       job.getConfiguration().set("mapred.jar", ormJarFile);
diff --git a/src/java/org/apache/sqoop/mapreduce/JobBase.java b/src/java/org/apache/sqoop/mapreduce/JobBase.java
index ddef421b..92a78ac6 100644
--- a/src/java/org/apache/sqoop/mapreduce/JobBase.java
+++ b/src/java/org/apache/sqoop/mapreduce/JobBase.java
@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -48,6 +49,9 @@ public class JobBase {
 
   public static final Log LOG = LogFactory.getLog(JobBase.class.getName());
 
+  public static final String SERIALIZE_SQOOPOPTIONS = "sqoop.jobbase.serialize.sqoopoptions";
+  public static final boolean SERIALIZE_SQOOPOPTIONS_DEFAULT = false;
+
   protected SqoopOptions options;
   protected Class<? extends Mapper> mapperClass;
   protected Class<? extends InputFormat> inputFormatClass;
@@ -308,6 +312,39 @@ public Job getJob() {
     return mrJob;
   }
 
+  /**
+   * Create new Job object in unified way for all types of jobs.
+   *
+   * @param configuration Hadoop configuration that should be used
+   * @return New job object, created object won't be persisted in the instance
+   */
+  public Job createJob(Configuration configuration) throws IOException {
+    // Put the SqoopOptions into job if requested
+    if(configuration.getBoolean(SERIALIZE_SQOOPOPTIONS, SERIALIZE_SQOOPOPTIONS_DEFAULT)) {
+      putSqoopOptionsToConfiguration(options, configuration);
+    }
+
+    return new Job(configuration);
+  }
+
+  /**
+   * Iterates over serialized form of SqoopOptions and put them into Configuration
+   * object.
+   *
+   * @param opts SqoopOptions that should be serialized
+   * @param configuration Target configuration object
+   */
+  public void putSqoopOptionsToConfiguration(SqoopOptions opts, Configuration configuration) {
+    for(Map.Entry<Object, Object> e : opts.writeProperties().entrySet()) {
+      String key = (String)e.getKey();
+      String value = (String)e.getValue();
+
+      // We don't need to do if(value is empty) because that is already done
+      // for us by the SqoopOptions.writeProperties() method.
+      configuration.set("sqoop.opt." + key, value);
+    }
+  }
+
   /**
    * Actually run the MapReduce job.
    */
diff --git a/src/java/org/apache/sqoop/mapreduce/MergeJob.java b/src/java/org/apache/sqoop/mapreduce/MergeJob.java
index 5f321279..4e2a9169 100644
--- a/src/java/org/apache/sqoop/mapreduce/MergeJob.java
+++ b/src/java/org/apache/sqoop/mapreduce/MergeJob.java
@@ -59,7 +59,7 @@ public MergeJob(final SqoopOptions opts) {
 
   public boolean runMergeJob() throws IOException {
     Configuration conf = options.getConf();
-    Job job = new Job(conf);
+    Job job = createJob(conf);
 
     String userClassName = options.getClassName();
     if (null == userClassName) {