SQOOP-2579: Sqoop2: Refactore RepositoryManager to not load structure from repository when loading jobs and links

(Abraham Fine via Jarek Jarcec Cecho)

core/src/main/java/org/apache/sqoop/repository/JdbcRepository.java

@@ -398,6 +398,20 @@ public Object doIt(Connection conn) {
     });
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @SuppressWarnings("unchecked")
+  @Override
+  public List<MLink> findLinksForConnectorUpgrade(final String connectorName) {
+    return (List<MLink>) doWithConnection(new DoWithConnection() {
+      @Override
+      public Object doIt(Connection conn) throws Exception {
+        return handler.findLinksForConnectorUpgrade(connectorName, conn);
+      }
+    });
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -541,6 +555,20 @@ public Object doIt(Connection conn) {
     });
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @SuppressWarnings("unchecked")
+  @Override
+  public List<MJob> findJobsForConnectorUpgrade(final long connectorId) {
+    return (List<MJob>) doWithConnection(new DoWithConnection() {
+      @Override
+      public Object doIt(Connection conn) throws Exception {
+        return handler.findJobsForConnectorUpgrade(connectorId, conn);
+      }
+    });
+  }
+
   /**
    * {@inheritDoc}
    */

core/src/main/java/org/apache/sqoop/repository/JdbcRepositoryHandler.java

