SQOOP-2588: Sqoop2: CommonRepositoryHandler#inUseLink only check if link is used in the from side of a job

(Dian Fu via Jarek Jarcec Cecho)

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

@@ -448,6 +448,7 @@ public boolean inUseLink(String linkName, Connection conn) {
 
     try (PreparedStatement stmt = conn.prepareStatement(crudQueries.getStmtSelectJobsForLinkCheck())) {
       stmt.setString(1, linkName);
+      stmt.setString(2, linkName);
       try (ResultSet rs = stmt.executeQuery()) {
 
         // Should be always valid in case of count(*) query

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

@@ -397,12 +397,18 @@ public class CommonRepositoryInsertUpdateDeleteSelectQuery {
 
   // DML: Check if there are jobs for given link
   private static final String STMT_SELECT_JOBS_FOR_LINK_CHECK =
-      "SELECT"
+      "SELECT SUM(CNT) FROM ("
-          + " count(*)"
+          + " SELECT count(*) as CNT"
           + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_JOB_NAME)
           + " INNER JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME)
           + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQB_FROM_LINK) + " = " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_ID)
-          + " WHERE " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_NAME) + " = ? ";
+          + " WHERE " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_NAME) + " = ? "
+          + " UNION ALL"
+          + " SELECT count(*) as CNT"
+          + " FROM " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_JOB_NAME)
+          + " INNER JOIN " + CommonRepoUtils.getTableName(SCHEMA_SQOOP, TABLE_SQ_LINK_NAME)
+          + " ON " + CommonRepoUtils.escapeColumnName(COLUMN_SQB_TO_LINK) + " = " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_ID)
+          + " WHERE " + CommonRepoUtils.escapeColumnName(COLUMN_SQ_LNK_NAME) + " = ? ) as JOB_COUNT";
 
   //DML: Select all jobs
   private static final String STMT_SELECT_JOB_ALL =

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

@@ -668,7 +668,7 @@ public void loadJobs(int version) throws Exception {
     case 7:
       for (String name : new String[] { "JA", "JB", "JC", "JD" }) {
         runQuery("INSERT INTO SQOOP.SQ_JOB(SQB_NAME, SQB_FROM_LINK, SQB_TO_LINK)" + " VALUES('"
-            + name + index + "', 1, 1)");
+            + name + index + "', 1, 2)");
       }
 
       // Odd IDs inputs have values

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

@@ -210,10 +210,12 @@ public void testInUseLink() throws Exception {
     loadLinksForLatestVersion();
 
     assertFalse(handler.inUseLink("CA", getDerbyDatabaseConnection()));
+    assertFalse(handler.inUseLink("CB", getDerbyDatabaseConnection()));
 
     loadJobsForLatestVersion();
 
     assertTrue(handler.inUseLink("CA", getDerbyDatabaseConnection()));
+    assertTrue(handler.inUseLink("CB", getDerbyDatabaseConnection()));
   }
 
   @Test

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

@@ -217,16 +217,20 @@ public void testInUseLink() throws Exception {
     // Create job and submission and make that job in use to make sure link is
     // in use.
     MLink linkA = handler.findLink(LINK_A_NAME, provider.getConnection());
+    MLink linkB = handler.findLink(LINK_B_NAME, provider.getConnection());
     MJob job = getJob("Job-A",
         handler.findConnector("A", provider.getConnection()),
-        handler.findConnector("B", provider.getConnection()), linkA,
+        handler.findConnector("B", provider.getConnection()),
-        handler.findLink(LINK_B_NAME, provider.getConnection()));
+        linkA,
+        linkB);
     handler.createJob(job, provider.getConnection());
     MSubmission submission = getSubmission(job, SubmissionStatus.RUNNING);
     handler.createSubmission(submission, provider.getConnection());
 
     assertTrue(handler.inUseLink(linkA.getName(),
         provider.getConnection()));
+    assertTrue(handler.inUseLink(linkB.getName(),
+        provider.getConnection()));
   }
 
   @Test

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

@@ -212,16 +212,18 @@ public void testInUseLink() throws Exception {
 
     // Create job and submission and make that job in use to make sure link is in use.
     MLink linkA = handler.findLink(LINK_A_NAME, provider.getConnection());
+    MLink linkB = handler.findLink(LINK_B_NAME, provider.getConnection());
     MJob job = getJob("Job-A",
         handler.findConnector("A", provider.getConnection()),
         handler.findConnector("B", provider.getConnection()),
         linkA,
-        handler.findLink(LINK_B_NAME, provider.getConnection()));
+        linkB);
     handler.createJob(job, provider.getConnection());
     MSubmission submission = getSubmission(job, SubmissionStatus.RUNNING);
     handler.createSubmission(submission, provider.getConnection());
 
     assertTrue(handler.inUseLink(linkA.getName(), provider.getConnection()));
+    assertTrue(handler.inUseLink(linkB.getName(), provider.getConnection()));
   }
 
   @Test