SQOOP-1428: Sqoop2: From/To: Rebase against Sqoop2 branch for SQOOP-777

core/src/main/java/org/apache/sqoop/framework/JobManager.java

@@ -351,8 +351,8 @@ public MSubmission submit(long jobId, HttpEventContext ctx) {
     request.setJobId(job.getPersistenceId());
     request.setNotificationUrl(notificationBaseUrl + jobId);
     Class<? extends IntermediateDataFormat<?>> dataFormatClass =
-      connector.getIntermediateDataFormat();
+      fromConnector.getIntermediateDataFormat();
-    request.setIntermediateDataFormat(connector.getIntermediateDataFormat());
+    request.setIntermediateDataFormat(fromConnector.getIntermediateDataFormat());
     // Create request object
 
     // Let's register all important jars

execution/mapreduce/src/main/java/org/apache/sqoop/execution/mapreduce/MapreduceExecutionEngine.java

@@ -45,7 +45,9 @@ public SubmissionRequest createSubmissionRequest() {
     return new MRSubmissionRequest();
   }
 
-  public void prepareSubmission(MRSubmissionRequest request) {
+  public void prepareSubmission(SubmissionRequest gRequest) {
+    MRSubmissionRequest request = (MRSubmissionRequest)gRequest;
+
     // Add jar dependencies
     addDependencies(request);
 

execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/ConfigurationUtils.java

@@ -235,23 +235,17 @@ public static Object getConfigFrameworkJob(Configuration configuration) {
    * @param job MapReduce Job object
    * @param schema Schema
    */
-  public static void setFromConnectorSchema(Job job, Schema schema) {
+  public static void setConnectorSchema(ConnectorType type, Job job, Schema schema) {
     if(schema != null) {
+      switch (type) {
+        case FROM:
           job.getCredentials().addSecretKey(SCHEMA_FROM_CONNECTOR_KEY, SchemaSerialization.extractSchema(schema).toJSONString().getBytes());
-    }
-  }
 
-  /**
+        case TO:
-   * Persist To Connector generated schema.
-   *
-   * @param job MapReduce Job object
-   * @param schema Schema
-   */
-  public static void setToConnectorSchema(Job job, Schema schema) {
-    if(schema != null) {
           job.getCredentials().addSecretKey(SCHEMA_TO_CONNECTOR_KEY, SchemaSerialization.extractSchema(schema).toJSONString().getBytes());
       }
     }
+  }
 
   /**
    * Persist Framework generated schema.

execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopMapper.java

@@ -65,10 +65,14 @@ public void run(Context context) throws IOException, InterruptedException {
     String extractorName = conf.get(JobConstants.JOB_ETL_EXTRACTOR);
     Extractor extractor = (Extractor) ClassUtils.instantiate(extractorName);
 
+    // Propagate connector schema in every case for now
+    // TODO: Change to coditional choosing between Connector schemas.
+    Schema schema = ConfigurationUtils.getConnectorSchema(ConnectorType.FROM, conf);
+
     String intermediateDataFormatName = conf.get(JobConstants
       .INTERMEDIATE_DATA_FORMAT);
     data = (IntermediateDataFormat) ClassUtils.instantiate(intermediateDataFormatName);
-    data.setSchema(ConfigurationUtils.getConnectorSchema(conf));
+    data.setSchema(schema);
     dataOut = new SqoopWritable();
 
     // Objects that should be pass to the Executor execution
@@ -76,10 +80,6 @@ public void run(Context context) throws IOException, InterruptedException {
     Object configConnection = null;
     Object configJob = null;
 
-    // Propagate connector schema in every case for now
-    // TODO: Change to coditional choosing between Connector schemas.
-    Schema schema = ConfigurationUtils.getConnectorSchema(ConnectorType.FROM, conf);
-
     // Get configs for extractor
     subContext = new PrefixContext(conf, JobConstants.PREFIX_CONNECTOR_FROM_CONTEXT);
     configConnection = ConfigurationUtils.getConnectorConnectionConfig(ConnectorType.FROM, conf);

execution/mapreduce/src/main/java/org/apache/sqoop/job/mr/SqoopOutputFormatLoadExecutor.java

@@ -73,7 +73,7 @@ public SqoopOutputFormatLoadExecutor(JobContext jobctx) {
     producer = new SqoopRecordWriter();
     data = (IntermediateDataFormat) ClassUtils.instantiate(context
       .getConfiguration().get(JobConstants.INTERMEDIATE_DATA_FORMAT));
-    data.setSchema(ConfigurationUtils.getConnectorSchema(context.getConfiguration()));
+    data.setSchema(ConfigurationUtils.getConnectorSchema(ConnectorType.FROM, context.getConfiguration()));
   }
 
   public RecordWriter<SqoopWritable, NullWritable> getRecordWriter() {

execution/mapreduce/src/test/java/org/apache/sqoop/job/mr/TestConfigurationUtils.java

@@ -96,13 +96,13 @@ public class TestConfigurationUtils {
 //
 //  @Test
 //  public void testConnectorSchema() throws Exception {
-//    ConfigurationUtils.setFromConnectorSchema(job, getSchema("a"));
+//    ConfigurationUtils.setConnectorSchema(job, getSchema("a"));
 //    assertEquals(getSchema("a"), ConfigurationUtils.getFromConnectorSchema(jobConf));
 //  }
 //
 //  @Test
 //  public void testConnectorSchemaNull() throws Exception {
-//    ConfigurationUtils.setFromConnectorSchema(job, null);
+//    ConfigurationUtils.setConnectorSchema(job, null);
 //    assertNull(ConfigurationUtils.getFromConnectorSchema(jobConf));
 //  }
 //

submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java

@@ -207,7 +207,8 @@ public boolean submit(SubmissionRequest generalRequest) {
       ConfigurationUtils.setFrameworkConnectionConfig(ConnectorType.FROM, job, request.getFrameworkConnectionConfig(ConnectorType.FROM));
       ConfigurationUtils.setFrameworkConnectionConfig(ConnectorType.TO, job, request.getFrameworkConnectionConfig(ConnectorType.TO));
       ConfigurationUtils.setConfigFrameworkJob(job, request.getConfigFrameworkJob());
-      ConfigurationUtils.setConnectorSchema(job, request.getSummary().getConnectorSchema());
+      // @TODO(Abe): Persist TO schema.
+      ConfigurationUtils.setConnectorSchema(ConnectorType.FROM, job, request.getSummary().getConnectorSchema());
 
       if(request.getJobName() != null) {
         job.setJobName("Sqoop: " + request.getJobName());