@@ -80,6 +80,15 @@ public abstract class JdbcRepositoryHandler {
   public abstract void registerConnector(MConnector mc, Connection conn);
 
 
+  /**
+   * Retrieve links which use the given connector and derive their structure
+   * entirely from the repository.
+   * @param connectorName Connector name whose links should be fetched
+   * @param conn JDBC link for querying repository
+   * @return List of MLinks that use <code>connectorID</code>.
+   */
+  public abstract List<MLink> findLinksForConnectorUpgrade(String connectorName, Connection conn);
+
   /**
    * Retrieve links which use the given connector.
    * @param connectorName Connector name whose links should be fetched
@@ -88,6 +97,16 @@ public abstract class JdbcRepositoryHandler {
    */
   public abstract List<MLink> findLinksForConnector(String connectorName, Connection conn);
 
+  /**
+   * Retrieve jobs which use the given link and derive their structure
+   * entirely from the repository.
+   *
+   * @param connectorId Connector ID whose jobs should be fetched
+   * @param conn JDBC link for querying repository
+   * @return List of MJobs that use <code>linkID</code>.
+   */
+  public abstract List<MJob> findJobsForConnectorUpgrade(long c, Connection conn);
+
   /**
    * Retrieve jobs which use the given link.
    *

core/src/main/java/org/apache/sqoop/repository/Repository.java

@@ -198,6 +198,14 @@ public abstract class Repository {
    */
   public abstract MLink findLink(String name);
 
+  /**
+   * Retrieve links which use the given connector deriving their structure
+   * entirely from the repository.
+   * @param connectorName Connector name whose links should be fetched
+   * @return List of MLink that use <code>connectorId</code>.
+   */
+  public abstract List<MLink> findLinksForConnectorUpgrade(String connectorName);
+
   /**
    * Retrieve links which use the given connector.
    * @param connectorName Connector name whose links should be fetched
@@ -278,6 +286,15 @@ public abstract class Repository {
    */
   public abstract List<MJob> findJobs();
 
+  /**
+   * Retrieve jobs which use the given link deriving structure entirely from
+   * the repository (rather than the connector itself).
+   *
+   * @param connectorId Connector ID whose jobs should be fetched
+   * @return List of MJobs that use <code>linkID</code>.
+   */
+  public abstract List<MJob> findJobsForConnectorUpgrade(long connectorId);
+
   /**
    * Retrieve jobs which use the given link.
    *
@@ -441,9 +458,9 @@ public final void upgradeConnector(MConnector oldConnector, MConnector newConnec
       // 1. Get an upgrader for the connector
       ConnectorConfigurableUpgrader upgrader = connector.getConfigurableUpgrader();
       // 2. Get all links associated with the connector.
-      List<MLink> existingLinksByConnector = findLinksForConnector(connectorName);
+      List<MLink> existingLinksByConnector = findLinksForConnectorUpgrade(connectorName);
       // 3. Get all jobs associated with the connector.
-      List<MJob> existingJobsByConnector = findJobsForConnector(connectorId);
+      List<MJob> existingJobsByConnector = findJobsForConnectorUpgrade(connectorId);
       // -- BEGIN TXN --
       tx = getTransaction();
       tx.begin();

core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java

@@ -224,8 +224,8 @@ public void testConnectorConfigUpgradeWithValidLinksAndJobs() {
     List<MJob> jobList = jobs(job(1, "JA", 1, 1, 1, 1), job(2, "JB", 1, 1, 2, 2));
 
     // mock necessary methods for upgradeConnector() procedure
-    doReturn(linkList).when(repoSpy).findLinksForConnector(anyString());
+    doReturn(linkList).when(repoSpy).findLinksForConnectorUpgrade(anyString());
-    doReturn(jobList).when(repoSpy).findJobsForConnector(anyLong());
+    doReturn(jobList).when(repoSpy).findJobsForConnectorUpgrade(anyLong());
     doNothing().when(repoSpy).updateLink(any(MLink.class), any(RepositoryTransaction.class));
     doNothing().when(repoSpy).updateJob(any(MJob.class), any(RepositoryTransaction.class));
     doNothing().when(repoSpy).upgradeConnectorAndConfigs(any(MConnector.class), any(RepositoryTransaction.class));
@@ -236,8 +236,8 @@ public void testConnectorConfigUpgradeWithValidLinksAndJobs() {
     InOrder txOrder = inOrder(repoTransactionMock);
     InOrder upgraderOrder = inOrder(connectorUpgraderMock);
 
-    repoOrder.verify(repoSpy, times(1)).findLinksForConnector(anyString());
+    repoOrder.verify(repoSpy, times(1)).findLinksForConnectorUpgrade(anyString());
-    repoOrder.verify(repoSpy, times(1)).findJobsForConnector(anyLong());
+    repoOrder.verify(repoSpy, times(1)).findJobsForConnectorUpgrade(anyLong());
     repoOrder.verify(repoSpy, times(1)).getTransaction();
     repoOrder.verify(repoSpy, times(1)).deleteJobInputs("JA", repoTransactionMock);
     repoOrder.verify(repoSpy, times(1)).deleteJobInputs("JB", repoTransactionMock);
@@ -355,13 +355,13 @@ public void testConnectorConfigUpgradeHandlerWithFindLinksForConnectorError() {
 
     SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000,
         "find links for connector error.");
-    doThrow(exception).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doThrow(exception).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
 
     try {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
       verifyNoMoreInteractions(repoHandlerMock);
       return ;
     }
@@ -383,18 +383,19 @@ public void testConnectorConfigUpgradeHandlerWithFindJobsForConnectorError() {
     when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector);
 
     List<MLink> linkList = links(link(1, "LA", 1), link(2, "LB", 1));
-    doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doReturn(linkList).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
 
     SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000,
         "find jobs for connector error.");
-    doThrow(exception).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class));
+    doThrow(exception).when(repoHandlerMock).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
 
     try {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-      verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findJobsForConnectorUpgrade(anyLong()
+        , any(Connection.class));
       verifyNoMoreInteractions(repoHandlerMock);
       return ;
     }
@@ -417,8 +418,9 @@ public void testConnectorConfigUpgradeHandlerWithDeleteJobInputsError() {
 
     List<MLink> linkList = links(link(1, "LA", 1), link(2, "LB", 1));
     List<MJob> jobList = jobs(job(1, "JA", 1, 1, 1, 1), job(2, "JB", 1, 1, 2, 1));
-    doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doReturn(linkList).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-    doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class));
+    doReturn(jobList).when(repoHandlerMock).findJobsForConnectorUpgrade
+      (anyLong(), any(Connection.class));
 
     SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000,
         "delete job inputs for connector error.");
@@ -428,8 +430,9 @@ public void testConnectorConfigUpgradeHandlerWithDeleteJobInputsError() {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-      verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findJobsForConnectorUpgrade(anyLong()
+        , any(Connection.class));
       verify(repoHandlerMock, times(1)).deleteJobInputs(anyString(), any(Connection.class));
       verifyNoMoreInteractions(repoHandlerMock);
       return ;
@@ -453,8 +456,8 @@ public void testConnectorConfigUpgradeHandlerWithDeleteLinkInputsError() {
 
     List<MLink> linkList = links(link(1, "LA", 1), link(2, "LB", 1));
     List<MJob> jobList = jobs(job(1, "JA", 1, 1, 1, 1), job(2, "JB", 1, 1, 2, 1));
-    doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doReturn(linkList).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-    doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class));
+    doReturn(jobList).when(repoHandlerMock).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteJobInputs(anyString(), any(Connection.class));
 
     SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000,
@@ -465,8 +468,8 @@ public void testConnectorConfigUpgradeHandlerWithDeleteLinkInputsError() {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-      verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteJobInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(1)).deleteLinkInputs(anyString(), any(Connection.class));
       verifyNoMoreInteractions(repoHandlerMock);
@@ -491,8 +494,8 @@ public void testConnectorConfigUpgradeHandlerWithUpdateConnectorError() {
 
     List<MLink> linkList = links(link(1, "LA", 1), link(2, "LB", 1));
     List<MJob> jobList = jobs(job(1, "JA", 1, 1, 1, 1), job(2, "JB", 1, 1, 2, 1));
-    doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doReturn(linkList).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-    doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class));
+    doReturn(jobList).when(repoHandlerMock).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteJobInputs(anyString(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteLinkInputs(anyString(), any(Connection.class));
 
@@ -504,8 +507,8 @@ public void testConnectorConfigUpgradeHandlerWithUpdateConnectorError() {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-      verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteJobInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteLinkInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(1)).upgradeConnectorAndConfigs(any(MConnector.class), any(Connection.class));
@@ -533,8 +536,8 @@ public void testConnectorConfigUpgradeHandlerWithUpdateLinkError() {
 
     List<MLink> linkList = links(link(1, "LA", 1), link(2, "LB", 1));
     List<MJob> jobList = jobs(job(1, "JA", 1, 1, 1, 1), job(2, "JB", 1, 1, 2, 1));
-    doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doReturn(linkList).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-    doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class));
+    doReturn(jobList).when(repoHandlerMock).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteJobInputs(anyString(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteLinkInputs(anyString(), any(Connection.class));
     doNothing().when(repoHandlerMock).upgradeConnectorAndConfigs(any(MConnector.class), any(Connection.class));
@@ -548,8 +551,8 @@ public void testConnectorConfigUpgradeHandlerWithUpdateLinkError() {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-      verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteJobInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteLinkInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(1)).upgradeConnectorAndConfigs(any(MConnector.class), any(Connection.class));
@@ -579,8 +582,8 @@ public void testConnectorConfigUpgradeHandlerWithUpdateJobError() {
 
     List<MLink> linkList = links(link(1, "LA", 1), link(2, "LB", 1));;
     List<MJob> jobList = jobs(job(1, "JA", 1, 1, 1, 1), job(2, "JB", 1, 1, 2, 1));
-    doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyString(), any(Connection.class));
+    doReturn(linkList).when(repoHandlerMock).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-    doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class));
+    doReturn(jobList).when(repoHandlerMock).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteJobInputs(anyString(), any(Connection.class));
     doNothing().when(repoHandlerMock).deleteLinkInputs(anyString(), any(Connection.class));
     doNothing().when(repoHandlerMock).upgradeConnectorAndConfigs(any(MConnector.class), any(Connection.class));
@@ -596,8 +599,8 @@ public void testConnectorConfigUpgradeHandlerWithUpdateJobError() {
       repoSpy.upgradeConnector(oldConnector, newConnector);
     } catch (SqoopException ex) {
       assertEquals(ex.getMessage(), exception.getMessage());
-      verify(repoHandlerMock, times(1)).findLinksForConnector(anyString(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findLinksForConnectorUpgrade(anyString(), any(Connection.class));
-      verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class));
+      verify(repoHandlerMock, times(1)).findJobsForConnectorUpgrade(anyLong(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteJobInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(2)).deleteLinkInputs(anyString(), any(Connection.class));
       verify(repoHandlerMock, times(1)).upgradeConnectorAndConfigs(any(MConnector.class), any(Connection.class));

repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryHandler.java

@@ -40,11 +40,13 @@
 import org.apache.sqoop.common.MutableMapContext;
 import org.apache.sqoop.common.SqoopException;
 import org.apache.sqoop.common.SupportedDirections;
+import org.apache.sqoop.connector.ConnectorManager;
 import org.apache.sqoop.driver.Driver;
 import org.apache.sqoop.error.code.CommonRepositoryError;
 import org.apache.sqoop.model.InputEditable;
 import org.apache.sqoop.model.MBooleanInput;
 import org.apache.sqoop.model.MConfig;
+import org.apache.sqoop.model.MConfigList;
 import org.apache.sqoop.model.MConfigType;
 import org.apache.sqoop.model.MConfigurableType;
 import org.apache.sqoop.model.MConnector;
@@ -179,6 +181,23 @@ public void registerConnector(MConnector mc, Connection conn) {
     insertConfigsForConnector(mc, conn);
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public List<MJob> findJobsForConnectorUpgrade(long connectorId, Connection conn) {
+    try (PreparedStatement stmt = conn.prepareStatement(crudQueries.getStmtSelectAllJobsForConnectorConfigurable())) {
+
+      stmt.setLong(1, connectorId);
+      stmt.setLong(2, connectorId);
+      return loadJobsForUpgrade(stmt, conn);
+
+    } catch (SQLException ex) {
+      logException(ex, connectorId);
+      throw new SqoopException(CommonRepositoryError.COMMON_0028, ex);
+    }
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -575,7 +594,23 @@ public List<MLink> findLinks(Connection conn) {
    * {@inheritDoc}
    */
   @Override
-  public List<MLink> findLinksForConnector(String connectorName, Connection conn) {
+  public List<MLink> findLinksForConnectorUpgrade(String connectorName, Connection conn) {
+    try (PreparedStatement linkByConnectorFetchStmt = conn.prepareStatement(crudQueries.getStmtSelectLinkForConnectorConfigurable())) {
+
+      linkByConnectorFetchStmt.setString(1, connectorName);
+      return loadLinksForUpgrade(linkByConnectorFetchStmt, conn);
+    } catch (SQLException ex) {
+      logException(ex, connectorName);
+      throw new SqoopException(CommonRepositoryError.COMMON_0020, ex);
+    }
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public List<MLink> findLinksForConnector(String connectorName,
+                                                  Connection conn) {
     try (PreparedStatement linkByConnectorFetchStmt = conn.prepareStatement(crudQueries.getStmtSelectLinkForConnectorConfigurable())) {
 
       linkByConnectorFetchStmt.setString(1, connectorName);
@@ -1471,7 +1506,7 @@ private List<MConnector> loadConnectors(PreparedStatement stmt, Connection conn)
     return connectors;
   }
 
-  private List<MLink> loadLinks(PreparedStatement stmt,
+  private List<MLink> loadLinksForUpgrade(PreparedStatement stmt,
                                           Connection conn)
       throws SQLException {
     List<MLink> links = new ArrayList<MLink>();
@@ -1516,7 +1551,48 @@ private List<MLink> loadLinks(PreparedStatement stmt,
     return links;
   }
 
-  private List<MJob> loadJobs(PreparedStatement stmt,
+  private List<MLink> loadLinks(PreparedStatement stmt,
+                                Connection conn)
+    throws SQLException {
+    List<MLink> links = new ArrayList<MLink>();
+
+    try (ResultSet rsConnection = stmt.executeQuery();
+         PreparedStatement configStmt = conn.prepareStatement(crudQueries.getStmtSelectConfigForConfiguration());
+         PreparedStatement inputStmt = conn.prepareStatement(crudQueries.getStmtFetchLinkInput());
+    ) {
+      while(rsConnection.next()) {
+        long id = rsConnection.getLong(1);
+        String name = rsConnection.getString(2);
+        long connectorId = rsConnection.getLong(3);
+        boolean enabled = rsConnection.getBoolean(4);
+        String creationUser = rsConnection.getString(5);
+        Date creationDate = rsConnection.getTimestamp(6);
+        String updateUser = rsConnection.getString(7);
+        Date lastUpdateDate = rsConnection.getTimestamp(8);
+        String connectorName = rsConnection.getString(9);
+
+        MLinkConfig connectorLinkConfig = ConnectorManager.getInstance().getConnectorConfigurable(connectorName).getLinkConfig().clone(false);
+        configStmt.setLong(1, connectorId);
+        inputStmt.setLong(1, id);
+        loadInputsForConfigs(connectorLinkConfig, configStmt, inputStmt);
+        MLink link = new MLink(connectorId, connectorLinkConfig);
+
+        link.setPersistenceId(id);
+        link.setName(name);
+        link.setCreationUser(creationUser);
+        link.setCreationDate(creationDate);
+        link.setLastUpdateUser(updateUser);
+        link.setLastUpdateDate(lastUpdateDate);
+        link.setEnabled(enabled);
+
+        links.add(link);
+      }
+    }
+
+    return links;
+  }
+
+  private List<MJob> loadJobsForUpgrade(PreparedStatement stmt,
                               Connection conn)
       throws SQLException {
     List<MJob> jobs = new ArrayList<MJob>();
@@ -1592,6 +1668,82 @@ private List<MJob> loadJobs(PreparedStatement stmt,
     return jobs;
   }
 
+  private List<MJob> loadJobs(PreparedStatement stmt,
+                              Connection conn)
+    throws SQLException {
+    List<MJob> jobs = new ArrayList<MJob>();
+
+    try (ResultSet rsJob = stmt.executeQuery();
+         PreparedStatement driverConfigfetchStmt = conn.prepareStatement
+           (crudQueries.getStmtSelectConfigForConfigurable());
+         PreparedStatement configStmt = conn.prepareStatement(crudQueries
+           .getStmtSelectConfigForConfiguration());
+         PreparedStatement jobInputFetchStmt = conn.prepareStatement(crudQueries.getStmtFetchJobInput());
+         PreparedStatement inputStmt = conn.prepareStatement(crudQueries
+           .getStmtFetchLinkInput())
+    ) {
+
+      // Note: Job does not hold a explicit reference to the driver since every
+      // job has the same driver
+      long driverId = this.findDriver(MDriver.DRIVER_NAME, conn).getPersistenceId();
+
+      while(rsJob.next()) {
+        long fromConnectorId = rsJob.getLong(1);
+        long toConnectorId = rsJob.getLong(2);
+        long id = rsJob.getLong(3);
+        String name = rsJob.getString(4);
+        long fromLinkId = rsJob.getLong(5);
+        long toLinkId = rsJob.getLong(6);
+        boolean enabled = rsJob.getBoolean(7);
+        String createBy = rsJob.getString(8);
+        Date creationDate = rsJob.getTimestamp(9);
+        String updateBy = rsJob.getString(10);
+        Date lastUpdateDate = rsJob.getTimestamp(11);
+        String fromConnectorName = rsJob.getString(12);
+        String toConnectorName = rsJob.getString(13);
+
+
+        driverConfigfetchStmt.setLong(1, driverId);
+        jobInputFetchStmt.setLong(1, id);
+
+        LOG.debug("building config for linkid: " + fromLinkId);
+        MFromConfig mFromConfig = ConnectorManager.getInstance().getConnectorConfigurable(fromConnectorName).clone(false).getFromConfig();
+        configStmt.setLong(1, fromConnectorId);
+        inputStmt.setLong(1, fromLinkId);
+        loadInputsForConfigs(mFromConfig, configStmt, jobInputFetchStmt);
+
+        LOG.debug("building config for linkid: " + toLinkId);
+        MToConfig mToConfig = ConnectorManager.getInstance().getConnectorConfigurable(toConnectorName).clone(false).getToConfig();
+        configStmt.setLong(1, toConnectorId);
+        loadInputsForConfigs(mToConfig, configStmt, jobInputFetchStmt);
+
+        List<MConfig> driverConfig = new ArrayList<MConfig>();
+
+        loadDriverConfigs(driverConfig, driverConfigfetchStmt, jobInputFetchStmt, 2, conn);
+
+        MJob job = new MJob(
+          fromConnectorId, toConnectorId,
+          fromLinkId, toLinkId,
+          new MFromConfig(mFromConfig.getConfigs()),
+          new MToConfig(mToConfig.getConfigs()),
+          new MDriverConfig(driverConfig));
+
+        job.setPersistenceId(id);
+        job.setName(name);
+        job.setCreationUser(createBy);
+        job.setCreationDate(creationDate);
+        job.setLastUpdateUser(updateBy);
+        job.setLastUpdateDate(lastUpdateDate);
+        job.setEnabled(enabled);
+
+        jobs.add(job);
+      }
+    }
+
+    return jobs;
+  }
+
+
   private void registerConfigDirection(Long configId, Direction direction, Connection conn)
       throws SQLException {
     try (PreparedStatement stmt = conn.prepareStatement(crudQueries.getStmtInsertSqConfigDirections())) {
@@ -1935,6 +2087,36 @@ private Direction findConfigDirection(long configId, Connection conn) throws SQL
     }
   }
 
+
+  private void loadInputsForConfigs(MConfigList mConfigList, PreparedStatement configStmt, PreparedStatement inputStmt) throws SQLException {
+    for (MConfig mConfig : mConfigList.getConfigs()) {
+      configStmt.setString(2, mConfig.getName());
+      ResultSet configResults = configStmt.executeQuery();
+
+      while (configResults.next()) {
+        long configId = configResults.getLong(1);
+        String configName = configResults.getString(3);
+        inputStmt.setLong(2, configId);
+        ResultSet inputResults = inputStmt.executeQuery();
+
+        while (inputResults.next()) {
+          long inputId = inputResults.getLong(1);
+          String inputName = inputResults.getString(2);
+          String value = inputResults.getString(10);
+          if (mConfig.getName().equals(configName) && mConfig.getInputNames().contains(inputName)) {
+            MInput mInput = mConfig.getInput(inputName);
+            mInput.setPersistenceId(inputId);
+            if (value == null) {
+              mInput.setEmpty();
+            } else {
+              mInput.restoreFromUrlSafeValueString(value);
+            }
+          }
+        }
+      }
+    }
+  }
+
   /**
    * Load configs and corresponding inputs related to a connector.
    *

repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryInsertUpdateDeleteSelectQuery.java

@@ -124,6 +124,18 @@ public class CommonRepositoryInsertUpdateDeleteSelectQuery {
           + " WHERE " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_CONFIGURABLE) + " = ? "
           + " ORDER BY " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_INDEX);
 
+    private static final String STMT_SELECT_CONFIG_FOR_CONFIGURATION =
+      "SELECT "
+        + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_ID) + ", "
+        + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_CONFIGURABLE) + ", "
+        + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_NAME) + ", "
+        + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_TYPE) + ", "
+        + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_INDEX)
+        + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_CONFIG_NAME)
+        + " WHERE " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_CONFIGURABLE) + " = ? "
+        + " AND " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_NAME) + " = ? "
+        + " ORDER BY " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_CFG_INDEX);
+
 
   //DML: Insert into config
   private static final String STMT_INSERT_INTO_CONFIG =
@@ -306,8 +318,12 @@ public class CommonRepositoryInsertUpdateDeleteSelectQuery {
           + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CREATION_USER) + ", "
           + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CREATION_DATE) + ", "
           + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_UPDATE_USER) + ", "
-          + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_UPDATE_DATE)
+          + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_UPDATE_DATE) + ", "
-          + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME);
+          + CommonRepoUtils.escapeColumnName(COLUMN_SQC_NAME)
+          + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME)
+          + " INNER JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_CONFIGURABLE_NAME)
+          + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CONFIGURABLE) + " = " + CommonRepoUtils.escapeColumnName(COLUMN_SQC_ID);
+
 
   // DML: Select one specific link by name by id
   private static final String STMT_SELECT_LINK_SINGLE_BY_ID =
@@ -329,7 +345,8 @@ public class CommonRepositoryInsertUpdateDeleteSelectQuery {
           + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CREATION_USER) + ", "
           + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CREATION_DATE) + ", "
           + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_UPDATE_USER) + ", "
-          + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_UPDATE_DATE)
+          + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_UPDATE_DATE) + ", "
+          + CommonRepoUtils.escapeColumnName(COLUMN_SQC_NAME)
           + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME)
           + " INNER JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_CONFIGURABLE_NAME)
           + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CONFIGURABLE) + " = " + CommonRepoUtils.escapeColumnName(COLUMN_SQC_ID)
@@ -423,12 +440,18 @@ public class CommonRepositoryInsertUpdateDeleteSelectQuery {
           + "JOB." + CommonRepoUtils.escapeColumnName(COLUMN_SQB_CREATION_USER) + ", "
           + "JOB." + CommonRepoUtils.escapeColumnName(COLUMN_SQB_CREATION_DATE) + ", "
           + "JOB." + CommonRepoUtils.escapeColumnName(COLUMN_SQB_UPDATE_USER) + ", "
-          + "JOB." + CommonRepoUtils.escapeColumnName(COLUMN_SQB_UPDATE_DATE)
+          + "JOB." + CommonRepoUtils.escapeColumnName(COLUMN_SQB_UPDATE_DATE) + ", "
+          + "FROM_CONF_NAME." + CommonRepoUtils.escapeColumnName(COLUMN_SQC_NAME) + ", "
+          + "TO_CONF_NAME." + CommonRepoUtils.escapeColumnName(COLUMN_SQC_NAME)
           + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_JOB_NAME) + " JOB"
           + " LEFT JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME) + " FROM_CONNECTOR"
           + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQB_FROM_LINK) + " = FROM_CONNECTOR." + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_ID)
           + " LEFT JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME) + " TO_CONNECTOR"
-          + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQB_TO_LINK) + " = TO_CONNECTOR." + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_ID);
+          + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQB_TO_LINK) + " = TO_CONNECTOR." + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_ID)
+          + " LEFT JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_CONFIGURABLE_NAME) + " FROM_CONF_NAME"
+          + " ON FROM_CONNECTOR. " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CONFIGURABLE) + " = FROM_CONF_NAME." + CommonRepoUtils.escapeColumnName(COLUMN_SQC_ID)
+          + " LEFT JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_CONFIGURABLE_NAME) + " TO_CONF_NAME"
+          + " ON TO_CONNECTOR. " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_CONFIGURABLE) + " = TO_CONF_NAME." + CommonRepoUtils.escapeColumnName(COLUMN_SQC_ID);
 
   //DML: Select all jobs with order
   private static final String STMT_SELECT_JOB_ALL_WITH_ORDER =
@@ -704,6 +727,10 @@ public String getStmtSelectConfigForConfigurable() {
     return STMT_SELECT_CONFIG_FOR_CONFIGURABLE;
   }
 
+  public String getStmtSelectConfigForConfiguration() {
+      return STMT_SELECT_CONFIG_FOR_CONFIGURATION;
+  }
+
   public String getStmtInsertIntoConfig() {
     return STMT_INSERT_INTO_CONFIG;
   }

repository/repository-derby/pom.xml

@@ -65,6 +65,12 @@ limitations under the License.
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 
 </project>

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/DerbyTestCase.java

@@ -17,40 +17,31 @@
  */
 package org.apache.sqoop.repository.derby;
 
-import static org.apache.sqoop.repository.derby.DerbySchemaUpgradeQuery.*;
-import static org.apache.sqoop.repository.derby.DerbySchemaCreateQuery.*;
-import static org.apache.sqoop.repository.derby.DerbySchemaInsertUpdateDeleteSelectQuery.*;
-import static org.testng.Assert.assertEquals;
-
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.LinkedList;
-import java.util.List;
-
 import org.apache.commons.lang.StringUtils;
 import org.apache.sqoop.common.Direction;
+import org.apache.sqoop.connector.ConnectorManager;
+import org.apache.sqoop.core.SqoopConfiguration;
+import org.apache.sqoop.core.TestUtils;
 import org.apache.sqoop.json.DriverBean;
-import org.apache.sqoop.model.InputEditable;
+import org.apache.sqoop.model.*;
-import org.apache.sqoop.model.MConfig;
+import org.apache.sqoop.repository.JdbcRepositoryProvider;
-import org.apache.sqoop.model.MConnector;
+import org.apache.sqoop.repository.RepoConfigurationConstants;
-import org.apache.sqoop.model.MDriver;
+import org.apache.sqoop.repository.RepositoryManager;
-import org.apache.sqoop.model.MDriverConfig;
-import org.apache.sqoop.model.MFromConfig;
-import org.apache.sqoop.model.MInput;
-import org.apache.sqoop.model.MJob;
-import org.apache.sqoop.model.MLink;
-import org.apache.sqoop.model.MLinkConfig;
-import org.apache.sqoop.model.MMapInput;
-import org.apache.sqoop.model.MStringInput;
-import org.apache.sqoop.model.MToConfig;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 
+import java.sql.*;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.apache.sqoop.repository.derby.DerbySchemaCreateQuery.*;
+import static org.apache.sqoop.repository.derby.DerbySchemaInsertUpdateDeleteSelectQuery.STMT_INSERT_DIRECTION;
+import static org.apache.sqoop.repository.derby.DerbySchemaUpgradeQuery.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
 /**
  * Abstract class with convenience methods for testing derby repository.
  */
@@ -60,13 +51,42 @@ abstract public class DerbyTestCase {
 
   public static final String JDBC_URL = "jdbc:derby:memory:myDB";
 
-  private Connection connection;
+  protected Connection connection;
+
+  ConnectorManager mockConnectorManager;
 
   @BeforeMethod(alwaysRun = true)
   public void setUp() throws Exception {
     // Create link to the database
     Class.forName(DERBY_DRIVER).newInstance();
     connection = DriverManager.getConnection(getStartJdbcUrl());
+
+    Properties testProperties = new Properties();
+    testProperties.setProperty(RepoConfigurationConstants
+      .SYSCFG_REPO_PROVIDER, JdbcRepositoryProvider.class.getCanonicalName());
+    testProperties.setProperty(RepoConfigurationConstants.SYSCFG_REPO_JDBC_HANDLER, DerbyRepositoryHandler.class.getCanonicalName());
+    testProperties.setProperty(RepoConfigurationConstants
+      .SYSCFG_REPO_JDBC_TX_ISOLATION, "serializable");
+    testProperties.setProperty(RepoConfigurationConstants
+      .SYSCFG_REPO_JDBC_MAX_CONN, "10");
+    testProperties.setProperty(RepoConfigurationConstants
+      .SYSCFG_REPO_JDBC_URL, getJdbcUrl());
+    testProperties.setProperty(RepoConfigurationConstants
+      .SYSCFG_REPO_JDBC_DRIVER, DERBY_DRIVER);
+    TestUtils.setupTestConfigurationWithExtraConfig(null, testProperties);
+
+    // We always needs schema for this test case
+    createOrUpgradeSchemaForLatestVersion();
+
+    SqoopConfiguration.getInstance().initialize();
+    RepositoryManager.getInstance().initialize();
+
+    mockConnectorManager = mock(ConnectorManager.class);
+
+    when(mockConnectorManager.getConnectorConfigurable(1L)).thenReturn(getConnector());
+    when(mockConnectorManager.getConnectorConfigurable("A")).thenReturn(getConnector());
+
+    ConnectorManager.setInstance(mockConnectorManager);
   }
 
   @AfterMethod(alwaysRun = true)
@@ -447,18 +467,18 @@ protected void addConnectorA() throws Exception {
       // First config
       runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
           + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I1', " + (i * 2 + 1) + ", 0, 'STRING', false, 30, 'CONNECTOR_ONLY')");
+          + " VALUES('A.I1', " + (i * 2 + 1) + ", 0, 'STRING', false, 30, 'CONNECTOR_ONLY')");
       runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
           + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I2', " + (i * 2 + 1) + ", 1, 'MAP', false, 30, 'CONNECTOR_ONLY')");
+          + " VALUES('A.I2', " + (i * 2 + 1) + ", 1, 'MAP', false, 30, 'CONNECTOR_ONLY')");
 
       // Second config
       runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
           + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I3', " + (i * 2 + 2) + ", 0, 'STRING', false, 30, 'CONNECTOR_ONLY')");
+          + " VALUES('A.I3', " + (i * 2 + 2) + ", 0, 'STRING', false, 30, 'CONNECTOR_ONLY')");
       runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
           + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I4', " + (i * 2 + 2) + ", 1, 'MAP', false, 30, 'CONNECTOR_ONLY')");
+          + " VALUES('A.I4', " + (i * 2 + 2) + ", 1, 'MAP', false, 30, 'CONNECTOR_ONLY')");
     }
 
   }
@@ -468,7 +488,7 @@ protected void loadConnectorAndDriverConfigVersion4() throws Exception {
     runInsertQuery("INSERT INTO SQOOP.SQ_CONFIGURABLE(SQC_NAME, SQC_CLASS, SQC_VERSION, SQC_TYPE)"
         + "VALUES('A', 'org.apache.sqoop.test.A', '1.0-test', 'CONNECTOR')");
 
-    for (String connector : new String[] { "1" }) {
+    String connector = "1";
       // Directions
       runInsertQuery("INSERT INTO SQOOP.SQ_CONNECTOR_DIRECTIONS(SQCD_CONNECTOR, SQCD_DIRECTION)"
           + "VALUES(" + connector + ", 1)");
@@ -477,7 +497,7 @@ protected void loadConnectorAndDriverConfigVersion4() throws Exception {
 
       // connector configs with connectorId as 1
       // all config names have to be unique per type
-      int index = 0;
+      Long index = 1L;
       for (String direction : new String[] { null, "1", "2" }) {
 
         String type;
@@ -486,7 +506,7 @@ protected void loadConnectorAndDriverConfigVersion4() throws Exception {
         } else {
           type = "JOB";
         }
-        String configName = "C1" + type + index;
+        String configName = type + index;
         configId = runInsertQuery("INSERT INTO SQOOP.SQ_CONFIG"
             + "(SQ_CFG_CONFIGURABLE, SQ_CFG_NAME, SQ_CFG_TYPE, SQ_CFG_INDEX) " + "VALUES("
             + connector + ", '" + configName + "', '" + type + "', 0)");
@@ -496,7 +516,11 @@ protected void loadConnectorAndDriverConfigVersion4() throws Exception {
               + "(SQ_CFG_DIR_CONFIG, SQ_CFG_DIR_DIRECTION) " + "VALUES(" + configId + ", "
               + direction + ")");
         }
-        configName = "C2" + type + index;
+
+        loadInputsForConfigVersion4(configName, index);
+
+        index++;
+        configName = type + index;
         configId = runInsertQuery("INSERT INTO SQOOP.SQ_CONFIG"
             + "(SQ_CFG_CONFIGURABLE, SQ_CFG_NAME, SQ_CFG_TYPE, SQ_CFG_INDEX) " + "VALUES("
             + connector + ", '" + configName + "', '" + type + "', 1)");
@@ -506,48 +530,51 @@ protected void loadConnectorAndDriverConfigVersion4() throws Exception {
               + "(SQ_CFG_DIR_CONFIG, SQ_CFG_DIR_DIRECTION) " + "VALUES(" + configId + ", "
               + direction + ")");
         }
+
+        loadInputsForConfigVersion4(configName, index);
+
         index++;
     }
-    }
 
     // insert a driver
     runInsertQuery("INSERT INTO SQOOP.SQ_CONFIGURABLE(SQC_NAME, SQC_CLASS, SQC_VERSION, SQC_TYPE)"
         + "VALUES('SqoopDriver', 'org.apache.sqoop.driver.Driver', '1.0-test', 'DRIVER')");
 
-    // driver config with driverId as 2
+    Long driverIndex1 = runInsertQuery("INSERT INTO SQOOP.SQ_CONFIG"
-    for (String type : new String[] { "JOB" }) {
-      runInsertQuery("INSERT INTO SQOOP.SQ_CONFIG"
         + "(SQ_CFG_CONFIGURABLE, SQ_CFG_NAME, SQ_CFG_TYPE, SQ_CFG_INDEX) " + "VALUES(2"
-          + ", 'd1', '" + type + "', 0)");
+        + ", 'DRIVER1', 'JOB', 0)");
-      runInsertQuery("INSERT INTO SQOOP.SQ_CONFIG"
+    Long driverIndex2 = runInsertQuery("INSERT INTO SQOOP.SQ_CONFIG"
         + "(SQ_CFG_CONFIGURABLE, SQ_CFG_NAME, SQ_CFG_TYPE, SQ_CFG_INDEX) " + "VALUES(2"
-          + ", 'd2', '" + type + "', 1)");
+        + ", 'DRIVER2', 'JOB', 1)");
+
+    loadInputsForConfigVersion4("DRIVER1", driverIndex1);
+    loadInputsForConfigVersion4("DRIVER2", driverIndex2);
   }
 
-    // Input entries
+  private void loadInputsForConfigVersion4(String configName, Long configFK) throws Exception {
-    // Connector LINK config: 0-3
+    //this configuration corresponds to what is generated by getConfigs
-    // Connector job (FROM) config: 4-7
-    // Connector job (TO) config: 8-11
-    // Driver JOB config: 12-15
-    for (int i = 0; i < 4; i++) {
-      // First config inputs with config ids 1,3, 5, 7
     runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
       + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I1', " + (i * 2 + 1) + ", 0, 'STRING', false, 30, 'USER_ONLY')");
+      + " VALUES('" + configName + ".I1', " + configFK + ", 0, 'STRING', false, 30, 'USER_ONLY')");
-      runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
-          + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I2', " + (i * 2 + 1) + ", 1, 'MAP', false, 30, 'CONNECTOR_ONLY')");
 
-      // Second config inputs with config ids 2, 4,,6 8
     runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
       + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I3', " + (i * 2 + 2) + ", 0, 'STRING', false, 30, 'USER_ONLY')");
+      + " VALUES('" + configName + ".I2', " + configFK + ", 1, 'MAP', false, 30, 'CONNECTOR_ONLY')");
+
     runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
       + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
-          + " VALUES('I4', " + (i * 2 + 2) + ", 1, 'MAP', false, 30, 'USER_ONLY')");
+      + " VALUES('" + configName + ".I3', " + configFK + ", 2, 'INTEGER', false, 30, 'ANY')");
-    }
+
+    runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
+      + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_STRLENGTH, SQI_EDITABLE)"
+      + " VALUES('" + configName + ".I4', " + configFK + ", 3, 'BOOLEAN', false, 30, 'USER_ONLY')");
+
+    runInsertQuery("INSERT INTO SQOOP.SQ_INPUT"
+      + "(SQI_NAME, SQI_CONFIG, SQI_INDEX, SQI_TYPE, SQI_STRMASK, SQI_EDITABLE, SQI_ENUMVALS)"
+      + " VALUES('" + configName + ".I5', " + configFK + ", 4, 'ENUM', false, 'ANY', 'YES,NO')");
   }
 
+
   /**
    *Load testing connector and driver config into repository.
    *
@@ -610,7 +637,7 @@ public void loadConnectionsOrLinks(int version) throws Exception {
       runQuery("INSERT INTO SQOOP.SQ_LINK(SQ_LNK_NAME, SQ_LNK_CONFIGURABLE) " + "VALUES('CB', 1)");
 
       for (String ci : new String[] { "1", "2" }) {
-        for (String i : new String[] { "1", "3", "13", "15" }) {
+        for (String i : new String[] { "1", "6", "11", "16", "21", "26" }) {
           runQuery("INSERT INTO SQOOP.SQ_LINK_INPUT"
               + "(SQ_LNKI_LINK, SQ_LNKI_INPUT, SQ_LNKI_VALUE) " + "VALUES(" + ci + ", " + i
               + ", 'Value" + i + "')");
@@ -671,14 +698,9 @@ public void loadJobs(int version) throws Exception {
             + name + index + "', 1, 2)");
       }
 
-      // Odd IDs inputs have values
+      // Every fifth input has a value (1 and 6 are LINK inputs, so not used here)
       for (String ci : new String[] { "1", "2", "3", "4" }) {
-        for (String i : new String[] { "5", "9", "13" }) {
+        for (String i : new String[] { "11", "16", "21", "26", "31", "36"  }) {
-          runQuery("INSERT INTO SQOOP.SQ_JOB_INPUT" + "(SQBI_JOB, SQBI_INPUT, SQBI_VALUE) "
-              + "VALUES(" + ci + ", " + i + ", 'Value" + i + "')");
-        }
-
-        for (String i : new String[] { "7", "11", "15" }) {
           runQuery("INSERT INTO SQOOP.SQ_JOB_INPUT" + "(SQBI_JOB, SQBI_INPUT, SQBI_VALUE) "
               + "VALUES(" + ci + ", " + i + ", 'Value" + i + "')");
         }
@@ -727,7 +749,8 @@ public void loadNonUniqueConfigurablesInVersion4() throws Exception {
     // Insert two configurable - CB and CB
     runInsertQuery("INSERT INTO SQOOP.SQ_CONFIGURABLE(SQC_NAME, SQC_CLASS, SQC_VERSION, SQC_TYPE)"
         + "VALUES('CB', 'org.apache.sqoop.test.B', '1.0-test', 'CONNECTOR')");
-    runInsertQuery("INSERT INTO SQOOP.SQ_CONFIGURABLE(SQC_NAME, SQC_CLASS, SQC_VERSION, SQC_TYPE)"
+    runInsertQuery("INSERT INTO SQOOP.SQ_CONFIGURABLE(SQC_NAME, SQC_CLASS, " +
+      "SQC_VERSION, SQC_TYPE)"
       + "VALUES('CB', 'org.apache.sqoop.test.B', '1.0-test', 'CONNECTOR')");
 
   }
@@ -830,7 +853,8 @@ public void addConnectorB() throws Exception {
         + "VALUES('B', 'org.apache.sqoop.test.B', '1.0-test', 'CONNECTOR')");
     runQuery("INSERT INTO SQOOP.SQ_CONNECTOR_DIRECTIONS (SQCD_CONNECTOR, SQCD_DIRECTION) VALUES ("
         + connectorId + ", 1)");
-    runQuery("INSERT INTO SQOOP.SQ_CONNECTOR_DIRECTIONS (SQCD_CONNECTOR, SQCD_DIRECTION) VALUES ("
+    runQuery("INSERT INTO SQOOP.SQ_CONNECTOR_DIRECTIONS (SQCD_CONNECTOR, " +
+      "SQCD_DIRECTION) VALUES ("
       + connectorId + ", 2)");
   }
 
@@ -960,51 +984,63 @@ protected void fillJob(MJob job) {
   }
 
   protected MLinkConfig getLinkConfig() {
-    return new MLinkConfig(getConfigs("l1", "l2"));
+    return new MLinkConfig(getConfigs("LINK1", "LINK2"));
   }
 
   protected MFromConfig getFromConfig() {
-    return new MFromConfig(getConfigs("from1", "from2"));
+    return new MFromConfig(getConfigs("JOB3", "JOB4"));
   }
 
   protected MFromConfig getBadFromConfig() {
-    return new MFromConfig(getBadConfigs("from1", "from2"));
+    return new MFromConfig(getBadConfigs("FROM1", "FROM2"));
   }
 
   protected MFromConfig getMultipleOverridesFromConfig() {
-    return new MFromConfig(getMultipleOverrideConfigs("from1", "from2"));
+    return new MFromConfig(getMultipleOverrideConfigs("FROM1", "FROM2"));
   }
 
   protected MFromConfig getNonExistentOverridesFromConfig() {
-    return new MFromConfig(getBadConfigsWithNonExistingInputOverrides("from1", "from2"));
+    return new MFromConfig(getBadConfigsWithNonExistingInputOverrides("FROM1", "FROM2"));
   }
 
   protected MToConfig getToConfig() {
-    return new MToConfig(getConfigs("to1", "to2"));
+    return new MToConfig(getConfigs("JOB5", "JOB6"));
   }
 
   protected MDriverConfig getDriverConfig() {
-    return new MDriverConfig(getConfigs("d1", "d2"));
+    return new MDriverConfig(getConfigs("DRIVER1", "DRIVER2"));
   }
 
   protected MDriverConfig getBadDriverConfig() {
-    return new MDriverConfig(getBadConfigsWithSelfOverrides("d1", "d2"));
+    return new MDriverConfig(getBadConfigsWithSelfOverrides("DRIVER1", "DRIVER2"));
   }
 
   protected List<MConfig> getConfigs(String configName1, String configName2) {
     List<MConfig> configs = new LinkedList<MConfig>();
 
     List<MInput<?>> inputs = new LinkedList<MInput<?>>();
-    MInput input = new MStringInput("I1", false, InputEditable.ANY, "I2", (short) 30);
+    MInput input = new MStringInput(configName1 + ".I1", false, InputEditable.USER_ONLY, configName1 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I2", false, InputEditable.ANY, StringUtils.EMPTY);
+    input = new MMapInput(configName1 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName1 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName1 + ".I3", false, InputEditable.ANY, configName1 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName1 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName1 + ".I5", false, InputEditable.ANY, configName1 + ".I4," + configName1 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName1, inputs));
 
     inputs = new LinkedList<MInput<?>>();
-    input = new MStringInput("I3", false, InputEditable.ANY, "I4", (short) 30);
+    input = new MStringInput(configName2 + ".I1", false, InputEditable.USER_ONLY, configName2 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I4", false, InputEditable.USER_ONLY, "I3");
+    input = new MMapInput(configName2 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName2 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName2 + ".I3", false, InputEditable.ANY, configName2 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName2 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName2 + ".I5", false, InputEditable.ANY, configName2 + ".I4," + configName2 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName2, inputs));
 
@@ -1016,16 +1052,28 @@ protected List<MConfig> getBadConfigs(String configName1, String configName2) {
 
     List<MInput<?>> inputs = new LinkedList<MInput<?>>();
     // I1 overrides another user_only attribute, hence a bad config
-    MInput input = new MStringInput("I1", false, InputEditable.USER_ONLY, "I2", (short) 30);
+    MInput input = new MStringInput(configName1 + ".I1", false, InputEditable.USER_ONLY, configName1 + ".I4", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I2", false, InputEditable.USER_ONLY, "I1");
+    input = new MMapInput(configName1 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName1 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName1 + ".I3", false, InputEditable.ANY, configName1 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName1 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName1 + ".I5", false, InputEditable.ANY, configName1 + ".I4," + configName1 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName1, inputs));
 
     inputs = new LinkedList<MInput<?>>();
-    input = new MStringInput("I3", false, InputEditable.ANY, StringUtils.EMPTY, (short) 30);
+    input = new MStringInput(configName2 + ".I1", false, InputEditable.USER_ONLY, configName2 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I4", false, InputEditable.ANY, StringUtils.EMPTY);
+    input = new MMapInput(configName2 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName2 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName2 + ".I3", false, InputEditable.ANY, configName2 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName2 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName2 + ".I5", false, InputEditable.ANY, configName2 + ".I4," + configName2 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName2, inputs));
 
@@ -1037,16 +1085,28 @@ protected List<MConfig> getBadConfigsWithSelfOverrides(String configName1, Strin
 
     List<MInput<?>> inputs = new LinkedList<MInput<?>>();
     // I1 overrides another user_only attribute, hence a bad config
-    MInput input = new MStringInput("I1", false, InputEditable.USER_ONLY, "I2", (short) 30);
+    MInput input = new MStringInput(configName1 + ".I1", false, InputEditable.USER_ONLY, configName1 + ".I4", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I2", false, InputEditable.USER_ONLY, "I2");
+    input = new MMapInput(configName1 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName1 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName1 + ".I3", false, InputEditable.ANY, configName1 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName1 + ".I4", false, InputEditable.USER_ONLY, configName1 + ".I4");
+    inputs.add(input);
+    input = new MEnumInput(configName1 + ".I5", false, InputEditable.ANY, configName1 + ".I4," + configName1 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName1, inputs));
 
     inputs = new LinkedList<MInput<?>>();
-    input = new MStringInput("I3", false, InputEditable.ANY, StringUtils.EMPTY, (short) 30);
+    input = new MStringInput(configName2 + ".I1", false, InputEditable.USER_ONLY, configName2 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I4", false, InputEditable.ANY, StringUtils.EMPTY);
+    input = new MMapInput(configName2 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName2 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName2 + ".I3", false, InputEditable.ANY, configName2 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName2 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName2 + ".I5", false, InputEditable.ANY, configName2 + ".I4," + configName2 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName2, inputs));
 
@@ -1057,19 +1117,28 @@ protected List<MConfig> getMultipleOverrideConfigs(String configName1, String co
     List<MConfig> configs = new LinkedList<MConfig>();
 
     List<MInput<?>> inputs = new LinkedList<MInput<?>>();
-    // I1 overrides another user_only attribute, hence a bad config
+    MInput input = new MStringInput(configName1 + ".I1", false, InputEditable.USER_ONLY, configName1 + ".I2", (short) 30);
-    MInput input = new MStringInput("I1", false, InputEditable.USER_ONLY, "I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I2", false, InputEditable.ANY, "I1,I3");
+    input = new MMapInput(configName1 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName1 + ".I1," + configName1 + ".I3");
     inputs.add(input);
-    input = new MMapInput("I3", false, InputEditable.CONNECTOR_ONLY, "I1");
+    input = new MIntegerInput(configName1 + ".I3", false, InputEditable.ANY, configName1 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName1 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName1 + ".I5", false, InputEditable.ANY, configName1 + ".I4," + configName1 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName1, inputs));
 
     inputs = new LinkedList<MInput<?>>();
-    input = new MStringInput("I3", false, InputEditable.ANY, StringUtils.EMPTY, (short) 30);
+    input = new MStringInput(configName2 + ".I1", false, InputEditable.USER_ONLY, configName2 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I4", false, InputEditable.ANY, StringUtils.EMPTY);
+    input = new MMapInput(configName2 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName2 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName2 + ".I3", false, InputEditable.ANY, configName2 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName2 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName2 + ".I5", false, InputEditable.ANY, configName2 + ".I4," + configName2 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName2, inputs));
 
@@ -1081,17 +1150,29 @@ protected List<MConfig> getBadConfigsWithNonExistingInputOverrides(String config
     List<MConfig> configs = new LinkedList<MConfig>();
 
     List<MInput<?>> inputs = new LinkedList<MInput<?>>();
-    // I1 overrides another user_only attribute, hence a bad config
+    // I2 overrides a nonexistant input
-    MInput input = new MStringInput("I1", false, InputEditable.USER_ONLY, "I2", (short) 30);
+    MInput input = new MStringInput(configName1 + ".I1", false, InputEditable.USER_ONLY, configName1 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I2", false, InputEditable.USER_ONLY, "Foo");
+    input = new MMapInput(configName1 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName1 + ".FOO");
+    inputs.add(input);
+    input = new MIntegerInput(configName1 + ".I3", false, InputEditable.ANY, configName1 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName1 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName1 + ".I5", false, InputEditable.ANY, configName1 + ".I4," + configName1 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName1, inputs));
 
     inputs = new LinkedList<MInput<?>>();
-    input = new MStringInput("I3", false, InputEditable.ANY, StringUtils.EMPTY, (short) 30);
+    input = new MStringInput(configName2 + ".I1", false, InputEditable.USER_ONLY, configName2 + ".I2", (short) 30);
     inputs.add(input);
-    input = new MMapInput("I4", false, InputEditable.ANY, StringUtils.EMPTY);
+    input = new MMapInput(configName2 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName2 + ".I5");
+    inputs.add(input);
+    input = new MIntegerInput(configName2 + ".I3", false, InputEditable.ANY, configName2 + ".I1");
+    inputs.add(input);
+    input = new MBooleanInput(configName2 + ".I4", false, InputEditable.USER_ONLY, StringUtils.EMPTY);
+    inputs.add(input);
+    input = new MEnumInput(configName2 + ".I5", false, InputEditable.ANY, configName2 + ".I4," + configName2 + ".I3", new String[] {"YES", "NO"});
     inputs.add(input);
     configs.add(new MConfig(configName2, inputs));
 
@@ -1138,7 +1219,7 @@ protected long countForTable(String table) throws Exception {
    */
   protected void assertCountForTable(String table, long expected) throws Exception {
     long count = countForTable(table);
-    assertEquals(expected, count);
+    assertEquals(count, expected);
   }
 
   /**

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestConnectorHandling.java

@@ -18,15 +18,17 @@
 package org.apache.sqoop.repository.derby;
 
 import org.apache.sqoop.common.SqoopException;
+import org.apache.sqoop.core.SqoopConfiguration;
 import org.apache.sqoop.model.MConnector;
+import org.apache.sqoop.repository.RepositoryManager;
+import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
 import java.util.List;
 
-import static org.testng.Assert.assertEquals;
+import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.*;
-import static org.testng.Assert.assertNull;
 
 /**
  * Test connector methods on Derby repository.
@@ -39,8 +41,14 @@ public class TestConnectorHandling extends DerbyTestCase {
   public void setUp() throws Exception {
     super.setUp();
     handler = new DerbyRepositoryHandler();
-    // We always needs schema for this test case
+  }
-    createOrUpgradeSchemaForLatestVersion();
+
+  @AfterMethod(alwaysRun = true)
+  public void tearDown() throws Exception {
+    super.tearDown();
+
+    SqoopConfiguration.getInstance().destroy();
+    RepositoryManager.getInstance().destroy();
   }
 
   @Test
@@ -48,7 +56,8 @@ public void testFindConnectorById() throws Exception {
     // On empty repository, no connectors should be there
     assertNull(handler.findConnector(1L, getDerbyDatabaseConnection()));
     // Load connector into repository
-    addConnectorA();
+    loadConnectorAndDriverConfig();
+    when(mockConnectorManager.getConnectorConfigurable(1L)).thenReturn(getConnector());
     // Retrieve it
     MConnector connector = handler.findConnector(1L, getDerbyDatabaseConnection());
     assertNotNull(connector);
@@ -63,7 +72,8 @@ public void testFindConnectorByName() throws Exception {
     // On empty repository, no connectors should be there
     assertNull(handler.findConnector("A", getDerbyDatabaseConnection()));
     // Load connector into repository
-    addConnectorA();
+    loadConnectorAndDriverConfig();
+    when(mockConnectorManager.getConnectorConfigurable(1L)).thenReturn(getConnector());
     // Retrieve it
     MConnector connector = handler.findConnector("A", getDerbyDatabaseConnection());
     assertNotNull(connector);
@@ -99,8 +109,8 @@ public void testRegisterConnector() throws Exception {
     // Now check content in corresponding tables
     assertCountForTable("SQOOP.SQ_CONFIGURABLE", 1);
     assertCountForTable("SQOOP.SQ_CONFIG", 6);
-    assertCountForTable("SQOOP.SQ_INPUT", 12);
+    assertCountForTable("SQOOP.SQ_INPUT", 30);
-    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 9);
+    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 30);
 
 
     // Registered connector should be easily recovered back
@@ -138,8 +148,8 @@ public void testFromDirection() throws Exception {
     // Now check content in corresponding tables
     assertCountForTable("SQOOP.SQ_CONFIGURABLE", 1);
     assertCountForTable("SQOOP.SQ_CONFIG", 4);
-    assertCountForTable("SQOOP.SQ_INPUT", 8);
+    assertCountForTable("SQOOP.SQ_INPUT", 20);
-    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 6);
+    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 20);
 
     // Registered connector should be easily recovered back
     MConnector retrieved = handler.findConnector("A", getDerbyDatabaseConnection());
@@ -159,8 +169,8 @@ public void testToDirection() throws Exception {
     // Now check content in corresponding tables
     assertCountForTable("SQOOP.SQ_CONFIGURABLE", 1);
     assertCountForTable("SQOOP.SQ_CONFIG", 4);
-    assertCountForTable("SQOOP.SQ_INPUT", 8);
+    assertCountForTable("SQOOP.SQ_INPUT", 20);
-    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 6);
+    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 20);
 
     // Registered connector should be easily recovered back
     MConnector retrieved = handler.findConnector("A", getDerbyDatabaseConnection());
@@ -180,8 +190,8 @@ public void testNeitherDirection() throws Exception {
     // Now check content in corresponding tables
     assertCountForTable("SQOOP.SQ_CONFIGURABLE", 1);
     assertCountForTable("SQOOP.SQ_CONFIG", 2);
-    assertCountForTable("SQOOP.SQ_INPUT", 4);
+    assertCountForTable("SQOOP.SQ_INPUT", 10);
-    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 3);
+    assertCountForTable("SQOOP.SQ_INPUT_RELATION", 10);
 
     // Registered connector should be easily recovered back
     MConnector retrieved = handler.findConnector("A", getDerbyDatabaseConnection());

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestDriverHandling.java

@@ -39,8 +39,6 @@ public class TestDriverHandling extends DerbyTestCase {
   public void setUp() throws Exception {
     super.setUp();
     handler = new DerbyRepositoryHandler();
-    // We always needs schema for this test case
-    createOrUpgradeSchemaForLatestVersion();
   }
 
   @Test

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestInputTypes.java

@@ -55,8 +55,6 @@ public void setUp() throws Exception {
     super.setUp();
 
     handler = new DerbyRepositoryHandler();
-    // We always needs schema for this test case
-    createOrUpgradeSchemaForLatestVersion();
   }
 
   /**
@@ -101,22 +99,22 @@ public void testEntityDataSerialization() throws Exception {
     // Connection object with all various values
     MLink link = new MLink(connector.getPersistenceId(), connector.getLinkConfig());
     MLinkConfig linkConfig = link.getConnectorLinkConfig();
-    assertEquals(linkConfig.getStringInput("l1.I1").getEditable(), InputEditable.ANY);
+    assertEquals(linkConfig.getStringInput("LINK1.I1").getEditable(), InputEditable.USER_ONLY);
-    assertEquals(linkConfig.getStringInput("l1.I1").getOverrides(), "l1.I2");
+    assertEquals(linkConfig.getStringInput("LINK1.I1").getOverrides(), "LINK1.I2");
-    assertEquals(linkConfig.getMapInput("l1.I2").getEditable(), InputEditable.CONNECTOR_ONLY);
+    assertEquals(linkConfig.getMapInput("LINK1.I2").getEditable(), InputEditable.CONNECTOR_ONLY);
-    assertEquals(linkConfig.getMapInput("l1.I2").getOverrides(), "l1.I5");
+    assertEquals(linkConfig.getMapInput("LINK1.I2").getOverrides(), "LINK1.I5");
-    assertEquals(linkConfig.getIntegerInput("l1.I3").getEditable(), InputEditable.ANY);
+    assertEquals(linkConfig.getIntegerInput("LINK1.I3").getEditable(), InputEditable.ANY);
-    assertEquals(linkConfig.getIntegerInput("l1.I3").getOverrides(), "l1.I1");
+    assertEquals(linkConfig.getIntegerInput("LINK1.I3").getOverrides(), "LINK1.I1");
-    assertEquals(linkConfig.getBooleanInput("l1.I4").getEditable(), InputEditable.USER_ONLY);
+    assertEquals(linkConfig.getBooleanInput("LINK1.I4").getEditable(), InputEditable.USER_ONLY);
-    assertEquals(linkConfig.getBooleanInput("l1.I4").getOverrides(), "");
+    assertEquals(linkConfig.getBooleanInput("LINK1.I4").getOverrides(), "");
-    assertEquals(linkConfig.getEnumInput("l1.I5").getEditable(), InputEditable.ANY);
+    assertEquals(linkConfig.getEnumInput("LINK1.I5").getEditable(), InputEditable.ANY);
-    assertEquals(linkConfig.getEnumInput("l1.I5").getOverrides(), "l1.I4,l1.I3");
+    assertEquals(linkConfig.getEnumInput("LINK1.I5").getOverrides(), "LINK1.I4,LINK1.I3");
 
-    linkConfig.getStringInput("l1.I1").setValue("A");
+    linkConfig.getStringInput("LINK1.I1").setValue("A");
-    linkConfig.getMapInput("l1.I2").setValue(map);
+    linkConfig.getMapInput("LINK1.I2").setValue(map);
-    linkConfig.getIntegerInput("l1.I3").setValue(1);
+    linkConfig.getIntegerInput("LINK1.I3").setValue(1);
-    linkConfig.getBooleanInput("l1.I4").setValue(true);
+    linkConfig.getBooleanInput("LINK1.I4").setValue(true);
-    linkConfig.getEnumInput("l1.I5").setValue("YES");
+    linkConfig.getEnumInput("LINK1.I5").setValue("YES");
 
     // Create the link in repository
     handler.createLink(link, getDerbyDatabaseConnection());
@@ -125,50 +123,13 @@ public void testEntityDataSerialization() throws Exception {
     // Retrieve created link
     MLink retrieved = handler.findLink(link.getPersistenceId(), getDerbyDatabaseConnection());
     linkConfig = retrieved.getConnectorLinkConfig();
-    assertEquals("A", linkConfig.getStringInput("l1.I1").getValue());
+    assertEquals("A", linkConfig.getStringInput("LINK1.I1").getValue());
-    assertEquals(map, linkConfig.getMapInput("l1.I2").getValue());
+    assertEquals(map, linkConfig.getMapInput("LINK1.I2").getValue());
-    assertEquals(1, (int) linkConfig.getIntegerInput("l1.I3").getValue());
+    assertEquals(1, (int) linkConfig.getIntegerInput("LINK1.I3").getValue());
-    assertEquals(true, (boolean) linkConfig.getBooleanInput("l1.I4").getValue());
+    assertEquals(true, (boolean) linkConfig.getBooleanInput("LINK1.I4").getValue());
-    assertEquals("YES", linkConfig.getEnumInput("l1.I5").getValue());
+    assertEquals("YES", linkConfig.getEnumInput("LINK1.I5").getValue());
-    assertEquals(linkConfig.getEnumInput("l1.I5").getEditable(), InputEditable.ANY);
+    assertEquals(linkConfig.getEnumInput("LINK1.I5").getEditable(), InputEditable.ANY);
-    assertEquals(linkConfig.getEnumInput("l1.I5").getOverrides(), "l1.I4,l1.I3");
+    assertEquals(linkConfig.getEnumInput("LINK1.I5").getOverrides(), "LINK1.I4,LINK1.I3");
 
   }
-
-  /**
-   * Overriding parent method to get forms with all supported data types.
-   *
-   * @return Forms with all data types
-   */
-  @Override
-  protected List<MConfig> getConfigs(String configName1, String configName2) {
-    List<MConfig> configs = new LinkedList<MConfig>();
-
-    List<MInput<?>> inputs;
-    MInput input;
-
-    inputs = new LinkedList<MInput<?>>();
-
-    input = new MStringInput(configName1 + ".I1", false, InputEditable.ANY, configName1 + ".I2",
-        (short) 30);
-    inputs.add(input);
-
-    input = new MMapInput(configName1 + ".I2", false, InputEditable.CONNECTOR_ONLY, configName1
-        + ".I5");
-    inputs.add(input);
-
-    input = new MIntegerInput(configName1 + ".I3", false, InputEditable.ANY, configName1 + ".I1");
-    inputs.add(input);
-
-    input = new MBooleanInput(configName1 + ".I4", false, InputEditable.USER_ONLY,
-        StringUtils.EMPTY);
-    inputs.add(input);
-
-    input = new MEnumInput(configName1 + ".I5", false, InputEditable.ANY, configName1 + ".I4,"
-        + configName1 + ".I3", new String[] { "YES", "NO" });
-    inputs.add(input);
-
-    configs.add(new MConfig(configName1, inputs));
-    return configs;
-  }
 }

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestJobHandling.java

@@ -22,6 +22,7 @@
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import java.sql.Connection;
 import java.util.HashMap;
@@ -51,8 +52,7 @@ public void setUp() throws Exception {
     super.setUp();
     derbyConnection = getDerbyDatabaseConnection();
     handler = new DerbyRepositoryHandler();
-    // We always needs create/ upgrade schema for this test case
+
-    createOrUpgradeSchemaForLatestVersion();
     loadConnectorAndDriverConfig();
     loadLinksForLatestVersion();
   }
@@ -73,23 +73,23 @@ public void testFindJob() throws Exception {
 
     configs = firstJob.getFromJobConfig().getConfigs();
     assertEquals(2, configs.size());
-    assertEquals("Value5", configs.get(0).getInputs().get(0).getValue());
+    assertEquals("Value11", configs.get(0).getInputs().get(0).getValue());
     assertNull(configs.get(0).getInputs().get(1).getValue());
-    assertEquals("Value5", configs.get(0).getInputs().get(0).getValue());
+    assertEquals("Value16", configs.get(1).getInputs().get(0).getValue());
     assertNull(configs.get(1).getInputs().get(1).getValue());
 
     configs = firstJob.getToJobConfig().getConfigs();
     assertEquals(2, configs.size());
-    assertEquals("Value9", configs.get(0).getInputs().get(0).getValue());
+    assertEquals("Value21", configs.get(0).getInputs().get(0).getValue());
     assertNull(configs.get(0).getInputs().get(1).getValue());
-    assertEquals("Value9", configs.get(0).getInputs().get(0).getValue());
+    assertEquals("Value26", configs.get(1).getInputs().get(0).getValue());
     assertNull(configs.get(1).getInputs().get(1).getValue());
 
     configs = firstJob.getDriverConfig().getConfigs();
     assertEquals(2, configs.size());
-    assertEquals("Value13", configs.get(0).getInputs().get(0).getValue());
+    assertEquals("Value31", configs.get(0).getInputs().get(0).getValue());
     assertNull(configs.get(0).getInputs().get(1).getValue());
-    assertEquals("Value15", configs.get(1).getInputs().get(0).getValue());
+    assertEquals("Value36", configs.get(1).getInputs().get(0).getValue());
     assertNull(configs.get(1).getInputs().get(1).getValue());
   }
 
@@ -112,7 +112,7 @@ public void testFindJobs() throws Exception {
   }
 
   @Test
-  public void testFindJobsByConnector() throws Exception {
+  public void testFindJobsForConnector() throws Exception {
     List<MJob> list;
     // Load empty list on empty repository
     list = handler.findJobs(derbyConnection);
@@ -129,6 +129,24 @@ public void testFindJobsByConnector() throws Exception {
     assertEquals("JD0", list.get(3).getName());
   }
 
+  @Test
+  public void testFindJobsByConnectorForUpgrade() throws Exception {
+    List<MJob> list;
+    // Load empty list on empty repository
+    list = handler.findJobs(derbyConnection);
+    assertEquals(0, list.size());
+    loadJobsForLatestVersion();
+
+    // Load all 4 jobs on loaded repository
+    list = handler.findJobsForConnectorUpgrade(1, derbyConnection);
+    assertEquals(4, list.size());
+
+    assertEquals("JA0", list.get(0).getName());
+    assertEquals("JB0", list.get(1).getName());
+    assertEquals("JC0", list.get(2).getName());
+    assertEquals("JD0", list.get(3).getName());
+  }
+
   @Test
   public void testFindJobsForNonExistingConnector() throws Exception {
     List<MJob> list;
@@ -137,7 +155,7 @@ public void testFindJobsForNonExistingConnector() throws Exception {
     assertEquals(0, list.size());
     loadJobsForLatestVersion();
 
-    list = handler.findJobsForConnector(11, derbyConnection);
+    list = handler.findJobsForConnectorUpgrade(11, derbyConnection);
     assertEquals(0, list.size());
   }
 

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestLinkHandling.java

@@ -22,6 +22,7 @@
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 import java.util.List;
 
@@ -45,8 +46,7 @@ public void setUp() throws Exception {
     super.setUp();
 
     handler = new DerbyRepositoryHandler();
-    // We always needs schema for this test case
+
-    createOrUpgradeSchemaForLatestVersion();
     // We always needs connector and framework structures in place
     loadConnectorAndDriverConfig();
   }
@@ -70,7 +70,7 @@ public void testFindLink() throws Exception {
     configs = linkA.getConnectorLinkConfig().getConfigs();
     assertEquals("Value1", configs.get(0).getInputs().get(0).getValue());
     assertNull(configs.get(0).getInputs().get(1).getValue());
-    assertEquals("Value3", configs.get(1).getInputs().get(0).getValue());
+    assertEquals("Value6", configs.get(1).getInputs().get(0).getValue());
     assertNull(configs.get(1).getInputs().get(1).getValue());
   }
 
@@ -92,7 +92,7 @@ public void testFindLinkByName() throws Exception {
     configs = linkA.getConnectorLinkConfig().getConfigs();
     assertEquals("Value1", configs.get(0).getInputs().get(0).getValue());
     assertNull(configs.get(0).getInputs().get(1).getValue());
-    assertEquals("Value3", configs.get(1).getInputs().get(0).getValue());
+    assertEquals("Value6", configs.get(1).getInputs().get(0).getValue());
     assertNull(configs.get(1).getInputs().get(1).getValue());
   }
 
@@ -125,7 +125,27 @@ public void testFindLinksByConnector() throws Exception {
     loadLinksForLatestVersion();
 
     // Load all two links on loaded repository
-    list = handler.findLinksForConnector("A", getDerbyDatabaseConnection());
+    list = handler.findLinksForConnector("A",
+      getDerbyDatabaseConnection());
+    assertEquals(2, list.size());
+
+    assertEquals("CA", list.get(0).getName());
+    assertEquals("CB", list.get(1).getName());
+  }
+
+  @Test
+  public void testFindLinksByConnectorForUpgrade() throws Exception {
+    List<MLink> list;
+
+    // Load empty list on empty repository
+    list = handler.findLinks(getDerbyDatabaseConnection());
+    assertEquals(0, list.size());
+
+    loadLinksForLatestVersion();
+
+    // Load all two links on loaded repository
+    list = handler.findLinksForConnectorUpgrade("A",
+      getDerbyDatabaseConnection());
     assertEquals(2, list.size());
 
     assertEquals("CA", list.get(0).getName());
@@ -141,7 +161,8 @@ public void testFindLinksByNonExistingConnector() throws Exception {
 
     loadLinksForLatestVersion();
 
-    list = handler.findLinksForConnector("NONEXISTCONNECTOR", getDerbyDatabaseConnection());
+    list = handler.findLinksForConnectorUpgrade("NONEXISTCONNECTOR",
+      getDerbyDatabaseConnection());
     assertEquals(0, list.size());
   }
 
@@ -238,7 +259,7 @@ public void testUpdateLink() throws Exception {
 
     assertEquals(1, link.getPersistenceId());
     assertCountForTable("SQOOP.SQ_LINK", 2);
-    assertCountForTable("SQOOP.SQ_LINK_INPUT", 6);
+    assertCountForTable("SQOOP.SQ_LINK_INPUT", 8);
 
     MLink retrieved = handler.findLink(1, getDerbyDatabaseConnection());
     assertEquals("name", link.getName());
@@ -275,7 +296,7 @@ public void testDeleteLink() throws Exception {
 
     handler.deleteLink("CA", getDerbyDatabaseConnection());
     assertCountForTable("SQOOP.SQ_LINK", 1);
-    assertCountForTable("SQOOP.SQ_LINK_INPUT", 4);
+    assertCountForTable("SQOOP.SQ_LINK_INPUT", 6);
 
     handler.deleteLink("CB", getDerbyDatabaseConnection());
     assertCountForTable("SQOOP.SQ_LINK", 0);

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestRepositoryUpgrade.java

@@ -21,6 +21,7 @@
 import static org.testng.Assert.assertTrue;
 
 import java.sql.Connection;
+import java.sql.DriverManager;
 import java.sql.SQLIntegrityConstraintViolationException;
 
 import org.testng.annotations.BeforeMethod;
@@ -32,7 +33,10 @@ public class TestRepositoryUpgrade extends DerbyTestCase {
 
   @BeforeMethod(alwaysRun = true)
   public void setUp() throws Exception {
-    super.setUp();
+    // Create link to the database
+    Class.forName(DERBY_DRIVER).newInstance();
+    connection = DriverManager.getConnection(getStartJdbcUrl());
+
     handler = new TestDerbyRepositoryHandler();
   }
 

repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestSubmissionHandling.java

@@ -45,8 +45,6 @@ public void setUp() throws Exception {
     super.setUp();
 
     handler = new DerbyRepositoryHandler();
-    // We always needs schema for this test case
-    super.createOrUpgradeSchemaForLatestVersion();
 
     // We always needs connector and framework structures in place
     loadConnectorAndDriverConfig();

repository/repository-mysql/src/test/java/org/apache/sqoop/integration/repository/mysql/TestJobHandling.java

@@ -136,7 +136,7 @@ public void testFindJobs() throws Exception {
   @Test
   public void testFindJobsByConnector() throws Exception {
     List<MJob> list = handler
-        .findJobsForConnector(
+        .findJobsForConnectorUpgrade(
           handler.findConnector("A", provider.getConnection())
             .getPersistenceId(), provider.getConnection());
     assertEquals(2, list.size());
@@ -147,7 +147,7 @@ public void testFindJobsByConnector() throws Exception {
   @Test
   public void testFindJobsForNonExistingConnector() throws Exception {
     List<MJob> list = handler
-        .findJobsForConnector(11, provider.getConnection());
+        .findJobsForConnectorUpgrade(11, provider.getConnection());
     assertEquals(0, list.size());
   }
 

repository/repository-mysql/src/test/java/org/apache/sqoop/integration/repository/mysql/TestLinkHandling.java

@@ -133,7 +133,7 @@ public void testFindLinksByConnector() throws Exception {
     List<MLink> list;
 
     // Load all two links on loaded repository
-    list = handler.findLinksForConnector("A", provider.getConnection());
+    list = handler.findLinksForConnectorUpgrade("A", provider.getConnection());
     assertEquals(1, list.size());
     assertEquals(LINK_A_NAME, list.get(0).getName());
 
@@ -143,13 +143,13 @@ public void testFindLinksByConnector() throws Exception {
     }
 
     // Load empty list on empty repository
-    list = handler.findLinksForConnector("A", provider.getConnection());
+    list = handler.findLinksForConnectorUpgrade("A", provider.getConnection());
     assertEquals(0, list.size());
   }
 
   @Test
   public void testFindLinksByNonExistingConnector() throws Exception {
-    List<MLink> list = handler.findLinksForConnector("NONEXISTCONNECTOR",
+    List<MLink> list = handler.findLinksForConnectorUpgrade("NONEXISTCONNECTOR",
       provider.getConnection());
     assertEquals(0, list.size());
   }

repository/repository-postgresql/src/test/java/org/apache/sqoop/integration/repository/postgresql/TestJobHandling.java

@@ -17,7 +17,6 @@
  */
 package org.apache.sqoop.integration.repository.postgresql;
 
-import org.apache.sqoop.common.Direction;
 import org.apache.sqoop.common.SqoopException;
 import org.apache.sqoop.common.test.db.TableName;
 import org.apache.sqoop.model.MConfig;
@@ -138,7 +137,7 @@ public void testFindJobs() throws Exception {
 
   @Test
   public void testFindJobsByConnector() throws Exception {
-    List<MJob> list = handler.findJobsForConnector(
+    List<MJob> list = handler.findJobsForConnectorUpgrade(
       handler.findConnector("A", provider.getConnection()).getPersistenceId(),
       provider.getConnection());
     assertEquals(2, list.size());
@@ -148,7 +147,8 @@ public void testFindJobsByConnector() throws Exception {
 
   @Test
   public void testFindJobsForNonExistingConnector() throws Exception {
-    List<MJob> list = handler.findJobsForConnector(11, provider.getConnection());
+    List<MJob> list = handler.findJobsForConnectorUpgrade(11, provider
+      .getConnection());
     assertEquals(0, list.size());
   }
 

repository/repository-postgresql/src/test/java/org/apache/sqoop/integration/repository/postgresql/TestLinkHandling.java

@@ -137,7 +137,7 @@ public void testFindLinksByConnector() throws Exception {
     List<MLink> list;
 
     // Load all two links on loaded repository
-    list = handler.findLinksForConnector("A", provider.getConnection());
+    list = handler.findLinksForConnectorUpgrade("A", provider.getConnection());
     assertEquals(1, list.size());
     assertEquals(LINK_A_NAME, list.get(0).getName());
 
@@ -147,13 +147,13 @@ public void testFindLinksByConnector() throws Exception {
     }
 
     // Load empty list on empty repository
-    list = handler.findLinksForConnector("A", provider.getConnection());
+    list = handler.findLinksForConnectorUpgrade("A", provider.getConnection());
     assertEquals(0, list.size());
   }
 
   @Test
   public void testFindLinksByNonExistingConnector() throws Exception {
-    List<MLink> list = handler.findLinksForConnector("NONEXISTCONNECTOR", provider.getConnection());
+    List<MLink> list = handler.findLinksForConnectorUpgrade("NONEXISTCONNECTOR", provider.getConnection());
     assertEquals(0, list.size());
   }